139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
"""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"):
|
|
if data["action"] == "get_login_info":
|
|
result = {"user_id": 88888, "nickname": "bot"}
|
|
else:
|
|
result = None
|
|
if data["action"].startswith("send_"):
|
|
self.replies.append(data["params"].get("message", ""))
|
|
await self.ws.send_str(json.dumps(
|
|
{"status": "ok", "retcode": 0, "data": result, "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 send_group_message(self, message, group_id: int = 999):
|
|
"""message 可为字符串或 CQ 段数组。"""
|
|
await self.ws.send_str(json.dumps({
|
|
"post_type": "message", "message_type": "group",
|
|
"group_id": group_id, "user_id": 10000, "message": message,
|
|
}))
|
|
|
|
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
|
|
|
|
|
|
async def _expect_no_reply(client, send, timeout: float = 3.0):
|
|
"""发送后确认机器人在超时内没有回复。"""
|
|
before = len(client.replies)
|
|
await send()
|
|
for _ in range(int(timeout * 10)):
|
|
if len(client.replies) > before:
|
|
raise AssertionError(f"不应回复但收到了: {client.replies[-1]}")
|
|
await asyncio.sleep(0.1)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_private_non_command_ignored(client):
|
|
# 私聊发普通消息/无文本消息不触发
|
|
await _expect_no_reply(client, lambda: client.send_message("今天天气怎么样"))
|
|
await _expect_no_reply(
|
|
client, lambda: client.send_message(
|
|
[{"type": "image", "data": {"file": "abc.jpg"}}]))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_requires_at_me(client):
|
|
# 群聊不 @ 不响应
|
|
await _expect_no_reply(client, lambda: client.send_group_message("/status"))
|
|
# @ 别人不响应
|
|
await _expect_no_reply(client, lambda: client.send_group_message(
|
|
[{"type": "at", "data": {"qq": "66666"}},
|
|
{"type": "text", "data": {"text": " /status"}}]))
|
|
# @ 本人才响应
|
|
await client.send_group_message(
|
|
[{"type": "at", "data": {"qq": "88888"}},
|
|
{"type": "text", "data": {"text": " /status"}}])
|
|
reply = await wait_reply(client)
|
|
assert "服务器状态" in reply
|