feat(awooop): 新增 truth-chain 查詢 API
All checks were successful
Code Review / ai-code-review (push) Successful in 10s
CD Pipeline / tests (push) Successful in 1m16s
CD Pipeline / build-and-deploy (push) Successful in 3m29s
CD Pipeline / post-deploy-checks (push) Successful in 1m19s

This commit is contained in:
Your Name
2026-05-12 22:55:36 +08:00
parent 56228dbb79
commit f7c84530d6
7 changed files with 777 additions and 0 deletions

View File

@@ -0,0 +1,48 @@
from __future__ import annotations
from src.services.awooop_truth_chain_service import _truth_status
def test_truth_status_marks_no_action_approval_as_manual_required() -> None:
status = _truth_status(
incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
approvals=[{"status": "APPROVED", "action": "未知操作 | NO_ACTION"}],
evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}],
automation_ops=[],
drift=None,
drift_repeat_count=0,
gateway_mcp_total=0,
legacy_mcp_total=8,
outbound_visible_total=0,
)
assert status["current_stage"] == "manual_required"
assert status["stage_status"] == "blocked"
assert status["needs_human"] is True
assert "approval_resolved_no_action_without_execution" in status["blockers"]
assert "all_evidence_sensors_failed" in status["blockers"]
assert "awooop_mcp_gateway_audit_empty" in status["blockers"]
def test_truth_status_marks_repeated_pending_drift_as_human_needed() -> None:
status = _truth_status(
incident=None,
approvals=[],
evidence_rows=[],
automation_ops=[],
drift={
"report_id": "7f858956",
"status": "pending",
"interpretation": {"confidence": 0.0},
},
drift_repeat_count=12,
gateway_mcp_total=0,
legacy_mcp_total=0,
outbound_visible_total=0,
)
assert status["current_stage"] == "dedup_or_repeat_updated"
assert status["stage_status"] == "pending"
assert status["needs_human"] is True
assert "drift_report_pending_without_resolution" in status["blockers"]
assert "drift_ai_confidence_zero" in status["blockers"]

View File

@@ -25,3 +25,13 @@ def test_recent_events_route_is_registered() -> None:
]
assert "/events/recent" in paths
def test_truth_chain_route_is_registered() -> None:
paths = [
route.path
for route in router.routes
if "GET" in getattr(route, "methods", set())
]
assert "/truth-chain/{source_id}" in paths