57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
"""路由主流程单元测试。"""
|
|
import pytest
|
|
|
|
from router_system.router import build_router
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normal_flow_code(router):
|
|
r = await router.route("用 Python 写一个快速排序函数")
|
|
assert r.domain == "code"
|
|
assert r.response
|
|
assert r.model_used
|
|
assert r.latency_ms >= 0
|
|
assert "expert" in r.route[1] or "classify" in r.route[1]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_low_confidence_direct_fallback(router):
|
|
r = await router.route("今天天气怎么样")
|
|
assert r.upgraded is True
|
|
assert "direct_fallback" in r.route
|
|
assert r.model_used == router.fallback.model
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_second_round(router):
|
|
q = "用 Python 写一个快速排序函数"
|
|
r1 = await router.route(q)
|
|
assert r1.cache_hit is False
|
|
r2 = await router.route(q)
|
|
assert r2.cache_hit is True
|
|
assert r2.cache_level in ("exact", "semantic")
|
|
assert r2.response == r1.response
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_judge_route_present(router):
|
|
r = await router.route("高血压患者日常饮食需要注意什么")
|
|
assert any(step.startswith("judge:") for step in r.route)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_recorded(router):
|
|
await router.route("写一个 python 函数")
|
|
await router.route("写一个 python 函数")
|
|
s = router.stats.summary()
|
|
assert s["total_requests"] == 2
|
|
assert s["domain_distribution"]["code"] == 2
|
|
assert s["cache_hit_rate"] > 0
|
|
|
|
|
|
def test_health(router):
|
|
h = router.health()
|
|
assert h["status"] == "ok"
|
|
assert "code" in h["domains"]
|
|
assert "math" in h["domains"]
|
|
|