96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
"""知识库消费指标单元测试（mock 数据，不依赖数据库）。"""
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
from app.kb_metrics import kb_metrics, kb_metrics_overview
|
||
|
||
|
||
def _usage(agent_id, kb_id, chunk_id, segment_id, status="consumed"):
|
||
return SimpleNamespace(agent_id=agent_id, kb_id=kb_id, chunk_id=chunk_id,
|
||
segment_id=segment_id, status=status, file_name="f.md",
|
||
heading="h", excerpt="e")
|
||
|
||
|
||
def _segment(sid, references=None, status="approved"):
|
||
return SimpleNamespace(id=sid, status=status, references=references or [])
|
||
|
||
|
||
def _session(usage_rows, segments, kbs=()):
|
||
def exec(*_a, **_k):
|
||
# 简化：按调用顺序不可靠，直接用鸭子类型区分——这里用闭包计数不必要，
|
||
# kb_metrics_overview 内部按顺序 exec 三次：usage / kbs / segments，
|
||
# kb_metrics 内部两次：usage / segments。用可迭代队列处理。
|
||
raise AssertionError("请使用带计划的 mock")
|
||
|
||
return SimpleNamespace(exec=exec)
|
||
|
||
|
||
class _PlannedSession:
|
||
def __init__(self, queues):
|
||
self.queues = list(queues)
|
||
|
||
def exec(self, *_a, **_k):
|
||
return SimpleNamespace(all=lambda: self.queues.pop(0) if self.queues else [])
|
||
|
||
def get(self, model, row_id):
|
||
return None
|
||
|
||
|
||
def test_metrics_empty():
|
||
session = _PlannedSession([[], []])
|
||
report = kb_metrics(session)
|
||
assert report["hit_rate"] == 0.0
|
||
assert report["repeat_rate"] == 0.0
|
||
assert report["reference_coverage"] == 0.0
|
||
|
||
|
||
def test_hit_rate_and_repeat_rate():
|
||
usage = [
|
||
_usage(1, 1, "c1", 101), _usage(1, 1, "c2", 101),
|
||
_usage(2, 1, "c1", 102),
|
||
# 重复消费：agent1 再次用 c1（segment 103）→ pair (1, c1) 两条
|
||
_usage(1, 1, "c1", 103),
|
||
]
|
||
session = _PlannedSession([usage, []])
|
||
report = kb_metrics(session)
|
||
assert report["detail"]["hit_requests"] == 3 # (1,101)(2,102)(1,103)
|
||
assert report["detail"]["repeated_pairs"] == 1
|
||
assert report["repeat_rate"] == round(1 / 3, 4)
|
||
assert report["reference_coverage"] == 0.0 # 没有 segment 数据
|
||
|
||
|
||
def test_reference_coverage():
|
||
usage = [_usage(1, 1, "c1", 101)]
|
||
segments = [_segment(101, references=[{"file": "f"}]), _segment(102, references=[])]
|
||
session = _PlannedSession([usage, segments])
|
||
report = kb_metrics(session)
|
||
assert report["reference_coverage"] == 0.5
|
||
|
||
|
||
def test_miss_requests_lower_hit_rate():
|
||
usage = [_usage(1, 1, "c1", 101)]
|
||
# 段 102 是已完成的生产段但完全没有消费记录 → 记一次未命中
|
||
segments = [_segment(101, references=[{"x": 1}]), _segment(102)]
|
||
session = _PlannedSession([usage, segments])
|
||
report = kb_metrics(session)
|
||
assert report["detail"]["search_requests"] == 2
|
||
assert report["detail"]["miss_requests"] == 1
|
||
assert report["hit_rate"] == 0.5
|
||
|
||
|
||
def test_overview_by_kb_sorted():
|
||
usage = [
|
||
_usage(1, 1, "c1", 101), _usage(1, 1, "c2", 101),
|
||
_usage(1, 2, "c3", 102),
|
||
]
|
||
kbs = [
|
||
SimpleNamespace(id=1, name="主库", file_count=2, chunk_count=10),
|
||
SimpleNamespace(id=2, name="副库", file_count=1, chunk_count=10),
|
||
]
|
||
# kb_metrics_overview 先调 kb_metrics（usage, segments），再查 usage 和 kbs
|
||
session = _PlannedSession([usage, [], usage, kbs])
|
||
report = kb_metrics_overview(session)
|
||
assert report["by_kb"][0]["kb_id"] == 1 # 消费比 0.2 > 0.1
|
||
assert report["by_kb"][0]["consumed"] == 2
|
||
assert report["top_chunks"][0]["chunk_id"] == "c1" # c1 有 2 条记录（同 agent 两段）
|