Files
awoooi/apps/api/tests/test_telegram_ai_automation_block.py
Your Name 06b40f316c
Some checks are pending
Ansible / Reboot Recovery Contract / validate (push) Waiting to run
CD Pipeline / build-and-deploy (push) Blocked by required conditions
CD Pipeline / post-deploy-checks (push) Blocked by required conditions
Code Review / ai-code-review (push) Successful in 23s
CD Pipeline / tests (push) Successful in 1m52s
fix(api): route repair candidates to controlled automation queue
2026-06-27 12:59:24 +08:00

254 lines
9.8 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from src.services.telegram_gateway import TelegramMessage
def test_action_required_card_exposes_ai_automation_on_fallback() -> None:
message = TelegramMessage(
status_emoji="🚨",
risk_level="CRITICAL",
resource_name="node-exporter-110",
root_cause="AI 分析超時90s降級至人工審核",
suggested_action="待分析",
estimated_downtime="5-15 min",
approval_id="test-approval-id",
incident_id="INC-20260429-TEST01",
primary_responsibility="INFRA",
confidence=0.0,
)
body = message.format()
assert "AI 自動化鏈路" in body
assert "rule_fallback" in body
assert "llm_timeout_ai_route_retry" in body
assert "OpenClaw" in body
assert "NemoTron" in body
assert "Hermes" in body
assert "ElephantAlpha" in body
assert "流程進度" in body
assert "執行:<code>no_action_or_observe</code>" in body
assert "等待人工批准" not in body
def test_repair_candidate_missing_card_exposes_ai_repair_candidate_package() -> None:
message = TelegramMessage(
status_emoji="",
risk_level="LOW",
resource_name="node-exporter-188",
root_cause="AI 選擇不執行修復,需人工判斷是否接手",
suggested_action="NO_ACTION - REPAIR_CANDIDATE_MISSING: LLM 分析失敗,尚未產生安全可執行修復指令",
estimated_downtime="unknown",
approval_id="test-approval-id",
incident_id="INC-20260611-34BBF5",
primary_responsibility="INFRA",
confidence=0.0,
alert_category="host_resource",
repair_candidate_blocker_summary="只命中通用兜底 PlayBook禁止當成修復命令",
repair_candidate_next_step=(
"建立專屬 PlayBook 草案:綁定 alertname / target selector補 MCP evidence refs、"
"修復命令、rollback、verifier plan 與 owner review。"
),
repair_candidate_required_fields=[
"alertname",
"target_selector",
"mcp_evidence_refs",
"repair_command",
"rollback_command",
"verifier_plan",
"owner_review",
],
repair_candidate_work_item_href=(
"https://awoooi.wooo.work/zh-TW/awooop/work-items?"
"project_id=awoooi&incident_id=INC-20260611-34BBF5"
"&work_item_id=repair-candidate-draft%3Aawoooi%3AINC-20260611-34BBF5"
),
repair_candidate_work_item_id="repair-candidate-draft:awoooi:INC-20260611-34BBF5",
)
body = message.format()
assert "缺少可執行修復候選,已產生人工處置包" not in body
assert "Mode<code>repair_candidate_missing_ai_generation</code>" in body
assert "AI 修復候選補齊包" in body
assert "只命中通用兜底 PlayBook" in body
assert "建立專屬 PlayBook 草案" in body
assert "PlayBook 草案欄位" in body
assert "repair_command" in body
assert "工作項目" in body
assert "AwoooP 修復候選草案" in body
assert "https://awoooi.wooo.work/zh-TW/awooop/work-items?" in body
assert "補證據node_exporter target up" in body
assert "AI 建立修復候選" in body
assert "自動化資產總帳" in body
assert "KM<code>Knowledge Base</code>" in body
assert "PlayBook<code>Work Items</code>" in body
assert "腳本/Ansible<code>Runs</code>" in body
assert "排程/監控:<code>Observability</code>" in body
assert "Verifier<code>事件時間線</code>" in body
assert "不可視為自動化完成" in body
assert "按鈕:<b>Work Item</b> 開啟受控佇列" in body
assert "<b>處置包</b> 看完整證據" in body
assert "修復候選狀態" in body
assert "等待人工批准" not in body
def test_repair_candidate_draft_ready_card_exposes_controlled_queue_handoff() -> None:
message = TelegramMessage(
status_emoji="",
risk_level="LOW",
resource_name="node-exporter-188",
root_cause=(
"LLM fallback 後未開 runtime gate已產生受控自動化修復候選草案。"
"阻擋PlayBook 只有觀察或診斷步驟"
),
suggested_action=(
"DRAFT_READY - REPAIR_CANDIDATE_OWNER_REVIEW_REQUIRED: "
"PlayBook 只有觀察或診斷步驟"
),
estimated_downtime="unknown",
approval_id="test-approval-id",
incident_id="INC-20260625-977E5F",
primary_responsibility="OPENCLAW_PLAYBOOK_DRAFT",
confidence=0.0,
alert_category="host_resource",
repair_candidate_blocker_summary="PlayBook 只有觀察或診斷步驟",
repair_candidate_next_step=(
"把診斷命令保留為 MCP evidence collector另建獨立修復步驟、rollback "
"與 verifier進入 no-write rehearsal / check-mode 後才可 controlled apply。"
),
repair_candidate_required_fields=[
"alertname",
"target_selector",
"mcp_evidence_refs",
"repair_command",
"rollback_command",
"verifier_plan",
"owner_review",
],
repair_candidate_promotion_summary=(
"route=host_service_route_after_owner_review; promotion=11/11; "
"blocked=0; status=controlled_playbook_queue_ready; runtime=false"
),
repair_candidate_work_item_href=(
"https://awoooi.wooo.work/zh-TW/awooop/work-items?"
"project_id=awoooi&incident_id=INC-20260625-977E5F"
"&work_item_id=repair-candidate-draft%3Aawoooi%3AINC-20260625-977E5F"
),
repair_candidate_work_item_id="repair-candidate-draft:awoooi:INC-20260625-977E5F",
)
body = message.format()
assert "修復候選已具體成形" in body
assert "Mode<code>repair_candidate_draft_ready_controlled_queue</code>" in body
assert "AI 受控自動化佇列" in body
assert "no-write rehearsal" in body
assert "AI 排程" in body
assert "PlayBook 只有觀察或診斷步驟" in body
assert "把診斷命令保留為 MCP evidence collector" in body
assert "候選升級合約" in body
assert "route=host_service_route_after_owner_review" in body
assert "promotion=11/11" in body
assert "AwoooP 修復候選草案" in body
assert "自動化資產總帳" in body
assert "按鈕:<b>Work Item</b> 開啟受控佇列" in body
assert "缺少可執行修復候選" not in body
assert "等待人工批准" not in body
def test_nemotron_card_exposes_same_ai_automation_chain() -> None:
message = TelegramMessage(
status_emoji="🚨",
risk_level="CRITICAL",
resource_name="awoooi-api",
root_cause="Pod restart loop",
suggested_action="restart deployment/awoooi-api",
estimated_downtime="30s",
approval_id="test-approval-id",
incident_id="INC-20260429-TEST02",
primary_responsibility="INFRA",
confidence=0.86,
ai_provider="openclaw_nemo",
ai_model="llama-3.1-nemotron",
nemotron_enabled=True,
playbook_name="restart_deployment",
)
body = message.format_with_nemotron()
assert "AI 自動化鏈路" in body
assert "OpenClaw Nemo" in body
assert "tool_ready" in body
assert "restart_deployment" in body
assert "流程進度" in body
def test_action_required_card_exposes_truth_chain_progress() -> None:
message = TelegramMessage(
status_emoji="⚠️",
risk_level="LOW",
resource_name="awoooi-api",
root_cause="restart spike",
suggested_action="kubectl rollout restart deployment/awoooi-api",
estimated_downtime="30s",
approval_id="approval-id",
incident_id="INC-20260513-TEST03",
primary_responsibility="INFRA",
confidence=0.91,
playbook_name="restart_deployment",
automation_quality={
"verdict": "auto_repaired_verified",
"facts": {
"auto_repair_execution_records": 1,
"automation_operation_records": 1,
"verification_result": "success",
"mcp_gateway_total": 5,
"knowledge_entries": 2,
},
},
)
body = message.format()
assert "流程進度" in body
assert "執行:<code>auto_repair_recorded:1</code>" in body
assert "驗證:<code>success</code>" in body
assert "KM<code>2</code>" in body
assert "MCP<code>5</code>" in body
assert "已驗證自動修復" in body
assert "已驗證自動修復完成" in body
assert "等待人工批准" not in body
def test_action_required_card_does_not_call_diagnostic_ops_auto_repair() -> None:
message = TelegramMessage(
status_emoji="🚨",
risk_level="CRITICAL",
resource_name="dirty-reboot-evidence",
root_cause="Expert System: 偵測到高錯誤率",
suggested_action="ssh 192.168.0.110 'df -h /data/minio'",
estimated_downtime="0 min",
approval_id="approval-id",
incident_id="INC-20260530-88D960",
primary_responsibility="INFRA",
confidence=0.0,
playbook_name="rule_catalog",
automation_quality={
"verdict": "auto_repaired_verification_degraded",
"facts": {
"auto_repair_execution_records": 0,
"automation_operation_records": 1,
"effective_execution_records": 0,
"verification_result": "degraded",
"mcp_gateway_total": 22,
"knowledge_entries": 2,
},
},
)
body = message.format()
assert "已記錄診斷/觀察驗證結果degraded" in body
assert "執行:<code>diagnostic_recorded:1</code>" in body
assert "已記錄診斷/觀察,尚未證明修復" in body
assert "已自動執行" not in body