LinuxAwesomeBot: OneBot 11 Linux monitoring bot (power/CPU/temp/alerts)

test_fake_onebot.py

@@ -0,0 +1,59 @@
+"""模拟 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())