74 lines
3.2 KiB
Python
74 lines
3.2 KiB
Python
"""全局总监 Agent 单元测试。"""
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
from app.agents.supervisor import supervisor_check
|
|
from app.models import Agent, Campaign, Canvas, Segment
|
|
|
|
|
|
def _session_with_campaign(round_: int = 1, status: str = "approved"):
|
|
import tempfile, pathlib
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False})
|
|
SQLModel.metadata.create_all(engine)
|
|
session = Session(engine)
|
|
supervisor = Agent(id=99, name="总监", role="qc", status="active", rules="r")
|
|
campaign = Campaign(id=1, name="测试任务", objective="目标", success_criteria="标准",
|
|
canvas_snapshot={"supervisor_agent_id": 99, "supervisor_mode": "always"},
|
|
status="running")
|
|
artifact = pathlib.Path(tempfile.mkdtemp()) / "seg_10.md"
|
|
artifact.write_text("这是生成的正文内容", encoding="utf-8")
|
|
seg = Segment(id=10, campaign_id=1, title="开场", artifact_path=str(artifact),
|
|
round=round_, status=status)
|
|
session.add(supervisor); session.add(campaign); session.add(seg)
|
|
session.commit()
|
|
return session, campaign, seg
|
|
|
|
|
|
def test_supervisor_disabled_when_no_snapshot():
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False})
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as s:
|
|
campaign = Campaign(id=2, name="无画布", status="running", canvas_snapshot={})
|
|
seg = Segment(id=20, campaign_id=2, title="段", status="approved")
|
|
s.add(campaign); s.add(seg); s.commit()
|
|
assert supervisor_check(s, seg, campaign) == {"checked": False}
|
|
|
|
|
|
def test_supervisor_on_drift_skips_clean_first_round():
|
|
session, campaign, seg = _session_with_campaign(round_=1, status="approved")
|
|
# on_drift 模式下首段且无异常 → 不审（省点数）
|
|
campaign.canvas_snapshot = {"supervisor_agent_id": 99, "supervisor_mode": "on_drift"}
|
|
session.add(campaign); session.commit()
|
|
assert supervisor_check(session, seg, campaign) == {"checked": False}
|
|
session.close()
|
|
|
|
|
|
def test_supervisor_always_reviews_and_redo():
|
|
session, campaign, seg = _session_with_campaign()
|
|
with patch("app.llm.chat_json") as mock:
|
|
mock.return_value = {"status": "ok", "text": json.dumps({
|
|
"on_track": False, "reason": "跑到了别的话题",
|
|
"guidance": "回到开场主题"})}
|
|
verdict = supervisor_check(session, seg, campaign)
|
|
assert verdict["checked"] and not verdict["on_track"]
|
|
assert verdict["reason"] == "跑到了别的话题"
|
|
assert verdict["guidance"] == "回到开场主题"
|
|
session.close()
|
|
|
|
|
|
def test_supervisor_off_mode():
|
|
session, campaign, seg = _session_with_campaign()
|
|
campaign.canvas_snapshot = {"supervisor_agent_id": 99, "supervisor_mode": "off"}
|
|
session.add(campaign); session.commit()
|
|
assert supervisor_check(session, seg, campaign) == {"checked": False}
|
|
session.close()
|
|
|
|
|
|
def test_supervisor_llm_failure_not_blocking():
|
|
session, campaign, seg = _session_with_campaign()
|
|
with patch("app.llm.chat_json", side_effect=Exception("限流")):
|
|
assert supervisor_check(session, seg, campaign) == {"checked": False}
|
|
session.close()
|