fix(api): restore CD router and alert tests
This commit is contained in:
@@ -3082,11 +3082,13 @@ class TelegramMessage:
|
||||
text = f"{self.root_cause} {self.suggested_action}".lower()
|
||||
if is_no_action_approval_action(self.suggested_action):
|
||||
if "draft_ready" in text or "owner_review_required" in text:
|
||||
return "repair_candidate_draft_ready_controlled_apply"
|
||||
return "repair_candidate_draft_ready_owner_review"
|
||||
if "repair_candidate_missing" in text:
|
||||
return "repair_candidate_missing_ai_generation"
|
||||
return "repair_candidate_missing_manual_handoff"
|
||||
return "ai_controlled_action_pending"
|
||||
if "超時" in text or "timeout" in text:
|
||||
if "人工" in text or self.suggested_action in {"待分析", "", "NO_ACTION"}:
|
||||
return "llm_timeout_manual_gate"
|
||||
return "llm_timeout_ai_route_retry"
|
||||
if self.confidence > 0 and self.suggested_action and self.suggested_action != "待分析":
|
||||
return "ai_proposal_ready"
|
||||
@@ -3139,10 +3141,10 @@ class TelegramMessage:
|
||||
return "🟠 AI 補救試跑證據查詢失敗,已排入 evidence connector 修復"
|
||||
if verdict == "approval_required":
|
||||
return "🟡 已進受控自動執行政策判定"
|
||||
if mode == "repair_candidate_draft_ready_controlled_apply":
|
||||
return "🟡 修復候選草案已產生,排入 controlled apply"
|
||||
if mode == "repair_candidate_missing_ai_generation":
|
||||
return "🟠 缺少可執行修復候選,AI 正在產生 PlayBook / verifier"
|
||||
if mode == "repair_candidate_draft_ready_owner_review":
|
||||
return "🟡 修復候選草案已產生,等待 owner review"
|
||||
if mode == "repair_candidate_missing_manual_handoff":
|
||||
return "🟠 缺少可執行修復候選,已產生人工處置包"
|
||||
if mode == "ai_controlled_action_pending":
|
||||
return "🟠 已排入 AI 受控處理"
|
||||
if verdict.startswith("manual_required"):
|
||||
@@ -3152,6 +3154,8 @@ class TelegramMessage:
|
||||
return "🔎 AI 已完成只讀診斷,排入受控修復候選"
|
||||
if state == "diagnosis_failed_manual_required":
|
||||
return "🔴 AI 診斷工具失敗,已排入 tool/connector 修復"
|
||||
if mode == "llm_timeout_manual_gate":
|
||||
return "🔴 AI 分析超時,已進人工審核安全閘門"
|
||||
if mode == "llm_timeout_ai_route_retry":
|
||||
return "🔴 AI 分析超時,已排入 AI Router fallback 重試"
|
||||
if action in {"NO_ACTION", "待分析", ""} or "invalid_target" in text:
|
||||
|
||||
@@ -1,14 +1,30 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from src.api.v1.platform import router
|
||||
|
||||
|
||||
def _get_paths_for_method(api_router: APIRouter, method: str) -> list[str]:
|
||||
paths: list[str] = []
|
||||
|
||||
for route in api_router.routes:
|
||||
nested_router = getattr(route, "original_router", None)
|
||||
if isinstance(nested_router, APIRouter):
|
||||
paths.extend(_get_paths_for_method(nested_router, method))
|
||||
continue
|
||||
|
||||
methods: Iterable[str] = getattr(route, "methods", set())
|
||||
if method in methods:
|
||||
paths.append(getattr(route, "path"))
|
||||
|
||||
return paths
|
||||
|
||||
|
||||
def test_runs_list_route_is_registered_before_dynamic_run_id() -> None:
|
||||
paths = [
|
||||
route.path
|
||||
for route in router.routes
|
||||
if "GET" in getattr(route, "methods", set())
|
||||
]
|
||||
paths = _get_paths_for_method(router, "GET")
|
||||
|
||||
assert "/runs/list" in paths
|
||||
assert "/runs/{run_id}/detail" in paths
|
||||
@@ -18,31 +34,19 @@ def test_runs_list_route_is_registered_before_dynamic_run_id() -> None:
|
||||
|
||||
|
||||
def test_recent_events_route_is_registered() -> None:
|
||||
paths = [
|
||||
route.path
|
||||
for route in router.routes
|
||||
if "GET" in getattr(route, "methods", set())
|
||||
]
|
||||
paths = _get_paths_for_method(router, "GET")
|
||||
|
||||
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())
|
||||
]
|
||||
paths = _get_paths_for_method(router, "GET")
|
||||
|
||||
assert "/truth-chain/{source_id}" in paths
|
||||
|
||||
|
||||
def test_truth_chain_quality_summary_route_is_registered_before_dynamic_source_id() -> None:
|
||||
paths = [
|
||||
route.path
|
||||
for route in router.routes
|
||||
if "GET" in getattr(route, "methods", set())
|
||||
]
|
||||
paths = _get_paths_for_method(router, "GET")
|
||||
|
||||
assert "/truth-chain/quality/summary" in paths
|
||||
assert "/truth-chain/{source_id}" in paths
|
||||
|
||||
Reference in New Issue
Block a user