feat(v3): Web 应用化基线(异步任务/SSE/llama-server 管理/Vue SPA 四页 + 设置页整页滚动修复)
This commit is contained in:
+56
-56
@@ -1,56 +1,56 @@
|
||||
"""路由主流程单元测试。"""
|
||||
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"]␍
|
||||
"""路由主流程单元测试。"""
|
||||
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 any("classify:" in s for s in r.route)
|
||||
|
||||
|
||||
@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"]
|
||||
|
||||
Reference in New Issue
Block a user