82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
"""推理链轨迹存储 / 查询测试（T3：整体项目部分拆解·先行实现）。"""
|
|
import pytest
|
|
|
|
from router_system.router import build_router
|
|
from router_system.trace import TraceStore
|
|
|
|
|
|
def test_trace_store_put_get():
|
|
ts = TraceStore()
|
|
ts.put("abc", {"query": "q1", "route": ["a", "b"]})
|
|
t = ts.get("abc")
|
|
assert t["query"] == "q1"
|
|
assert ts.get("not-exist") is None
|
|
|
|
|
|
def test_trace_store_ring_eviction():
|
|
ts = TraceStore(max_entries=3)
|
|
for i in range(5):
|
|
ts.put(f"id{i}", {"i": i})
|
|
assert ts.size() == 3
|
|
assert ts.get("id0") is None # 最旧被淘汰
|
|
assert ts.get("id4") is not None
|
|
|
|
|
|
def test_trace_store_overwrite():
|
|
ts = TraceStore()
|
|
ts.put("a", {"v": 1})
|
|
ts.put("a", {"v": 2})
|
|
assert ts.get("a")["v"] == 2
|
|
assert ts.size() == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_records_trace():
|
|
r = build_router()
|
|
res = await r.route("求解方程 x^2 - 5x + 6 = 0")
|
|
assert res.request_id
|
|
trace = r.trace_store.get(res.request_id)
|
|
assert trace is not None
|
|
assert trace["domain"] == "math"
|
|
assert trace["domain_group"] == "tech"
|
|
assert "plan:multi" in " ".join(trace["route"])
|
|
assert trace["quality_score"] > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trace_cache_hit_roundtrip():
|
|
r = build_router()
|
|
q = "加班费怎么计算"
|
|
r1 = await r.route(q)
|
|
r2 = await r.route(q) # 缓存命中
|
|
assert r2.cache_hit is True
|
|
trace = r.trace_store.get(r2.request_id)
|
|
assert trace["cache_hit"] is True
|
|
assert trace["cache_level"] == "exact"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trace_subdomain_fields():
|
|
r = build_router()
|
|
res = await r.route("基金定投的收益率怎么计算")
|
|
trace = r.trace_store.get(res.request_id)
|
|
assert trace["subdomain"] == "investing"
|
|
assert trace["subdomain2"] == "investing"
|
|
assert trace["domain_group"] == "professional"
|
|
|
|
|
|
def test_gateway_trace_endpoint():
|
|
from fastapi.testclient import TestClient
|
|
from gateway.api import app
|
|
c = TestClient(app, raise_server_exceptions=False)
|
|
chat = c.post("/chat/legacy", json={"query": "请用 Python 实现快速排序的迭代版本，并分析其时间与空间复杂度"})
|
|
rid = chat.json().get("request_id")
|
|
assert rid
|
|
t = c.get(f"/traces/{rid}")
|
|
assert t.status_code == 200
|
|
body = t.json()
|
|
assert body["domain"] == "code"
|
|
assert "subdomain:algorithm" in " ".join(body["route"])
|
|
miss = c.get("/traces/不存在的id")
|
|
assert miss.status_code == 404
|