e0a86b6254
Some checks failed
Code Review / ai-code-review (push) Successful in 13s
CD Pipeline / tests (push) Failing after 1m8s
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
343 lines
12 KiB
Python
343 lines
12 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_normalizes_missing_evidence_blockers() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "execution_succeeded",
|
|
"stage_status": "success",
|
|
"needs_human": True,
|
|
"blockers": [],
|
|
},
|
|
automation_quality={
|
|
"verdict": "execution_unverified",
|
|
"facts": {
|
|
"effective_execution_records": 1,
|
|
"auto_repair_execution_records": 0,
|
|
"verification_result": None,
|
|
"knowledge_entries": 0,
|
|
},
|
|
"blockers": [
|
|
"auto_repair_recorded",
|
|
"verification_recorded",
|
|
"learning_recorded",
|
|
],
|
|
},
|
|
)
|
|
|
|
assert "auto_repair_missing" in outcome["blockers"]
|
|
assert "verification_missing" in outcome["blockers"]
|
|
assert "learning_missing" in outcome["blockers"]
|
|
assert "auto_repair_recorded" not in outcome["blockers"]
|
|
assert "verification_recorded" not in outcome["blockers"]
|
|
assert "learning_recorded" not in outcome["blockers"]
|
|
|
|
|
|
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_controlled_verifier_required"
|
|
assert outcome["needs_human"] is False
|
|
assert outcome["next_action"] == "run_post_execution_verifier_or_rollback"
|
|
|
|
|
|
def test_operator_outcome_marks_ansible_check_mode_as_controlled_apply_ready() -> 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"] == "controlled_apply_queued"
|
|
assert outcome["needs_human"] is False
|
|
assert outcome["next_action"] == "wait_for_controlled_apply_and_post_apply_verifier"
|
|
assert outcome["execution_result"]["completion_status"] == (
|
|
"dry_run_passed_controlled_apply_queued"
|
|
)
|
|
assert outcome["execution_result"]["command_status"] == "check_mode_succeeded"
|
|
assert outcome["execution_result"]["repair_status"] == "controlled_apply_pending"
|
|
|
|
|
|
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_ai_retry"
|
|
assert outcome["needs_human"] is False
|
|
assert outcome["notification"]["mode"] == "action_required"
|
|
assert outcome["next_action"] == "ai_retry_or_rebuild_controlled_packet"
|
|
assert outcome["execution_result"]["completion_status"] == "expired_route_requeued"
|
|
|
|
|
|
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 "controlled_apply_evaluation" in text
|
|
assert "人工: <code>no</code>" in text
|
|
assert "telegram_sre_war_room" in text
|
|
assert "collect_evidence_or_generate_playbook_candidate" in text
|
|
|
|
|
|
def test_operator_outcome_marks_ansible_check_mode_blockers_as_controlled_apply_ready() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "execution_succeeded",
|
|
"stage_status": "success",
|
|
"needs_human": True,
|
|
"blockers": ["incident_open_after_successful_execution"],
|
|
},
|
|
automation_quality={
|
|
"verdict": "ansible_check_mode_only",
|
|
"facts": {
|
|
"ansible_check_mode_total": 1,
|
|
"ansible_apply_total": 0,
|
|
"auto_repair_execution_records": 0,
|
|
"effective_execution_records": 1,
|
|
"automation_operation_records": 2,
|
|
"verification_result": None,
|
|
"knowledge_entries": 0,
|
|
},
|
|
"blockers": ["auto_repair_recorded", "verification_recorded", "learning_recorded"],
|
|
},
|
|
)
|
|
|
|
assert outcome["state"] == "controlled_apply_queued"
|
|
assert outcome["next_action"] == "wait_for_controlled_apply_and_post_apply_verifier"
|
|
assert "受控自動 apply" in outcome["summary_zh"]
|
|
assert outcome["execution_result"]["completion_status"] == (
|
|
"dry_run_passed_controlled_apply_queued"
|
|
)
|
|
assert "auto_repair_missing" in outcome["blockers"]
|
|
assert "verification_missing" in outcome["blockers"]
|
|
assert "learning_missing" in outcome["blockers"]
|
|
|
|
|
|
def test_operator_outcome_keeps_failed_ansible_check_mode_with_ai_repair() -> None:
|
|
outcome = build_operator_outcome(
|
|
truth_status={
|
|
"current_stage": "execution_failed",
|
|
"stage_status": "failed",
|
|
"needs_human": True,
|
|
"blockers": ["ansible_check_mode_failed"],
|
|
},
|
|
automation_quality={
|
|
"verdict": "ansible_check_mode_only",
|
|
"facts": {
|
|
"ansible_check_mode_total": 1,
|
|
"ansible_apply_total": 0,
|
|
"auto_repair_execution_records": 0,
|
|
"effective_execution_records": 1,
|
|
"automation_operation_records": 2,
|
|
"verification_result": None,
|
|
"knowledge_entries": 0,
|
|
},
|
|
"blockers": ["ansible_check_mode_failed"],
|
|
},
|
|
)
|
|
|
|
assert outcome["state"] == "ai_playbook_repair_required"
|
|
assert outcome["needs_human"] is False
|
|
assert outcome["next_action"] == "auto_generate_playbook_or_transport_fix_candidate"
|
|
assert outcome["execution_result"]["completion_status"] == (
|
|
"dry_run_failed_ai_repairing_playbook_or_transport"
|
|
)
|
|
|
|
|
|
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
|