60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
"""模拟 OneBot 实现端连接机器人，验证命令与告警。
|
|
|
|
用法:
|
|
python test_fake_onebot.py # 测试命令: status/top/help
|
|
python test_fake_onebot.py alert # 同时测试高负载告警(发送假的高CPU事件不可能,改为直接调用不适用;
|
|
# 告警建议在 Linux 上把 config.toml 阈值调低实测)
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
|
|
import aiohttp
|
|
|
|
WS_URL = "ws://127.0.0.1:8765/ws"
|
|
|
|
|
|
async def main():
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.ws_connect(WS_URL) as ws:
|
|
print("[fake] 已连接机器人")
|
|
|
|
async def send_event(text, message_type="private"):
|
|
ev = {
|
|
"post_type": "message",
|
|
"message_type": message_type,
|
|
"group_id": 123456789,
|
|
"user_id": 10000,
|
|
"message": text,
|
|
}
|
|
await ws.send_str(json.dumps(ev))
|
|
print(f"[fake] -> 事件: {text}")
|
|
|
|
async def reader():
|
|
async for msg in ws:
|
|
data = json.loads(msg.data)
|
|
if data.get("action"):
|
|
action = data["action"]
|
|
params = data.get("params", {})
|
|
content = params.get("message", "")
|
|
print(f"[fake] <- API {action}: \n{content}\n")
|
|
await ws.send_str(json.dumps(
|
|
{"status": "ok", "retcode": 0, "data": None, "echo": data.get("echo")}))
|
|
|
|
task = asyncio.create_task(reader())
|
|
|
|
for cmd in ["状态", "功耗", "top", "温度", "网络", "帮助"]:
|
|
await send_event(cmd)
|
|
await asyncio.sleep(0.5)
|
|
|
|
print("[fake] 等待全部回复(最长30秒)...")
|
|
await asyncio.sleep(30)
|
|
task.cancel()
|
|
print("[fake] 测试完成")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if sys.platform == "win32":
|
|
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
|
asyncio.run(main())
|