207 lines
7.4 KiB
Python
207 lines
7.4 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.services.signal.signal_planner.mixin_loaders import SignalPlannerLoadersMixin
|
|
from app.services.signal.signal_planner.service import SignalPlannerService
|
|
|
|
|
|
class _FakeScalarResult:
|
|
def __init__(self, row):
|
|
self._row = row
|
|
|
|
def scalar_one_or_none(self):
|
|
return self._row
|
|
|
|
|
|
class _FakeDB:
|
|
def __init__(self, rows):
|
|
self._rows = list(rows)
|
|
self.statements = []
|
|
|
|
async def execute(self, statement):
|
|
self.statements.append(statement)
|
|
if self._rows:
|
|
return _FakeScalarResult(self._rows.pop(0))
|
|
return _FakeScalarResult(None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_planner_latest_charter_payload_flattens_modules_and_orders_latest() -> None:
|
|
charter = SimpleNamespace(
|
|
id=77,
|
|
title="支付优化宪章",
|
|
version=3,
|
|
status="aligned",
|
|
content={
|
|
"modules": {
|
|
"module_1": {
|
|
"problem_statement": "减少支付失败",
|
|
"goal_statement": "提升支付成功率",
|
|
"kpis": "成功率 >= 99%",
|
|
},
|
|
"module_2": {
|
|
"in_scope": "支付链路优化",
|
|
"out_of_scope": "结算清分重构",
|
|
},
|
|
"module_4": {
|
|
"milestones": "需求冻结\n开发上线",
|
|
},
|
|
"module_5": {
|
|
"risk_register": "第三方通道不稳定",
|
|
},
|
|
"module_6": {
|
|
"governance_rule": "对外变更需复核",
|
|
},
|
|
}
|
|
},
|
|
)
|
|
db = _FakeDB([charter])
|
|
|
|
payload = await SignalPlannerLoadersMixin._load_latest_charter_payload(
|
|
db,
|
|
tenant_id=1,
|
|
project_id=42,
|
|
)
|
|
|
|
assert payload["title"] == "支付优化宪章"
|
|
assert payload["summary"] == "减少支付失败"
|
|
assert payload["objectives"] == ["提升支付成功率"]
|
|
assert payload["in_scope"] == ["支付链路优化"]
|
|
assert payload["out_of_scope"] == ["结算清分重构"]
|
|
assert payload["milestones"] == ["需求冻结", "开发上线"]
|
|
assert "对外变更需复核" in payload["constraints"][0]
|
|
|
|
sql_text = str(db.statements[0]).lower()
|
|
assert "order by" in sql_text
|
|
assert "updated_at" in sql_text
|
|
assert "id desc" in sql_text
|
|
|
|
|
|
def test_signal_planner_prompt_embeds_charter_context() -> None:
|
|
signal = SimpleNamespace(
|
|
id=12,
|
|
title="继续优化支付失败重试",
|
|
summary="提升支付成功率",
|
|
content="需要优化支付失败重试策略",
|
|
matched_project_id=42,
|
|
strategic_analysis={
|
|
"ai_analysis": {
|
|
"summary": "支付链路存在失败重试问题",
|
|
"requirement_project_scope": "支付优化",
|
|
}
|
|
},
|
|
)
|
|
prompt = SignalPlannerService._build_prompt(
|
|
signal=signal,
|
|
project_title="支付优化",
|
|
charter_json={
|
|
"id": 77,
|
|
"title": "支付优化宪章",
|
|
"objectives": ["提升支付成功率"],
|
|
"scope": "支付链路优化",
|
|
"success_criteria": ["成功率 >= 99%"],
|
|
"constraints": ["对外变更需复核"],
|
|
"milestones": ["需求冻结", "开发上线"],
|
|
"risks": ["第三方通道不稳定"],
|
|
},
|
|
recent_signals=[{"id": 1, "title": "支付失败告警"}],
|
|
members=[{"user_id": 9, "name": "Alice", "workload_ratio": 42.5}],
|
|
)
|
|
|
|
assert "charter_json" in prompt
|
|
assert "支付优化宪章" in prompt
|
|
assert "success_criteria" in prompt
|
|
assert "对外变更需复核" in prompt
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_planner_analyze_degrades_without_charter(monkeypatch) -> None:
|
|
signal = SimpleNamespace(
|
|
id=12,
|
|
title="继续优化支付失败重试",
|
|
summary="提升支付成功率",
|
|
content="需要优化支付失败重试策略",
|
|
matched_project_id=42,
|
|
strategic_analysis={
|
|
"ai_analysis": {
|
|
"summary": "支付链路存在失败重试问题",
|
|
"requirement_project_scope": "支付优化",
|
|
}
|
|
},
|
|
)
|
|
captured: dict[str, object] = {}
|
|
|
|
async def _fake_load_signal(db, *, tenant_id: int, signal_id: int):
|
|
_ = (db, tenant_id, signal_id)
|
|
return signal
|
|
|
|
async def _fake_load_project_title(db, *, tenant_id: int, project_id):
|
|
_ = (db, tenant_id, project_id)
|
|
return "支付优化"
|
|
|
|
async def _fake_load_charter_payload(db, *, tenant_id: int, project_id):
|
|
_ = (db, tenant_id, project_id)
|
|
return {}
|
|
|
|
async def _fake_load_recent_signals(db, *, tenant_id: int, project_id, exclude_signal_id: int):
|
|
_ = (db, tenant_id, project_id, exclude_signal_id)
|
|
return []
|
|
|
|
async def _fake_load_members(db, *, tenant_id: int, project_id):
|
|
_ = (db, tenant_id, project_id)
|
|
return [{"user_id": 9, "name": "Alice", "workload_ratio": 42.5}]
|
|
|
|
async def _fake_generate_structured_object(prompt: str, **kwargs):
|
|
captured["prompt"] = prompt
|
|
captured["kwargs"] = kwargs
|
|
return SimpleNamespace(
|
|
payload={
|
|
"summary": "聚焦支付失败重试",
|
|
"priority": "high",
|
|
"assignee_user_id": 9,
|
|
"assignee_reason": "技能匹配且负载可接受",
|
|
"execution_plan": ["梳理失败场景", "拆解重试策略"],
|
|
"risks": ["第三方通道波动"],
|
|
"confidence": 0.91,
|
|
}
|
|
)
|
|
|
|
async def _fake_get_effective_routing(db, *, tenant_id: int, function_name: str):
|
|
_ = (db, tenant_id, function_name)
|
|
return {"source": "override", "provider": "glm", "model": "qwen-plus"}
|
|
|
|
async def _fake_merge_and_save(db, *, signal_id: int, planner_result: dict, tenant_id: int):
|
|
_ = (db, signal_id, tenant_id)
|
|
captured["planner_result"] = planner_result
|
|
return signal
|
|
|
|
class _FakeDB:
|
|
async def commit(self):
|
|
return None
|
|
|
|
async def rollback(self):
|
|
return None
|
|
|
|
monkeypatch.setattr(SignalPlannerService, "_load_signal", _fake_load_signal)
|
|
monkeypatch.setattr(SignalPlannerService, "_load_project_title", _fake_load_project_title)
|
|
monkeypatch.setattr(SignalPlannerService, "_load_latest_charter_payload", _fake_load_charter_payload)
|
|
monkeypatch.setattr(SignalPlannerService, "_load_recent_signal_summaries", _fake_load_recent_signals)
|
|
monkeypatch.setattr(SignalPlannerService, "_load_member_profiles", _fake_load_members)
|
|
monkeypatch.setattr("app.services.signal.signal_planner.mixin_analyze.llm_service.generate_structured_object", _fake_generate_structured_object)
|
|
monkeypatch.setattr("app.services.signal.signal_planner.mixin_analyze.ModelRoutingService.get_effective_routing", _fake_get_effective_routing)
|
|
monkeypatch.setattr("app.services.signal.signal_planner.mixin_analyze.SignalStrategicAnalysisService.merge_and_save", _fake_merge_and_save)
|
|
|
|
await SignalPlannerService._analyze_with_session(
|
|
db=_FakeDB(),
|
|
signal_id=12,
|
|
tenant_id=1,
|
|
source_message_id="msg-1",
|
|
source_chat_id="chat-1",
|
|
dedupe_seed="seed-1",
|
|
)
|
|
|
|
assert "charter_json" in str(captured["prompt"])
|
|
assert "{}" in str(captured["prompt"])
|
|
assert captured["planner_result"]["summary"] == "聚焦支付失败重试"
|