115 lines
4.5 KiB
Python
115 lines
4.5 KiB
Python
"""准备度检查的单元测试（不依赖网络与真实 Provider）。"""
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
from app.readiness import readiness_report
|
||
|
||
|
||
def _campaign(cid=1, owner=7, org=1):
|
||
return SimpleNamespace(id=cid, owner_user_id=owner, organization_id=org)
|
||
|
||
|
||
def _segment(agent_id=11):
|
||
return SimpleNamespace(agent_id=agent_id)
|
||
|
||
|
||
def _agent(id=11, name="开场段 Agent", role="production", status="active", kb_ids=(1,)):
|
||
return SimpleNamespace(id=id, name=name, role=role, status=status, kb_ids=list(kb_ids))
|
||
|
||
|
||
class _Query:
|
||
def __init__(self, rows):
|
||
self.rows = rows
|
||
|
||
def all(self):
|
||
return self.rows
|
||
|
||
|
||
def _session_with(segments, agents_by_id, qc_rows, composer_rows):
|
||
"""构造最小 session mock：get 按 Agent 精确命中，exec 按调用顺序返回。"""
|
||
plan = [_Query(segments), _Query(qc_rows), _Query(composer_rows)]
|
||
# kb 查询按需追加
|
||
state = {"plan_index": 0}
|
||
|
||
def exec(*_a, **_k):
|
||
query = plan[min(state["plan_index"], len(plan) - 1)]
|
||
state["plan_index"] += 1
|
||
return query
|
||
|
||
def get(model, row_id):
|
||
if model.__name__ == "Agent":
|
||
return agents_by_id.get(row_id)
|
||
if model.__name__ == "KnowledgeBase":
|
||
return SimpleNamespace(id=row_id, name="测试库", index_status="ready")
|
||
return None
|
||
|
||
return SimpleNamespace(exec=exec, get=get)
|
||
|
||
|
||
POINTS_OK = {"points": 999999, "token_equivalent": 99999, "points_per_token": 10}
|
||
|
||
|
||
def test_no_segments_blocks():
|
||
session = _session_with([], {}, [], [])
|
||
report = readiness_report(session, _campaign())
|
||
assert report["ready"] is False
|
||
assert any("拆解" in b for b in report["blockers"])
|
||
|
||
|
||
def test_disabled_agent_blocks():
|
||
agent = _agent(status="disabled")
|
||
session = _session_with([_segment()], {11: agent}, [_agent(21, "质检", "qc")], [_agent(22, "组合", "composer")])
|
||
with patch("app.readiness.point_state", return_value=POINTS_OK):
|
||
report = readiness_report(session, _campaign())
|
||
assert any("停用" in b for b in report["blockers"])
|
||
|
||
|
||
def test_missing_kb_blocks():
|
||
agent = _agent(kb_ids=())
|
||
session = _session_with([_segment()], {11: agent}, [_agent(21, "质检", "qc")], [_agent(22, "组合", "composer")])
|
||
with patch("app.readiness.point_state", return_value=POINTS_OK), \
|
||
patch("app.readiness._provider_ready", return_value=(True, "")):
|
||
report = readiness_report(session, _campaign())
|
||
assert any("未挂接知识库" in b for b in report["blockers"])
|
||
|
||
|
||
def test_low_points_warns_not_blocks():
|
||
agent = _agent()
|
||
session = _session_with([_segment()], {11: agent}, [_agent(21, "质检", "qc")], [_agent(22, "组合", "composer")])
|
||
with patch("app.readiness.point_state", return_value={"points": 5000, "token_equivalent": 500, "points_per_token": 10}), \
|
||
patch("app.readiness._provider_ready", return_value=(True, "")):
|
||
report = readiness_report(session, _campaign())
|
||
assert report["ready"] is True
|
||
assert any("点数偏低" in w for w in report["warnings"])
|
||
|
||
|
||
def test_zero_points_blocks():
|
||
agent = _agent()
|
||
session = _session_with([_segment()], {11: agent}, [_agent(21, "质检", "qc")], [_agent(22, "组合", "composer")])
|
||
with patch("app.readiness.point_state", return_value={"points": 0, "token_equivalent": 0, "points_per_token": 10}), \
|
||
patch("app.readiness._provider_ready", return_value=(True, "")):
|
||
report = readiness_report(session, _campaign())
|
||
assert report["ready"] is False
|
||
assert any("点数余额不足" in b for b in report["blockers"])
|
||
|
||
|
||
def test_missing_qc_composer_blocks():
|
||
agent = _agent()
|
||
session = _session_with([_segment()], {11: agent}, [], [])
|
||
with patch("app.readiness.point_state", return_value=POINTS_OK), \
|
||
patch("app.readiness._provider_ready", return_value=(True, "")):
|
||
report = readiness_report(session, _campaign())
|
||
assert any("质检 Agent" in b for b in report["blockers"])
|
||
assert any("组合 Agent" in b for b in report["blockers"])
|
||
|
||
|
||
def test_disabled_qc_blocks():
|
||
agent = _agent()
|
||
session = _session_with([_segment()], {11: agent},
|
||
[_agent(21, "质检", "qc", status="disabled")],
|
||
[_agent(22, "组合", "composer")])
|
||
with patch("app.readiness.point_state", return_value=POINTS_OK), \
|
||
patch("app.readiness._provider_ready", return_value=(True, "")):
|
||
report = readiness_report(session, _campaign())
|
||
assert any("质检 Agent" in b for b in report["blockers"])
|