d7b3997b4a
Some checks failed
Code Review / ai-code-review (push) Successful in 14s
CD Pipeline / tests (push) Successful in 1m35s
CD Pipeline / build-and-deploy (push) Successful in 5m0s
CD Pipeline / post-deploy-checks (push) Successful in 1m33s
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
244 lines
8.8 KiB
Python
244 lines
8.8 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
from types import SimpleNamespace
|
|
from uuid import UUID
|
|
|
|
from src.services.approval_execution import ApprovalExecutionService
|
|
from src.services.operator_outcome import build_operator_outcome
|
|
|
|
|
|
def test_operator_outcome_marks_diagnostic_only_as_manual_action_required() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "execution_succeeded",
|
|
"stage_status": "success",
|
|
"needs_human": False,
|
|
"blockers": [],
|
|
},
|
|
automation_quality={
|
|
"verdict": "auto_repaired_verification_degraded",
|
|
"facts": {
|
|
"automation_operation_records": 1,
|
|
"effective_execution_records": 0,
|
|
"auto_repair_execution_records": 0,
|
|
"verification_result": "degraded",
|
|
"mcp_gateway_total": 22,
|
|
"knowledge_entries": 4,
|
|
},
|
|
"blockers": ["verification_recorded"],
|
|
},
|
|
source_id="INC-20260530-88D960",
|
|
)
|
|
|
|
assert outcome["schema_version"] == "operator_outcome_v1"
|
|
assert outcome["state"] == "diagnostic_only_manual_review"
|
|
assert outcome["needs_human"] is True
|
|
assert outcome["notification"]["mode"] == "action_required"
|
|
assert "telegram_sre_war_room" in outcome["notification"]["channels"]
|
|
assert outcome["next_action"] == "manual_review_or_collect_repair_evidence"
|
|
assert outcome["execution_result"]["completion_status"] == "completed_no_repair"
|
|
assert outcome["execution_result"]["repair_status"] == "not_executed"
|
|
assert outcome["execution_result"]["failure_status"] == "no_command_failed"
|
|
|
|
|
|
def test_operator_outcome_marks_unverified_execution_as_human_review() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "execution_succeeded",
|
|
"stage_status": "success",
|
|
"needs_human": False,
|
|
"blockers": [],
|
|
},
|
|
automation_quality={
|
|
"verdict": "execution_unverified",
|
|
"facts": {
|
|
"effective_execution_records": 1,
|
|
"auto_repair_execution_records": 0,
|
|
"verification_result": None,
|
|
},
|
|
"blockers": ["verification_recorded"],
|
|
},
|
|
)
|
|
|
|
assert outcome["state"] == "execution_unverified_manual_required"
|
|
assert outcome["needs_human"] is True
|
|
assert outcome["next_action"] == "run_or_review_post_execution_verification"
|
|
|
|
|
|
def test_operator_outcome_marks_ansible_check_mode_as_dry_run_only() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "execution_succeeded",
|
|
"stage_status": "success",
|
|
"needs_human": False,
|
|
"blockers": [],
|
|
},
|
|
automation_quality={
|
|
"verdict": "ansible_check_mode_only",
|
|
"facts": {
|
|
"effective_execution_records": 1,
|
|
"auto_repair_execution_records": 0,
|
|
"ansible_check_mode_total": 1,
|
|
"ansible_apply_total": 0,
|
|
"verification_result": None,
|
|
},
|
|
"blockers": [],
|
|
},
|
|
)
|
|
|
|
assert outcome["state"] == "dry_run_only_owner_review_required"
|
|
assert outcome["needs_human"] is True
|
|
assert outcome["next_action"] == "owner_review_apply_gate_or_create_verifier_plan"
|
|
assert outcome["execution_result"]["completion_status"] == "dry_run_completed_no_apply"
|
|
assert outcome["execution_result"]["command_status"] == "check_mode_succeeded"
|
|
assert outcome["execution_result"]["repair_status"] == "not_executed"
|
|
|
|
|
|
def test_operator_outcome_marks_verified_repair_as_result_only() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "execution_succeeded",
|
|
"stage_status": "success",
|
|
"needs_human": False,
|
|
"blockers": [],
|
|
},
|
|
automation_quality={
|
|
"verdict": "auto_repaired_verified",
|
|
"facts": {
|
|
"effective_execution_records": 1,
|
|
"auto_repair_execution_records": 1,
|
|
"verification_result": "success",
|
|
},
|
|
"blockers": [],
|
|
},
|
|
)
|
|
|
|
assert outcome["state"] == "completed_verified"
|
|
assert outcome["needs_human"] is False
|
|
assert outcome["notification"]["mode"] == "result_only"
|
|
assert outcome["execution_result"]["completion_status"] == "completed_verified"
|
|
assert outcome["execution_result"]["repair_status"] == "verified_repaired"
|
|
|
|
|
|
def test_operator_outcome_marks_rejected_approval_as_closed_no_execution() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "approval_rejected",
|
|
"stage_status": "closed",
|
|
"needs_human": False,
|
|
"blockers": [],
|
|
},
|
|
automation_quality={
|
|
"verdict": "approval_rejected_no_execution",
|
|
"facts": {},
|
|
"blockers": [],
|
|
},
|
|
)
|
|
|
|
assert outcome["state"] == "approval_rejected_no_execution"
|
|
assert outcome["needs_human"] is False
|
|
assert outcome["human_action_reason"] == "approval_rejected"
|
|
assert outcome["notification"]["mode"] == "result_only"
|
|
assert outcome["next_action"] == "monitor_or_reopen_if_alert_recurs"
|
|
assert outcome["execution_result"]["completion_status"] == "closed_no_execution"
|
|
|
|
|
|
def test_operator_outcome_marks_expired_approval_as_manual_review() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "approval_expired",
|
|
"stage_status": "expired",
|
|
"needs_human": True,
|
|
"blockers": ["approval_expired_without_operator_decision"],
|
|
},
|
|
automation_quality={
|
|
"verdict": "approval_expired_manual_review",
|
|
"facts": {},
|
|
"blockers": [],
|
|
},
|
|
)
|
|
|
|
assert outcome["state"] == "approval_expired_manual_review"
|
|
assert outcome["needs_human"] is True
|
|
assert outcome["notification"]["mode"] == "action_required"
|
|
assert outcome["next_action"] == "reopen_close_or_escalate_expired_approval"
|
|
assert outcome["execution_result"]["completion_status"] == "expired_no_execution"
|
|
|
|
|
|
def test_execution_result_message_includes_operator_outcome_and_human_channels() -> None:
|
|
service = ApprovalExecutionService()
|
|
approval = SimpleNamespace(
|
|
id=UUID("11111111-1111-4111-8111-111111111111"),
|
|
incident_id="INC-20260531-ABC123",
|
|
action="OBSERVE",
|
|
)
|
|
text = service._format_execution_result_message(
|
|
approval=approval,
|
|
success=True,
|
|
error=None,
|
|
no_action=True,
|
|
km_info="",
|
|
outcome=build_operator_outcome(
|
|
truth_status={"needs_human": True, "blockers": ["manual_gate"]},
|
|
automation_quality={
|
|
"verdict": "manual_required_no_action",
|
|
"facts": {},
|
|
"blockers": [],
|
|
},
|
|
),
|
|
)
|
|
|
|
assert "處置結果" in text
|
|
assert "執行判定" in text
|
|
assert "not_started_no_action" in text
|
|
assert "not_executed" in text
|
|
assert "人工: <code>yes</code>" in text
|
|
assert "telegram_sre_war_room" in text
|
|
assert "manual_review_no_action_decision" in text
|
|
|
|
|
|
def test_execution_result_message_does_not_call_diagnostic_success_repair_done() -> None:
|
|
service = ApprovalExecutionService()
|
|
approval = SimpleNamespace(
|
|
id=UUID("22222222-2222-4222-8222-222222222222"),
|
|
incident_id="INC-20260531-DIAG01",
|
|
action="ssh diagnose disk usage",
|
|
)
|
|
|
|
text = service._format_execution_result_message(
|
|
approval=approval,
|
|
success=True,
|
|
error=None,
|
|
no_action=False,
|
|
km_info="",
|
|
outcome=build_operator_outcome(
|
|
truth_status={"needs_human": False, "blockers": []},
|
|
automation_quality={
|
|
"verdict": "auto_repaired_verification_degraded",
|
|
"facts": {
|
|
"automation_operation_records": 1,
|
|
"effective_execution_records": 0,
|
|
"auto_repair_execution_records": 0,
|
|
"verification_result": "degraded",
|
|
},
|
|
"blockers": ["diagnostic_only"],
|
|
},
|
|
),
|
|
)
|
|
|
|
assert "已記錄診斷，尚未證明修復" in text
|
|
assert "completed_no_repair" in text
|
|
assert "no_command_failed" in text
|
|
assert "執行成功" not in text
|
|
assert "修復完成" not in text
|
|
|
|
|
|
def test_execution_result_sender_has_standalone_fallback_when_original_card_missing() -> None:
|
|
source = inspect.getsource(ApprovalExecutionService._push_execution_result_to_alert)
|
|
|
|
assert "push_execution_result_no_msg_id_standalone" in source
|
|
assert "reply_to_message_id" in source
|
|
assert "delivery" in source
|
|
assert "standalone" in source
|