Files
xiaonuan/tests/test_router.py
T
snowy-yang 6981c6df4d 完善测试:pytest 套件(路由器/工具循环/WS全链路/投递/PG集成)
- tests/conftest.py: 内存版 FakeDB 复刻 db.py 全部签名,绕过 lifespan 的测试应用
- test_chat_flow.py: WebSocket 全链路(回声/路由信号落库/档案/慢信/连续天数)
- test_delivery.py: 事件/每日任务/关怀投递(在线与离线)、天气工具城市链、system prompt 注入
- test_router.py: 小模型路由清洗(去重/丢过去事件/非法作息/500优雅降级/无模型空判定)
- test_tool_loop.py: 原生工具调用循环(喂结果/超轮次强制收尾/坏参数)
- test_pure.py: 纯逻辑(连续天数/哈希槽位/时间间隔/模板/向量)
- test_db_pg.py: PG 集成测试,无 PG_DSN 自动跳过,配好即真实跑库
- 旧散装脚本移除;tools_live/ws_client/tui_smoke 保留为手动脚本
2026-09-23 20:20:37 +08:00

85 lines
3.4 KiB
Python

"""小模型路由的离线测试(mock 模型服务,不发真实请求)"""
import asyncio
import json
from datetime import datetime
import httpx
from app import connector
NOW = datetime(2026, 9, 21, 19, 30)
def mock_model(content: str):
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"choices": [{"message": {
"role": "assistant", "content": content}}]})
return handler
def enable_router(monkeypatch, content=None, status=200):
monkeypatch.setattr(connector, "SMALL_KEY", "k")
monkeypatch.setattr(connector, "SMALL_MODEL", "m")
if status == 200:
transport = httpx.MockTransport(mock_model(content or ""))
else:
transport = httpx.MockTransport(lambda r: httpx.Response(status))
monkeypatch.setattr(connector, "_TEST_TRANSPORT", transport)
def test_route_full_parse(monkeypatch):
raw = json.dumps({
"view": "memories",
"letter": {"days": 7, "body": "给未来的我:加油"},
"negative_mood": True,
"city_hint": "杭州",
"facts": ["用户叫小明", "用户叫小明", "用户喜欢打篮球"],
"events": [{"kind": "alarm", "fire_at": "2026-09-22T07:00:00",
"text": "起床", "repeat": "daily"},
{"kind": "reminder", "fire_at": "2020-01-01T00:00:00",
"text": "过去的应被丢弃"}],
"routines": [{"label": "到校", "time": "07:40", "kind": "inferred"},
{"label": "打游戏", "time": "22:00", "kind": "explicit"}],
"promises": [{"note": "面试", "due_at": None}],
"lore": [{"topic": "诞生", "body": "小暖是用户亲手造的"}],
})
enable_router(monkeypatch, "好的 " + raw + " (忽略我)")
sig = asyncio.run(connector.route(
"我叫小明,我在杭州,明早7点叫我起床,明天面试", NOW, ["用户叫小明"]))
assert sig["view"] == "memories"
assert sig["letter"] == (7, "给未来的我:加油")
assert sig["negative_mood"] is True
assert sig["city_hint"] == "杭州"
assert sig["facts"] == ["用户喜欢打篮球"] # 去重 + 已有的不要
assert len(sig["events"]) == 1
assert sig["events"][0]["kind"] == "alarm"
assert sig["events"][0]["repeat"] == "daily"
assert sig["events"][0]["fire_at"].day == 22
assert sig["routines"] == [{"label": "到校", "minute": 460,
"kind": "inferred"}]
assert sig["promises"][0]["note"] == "面试"
assert (sig["promises"][0]["due_at"].day,
sig["promises"][0]["due_at"].hour) == (22, 18) # null -> 次日18点
assert sig["lore"] == [{"topic": "诞生", "body": "小暖是用户亲手造的"}]
def test_route_garbage_returns_empty(monkeypatch):
enable_router(monkeypatch, "这不是 JSON")
sig = asyncio.run(connector.route("随便", NOW, []))
assert sig["view"] is None and sig["facts"] == [] and sig["letter"] is None
def test_route_server_error_returns_empty(monkeypatch):
enable_router(monkeypatch, status=500)
sig = asyncio.run(connector.route("随便", NOW, []))
assert sig["view"] is None
def test_route_without_model_is_noop(monkeypatch):
monkeypatch.setattr(connector, "SMALL_KEY", "")
monkeypatch.setattr(connector, "SMALL_MODEL", "")
sig = asyncio.run(connector.route("明早7点叫我起床", NOW, []))
assert sig["view"] is None and sig["events"] == []