hhhge/LinuxAwesomeBot
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: replace ad-hoc fake client script with pytest suite

Browse Source
This commit is contained in:
snowy-yang committed 2026-09-12 19:30:46 +08:00
1 parent 05782a2fc2
commit 82ec179d21
6 files changed
+189 -67
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+1
View File
@@ -9,3 +9,4 @@ wheels/
# Virtual environments
.venv
config.toml
test_fake_onebot.py
pyproject.toml
+6 -7
View File
@@ -1,7 +1,7 @@
[project]
name = "linuxawesomebot"
version = "0.1.0"
description = "Add your description here"
description = "OneBot 11 Linux monitoring bot: power/CPU/memory/temps with high-load alerts"
readme = "README.md"
authors = [
{ name = "snowy-yang", email = "102860110+snowy-yang@users.noreply.github.com" }
@@ -12,9 +12,8 @@ dependencies = [
"psutil>=7.2.2",
]
[project.scripts]
linuxawesomebot = "linuxawesomebot:main"
[build-system]
requires = ["uv_build>=0.12.12,<0.13.0"]
build-backend = "uv_build"
[dependency-groups]
dev = [
"pytest>=9.1.1",
"pytest-asyncio>=1.4.0",
]
pytest.ini
+3
View File
@@ -0,0 +1,3 @@
[pytest]
asyncio_mode = strict
asyncio_default_fixture_loop_scope = function
test_fake_onebot.py
-59
View File
@@ -1,59 +0,0 @@
"""模拟 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())
tests/test_bot.py
+92
View File
@@ -0,0 +1,92 @@
"""pytest 异步测试：启动 OneBotServer，用内置的模拟 OneBot 客户端验证命令回复。"""
import asyncio
import json
import pytest
import pytest_asyncio
from aiohttp import ClientSession
import bot as bot_mod
HOST = "127.0.0.1"
PORT = 18765
class FakeOneBotClient:
"""模拟 OneBot 实现端：连入反向 WS，回执 API 调用并收集发送的消息。"""
def __init__(self, session: ClientSession):
self.session = session
self.ws = None
self.replies: list[str] = []
async def connect(self):
self.ws = await self.session.ws_connect(f"ws://{HOST}:{PORT}/ws")
self._reader = asyncio.create_task(self._read())
async def _read(self):
async for msg in self.ws:
data = json.loads(msg.data)
if data.get("action"): # API 请求：记录消息内容并回执
self.replies.append(data["params"].get("message", ""))
await self.ws.send_str(json.dumps(
{"status": "ok", "retcode": 0, "data": None, "echo": data.get("echo")}))
async def send_message(self, text: str):
await self.ws.send_str(json.dumps({
"post_type": "message", "message_type": "private",
"user_id": 10000, "message": text,
}))
async def close(self):
self._reader.cancel()
await self.ws.close()
@pytest_asyncio.fixture
async def client():
bot = bot_mod.Bot({
"server": {"host": HOST, "port": PORT, "access_token": ""},
"alert": {"groups": [], "friends": []},
})
await bot.ob.start(bot.on_event)
async with ClientSession() as session:
fake = FakeOneBotClient(session)
await fake.connect()
yield fake
await fake.close()
await bot.ob.stop()
async def wait_reply(fake: FakeOneBotClient, timeout: float = 30) -> str:
"""等待并返回下一条机器人发出的消息。"""
for _ in range(int(timeout * 10)):
if fake.replies:
return fake.replies.pop(0)
await asyncio.sleep(0.1)
raise TimeoutError("机器人未在超时时间内回复")
@pytest.mark.asyncio
@pytest.mark.parametrize("cmd,expect", [
("状态", "服务器状态"),
("功耗", "功耗"),
("top", "TOP10"),
("温度", "温度"),
("网络", "网络速率"),
("帮助", "可用命令"),
("/help", "可用命令"),
("!状态", "服务器状态"),
])
async def test_commands(client, cmd, expect):
await client.send_message(cmd)
reply = await wait_reply(client)
assert expect in reply
@pytest.mark.asyncio
async def test_status_cpu_display(client):
await client.send_message("状态")
reply = await wait_reply(client)
# Linux 风格总占用 + 核心
assert "核心" in reply and "/" in reply
uv.lock
Generated
+87 -1
View File
@@ -133,6 +133,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" },
]
[[package]]
name = "colorama"
version = "0.4.6"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
]
[[package]]
name = "frozenlist"
version = "1.8.0"
@@ -231,21 +240,42 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/57/b0/0e52c878c53f245edd3a11020f20979b3f490f245af532c7cae3027754b5/idna-3.19-py3-none-any.whl", hash = "sha256:815e7be7a7806d54abb586dc943addc79e8b2ee16915059658cbeff4b1b43bf4", size = 68550, upload-time = "2026-08-18T05:14:22.343Z" },
]
[[package]]
name = "iniconfig"
version = "2.3.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
]
[[package]]
name = "linuxawesomebot"
version = "0.1.0"
source = { editable = "." }
source = { virtual = "." }
dependencies = [
{ name = "aiohttp" },
{ name = "psutil" },
]
[package.dev-dependencies]
dev = [
{ name = "pytest" },
{ name = "pytest-asyncio" },
]
[package.metadata]
requires-dist = [
{ name = "aiohttp", specifier = ">=3.14.3" },
{ name = "psutil", specifier = ">=7.2.2" },
]
[package.metadata.requires-dev]
dev = [
{ name = "pytest", specifier = ">=9.1.1" },
{ name = "pytest-asyncio", specifier = ">=1.4.0" },
]
[[package]]
name = "multidict"
version = "6.8.0"
@@ -384,6 +414,24 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b1/ee/be4e1a4b7a2b27f4fb6936510d4bebcb41b0562c946930ad26916e069cf9/multidict-6.8.0-py3-none-any.whl", hash = "sha256:75daa15ca16d6285eb2e104b2f05ee6f8d9836c68da3ce5c85f615a0450eed0e", size = 16297, upload-time = "2026-09-09T13:57:56.106Z" },
]
[[package]]
name = "packaging"
version = "26.3"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/7d/fa/3944b40b07da9ce895c0e6303a5ab7d53da063554f534556b134a54d6093/packaging-26.3.tar.gz", hash = "sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79", size = 313412, upload-time = "2026-08-04T18:15:28.737Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956, upload-time = "2026-08-04T18:15:27.159Z" },
]
[[package]]
name = "pluggy"
version = "1.6.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
]
[[package]]
name = "propcache"
version = "0.5.2"
@@ -506,6 +554,44 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/8c/c7/7bb2e321574b10df20cbde462a94e2b71d05f9bbda251ef27d104668306a/psutil-7.2.2-cp37-abi3-win_arm64.whl", hash = "sha256:8c233660f575a5a89e6d4cb65d9f938126312bca76d8fe087b947b3a1aaac9ee", size = 134617, upload-time = "2026-01-28T18:15:36.514Z" },
]
[[package]]
name = "pygments"
version = "2.21.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329, upload-time = "2026-08-17T08:02:48.824Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" },
]
[[package]]
name = "pytest"
version = "9.1.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
{ name = "iniconfig" },
{ name = "packaging" },
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" },
]
[[package]]
name = "pytest-asyncio"
version = "1.4.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
{ name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/43/7c/d36d04db312ecf4298932ef77e6e4a9e8ad017906e24e34f0b0c361a2473/pytest_asyncio-1.4.0.tar.gz", hash = "sha256:c6c0d2259945122819f171a32ecea2c349ead889ee28176caaf492143424be42", size = 58514, upload-time = "2026-05-26T09:56:04.083Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" },
]
[[package]]
name = "typing-extensions"
version = "4.16.0"