feat(reports): 新增報表資料源健康 read model

apps/api/tests/test_ai_agent_report_source_health_api.py

@@ -0,0 +1,85 @@
+from fastapi.testclient import TestClient
+
+from src.main import app
+
+
+_PUBLIC_FORBIDDEN_TERMS = [
+    "工作視窗",
+    "對話內容",
+    "批准！繼續",
+    "In app browser",
+    "My request for Codex",
+    "browser_context",
+    "codex_user_message",
+    "prompt_text",
+    "raw prompt",
+    "raw_payload",
+    "private reasoning",
+    "chain_of_thought",
+    "authorization header",
+    "secret value",
+    "raw Telegram payload",
+]
+
+
+def _collect_strings(value):
+    if isinstance(value, str):
+        return [value]
+    if isinstance(value, list):
+        strings = []
+        for item in value:
+            strings.extend(_collect_strings(item))
+        return strings
+    if isinstance(value, dict):
+        strings = []
+        for item in value.values():
+            strings.extend(_collect_strings(item))
+        return strings
+    return []
+
+
+def test_get_ai_agent_report_source_health_api():
+    client = TestClient(app)
+    response = client.get("/api/v1/agents/agent-report-source-health")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["schema_version"] == "ai_agent_report_source_health_v1"
+    assert data["program_status"]["current_task_id"] == "P2-109"
+    assert data["program_status"]["next_task_id"] == "P2-110"
+    assert data["program_status"]["read_only_mode"] is True
+    assert data["program_status"]["overall_completion_percent"] == 100
+    assert data["rollups"]["source_count"] == 5
+    assert data["rollups"]["no_send_preview_count"] == 3
+    assert data["rollups"]["live_send_allowed_count"] == 0
+    assert data["rollups"]["runtime_gate_count"] == 0
+    assert data["activation_boundaries"]["telegram_send_enabled"] is False
+    assert data["activation_boundaries"]["gateway_queue_write_enabled"] is False
+    assert data["activation_boundaries"]["scheduler_change_enabled"] is False
+    assert data["activation_boundaries"]["ai_runtime_execution_enabled"] is False
+    assert data["activation_boundaries"]["production_write_enabled"] is False
+
+    source_ids = {source["source_id"] for source in data["source_health"]}
+    assert source_ids == {
+        "incident_summary",
+        "resolution_stats",
+        "ai_performance",
+        "disposition_stats",
+        "report_status_board",
+    }
+    asset_labels = {asset["label"] for asset in data["automation_assets"]}
+    assert asset_labels == {"KM", "PlayBook", "腳本", "排程", "Verifier"}
+    for preview in data["no_send_previews"]:
+        assert preview["delivery_state"] == "no_send_preview"
+        assert preview["live_send_allowed"] is False
+        assert preview["gateway_queue_write_allowed"] is False
+
+
+def test_get_ai_agent_report_source_health_api_redacts_public_terms():
+    client = TestClient(app)
+    response = client.get("/api/v1/agents/agent-report-source-health")
+
+    assert response.status_code == 200
+    all_text = "\n".join(_collect_strings(response.json()))
+    for term in _PUBLIC_FORBIDDEN_TERMS:
+        assert term not in all_text