551227f3bb
Some checks failed
CD Pipeline / tests (push) Successful in 1m45s
CD Pipeline / build-and-deploy (push) Successful in 4m55s
CD Pipeline / post-deploy-checks (push) Successful in 1m58s
Code Review / ai-code-review (push) Has been cancelled
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
347 lines
13 KiB
Python
347 lines
13 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_ai_repair_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_ai_repair_required"
|
||
assert outcome["needs_human"] is False
|
||
assert outcome["notification"]["mode"] == "action_required"
|
||
assert "telegram_sre_war_room" in outcome["notification"]["channels"]
|
||
assert outcome["next_action"] == "auto_generate_repair_candidate_from_diagnostic_evidence"
|
||
assert outcome["execution_result"]["completion_status"] == (
|
||
"diagnostic_completed_ai_repair_queued"
|
||
)
|
||
assert outcome["execution_result"]["repair_status"] == (
|
||
"playbook_or_transport_repair_required"
|
||
)
|
||
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 "只完成診斷/觀察；AI 已排入修復候選補齊" in text
|
||
assert "diagnostic_completed_ai_repair_queued" 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
|