from __future__ import annotations import inspect from datetime import datetime, timedelta, timezone from types import SimpleNamespace from src.services.awooop_ansible_audit_service import ( build_ansible_decision_audit_payload, build_ansible_truth, ) from src.services.awooop_truth_chain_service import ( _automation_quality_score_bucket, build_automation_quality, build_incident_reconciliation, _clean_row, _incident_fingerprints, _summarize_gateway_mcp, _truth_status, fetch_truth_chain, summarize_automation_quality_records, ) from src.services.drift_repeat_state import ( build_drift_fingerprint, build_drift_repeat_state, ) def test_clean_row_parses_json_text_fields_for_gateway_visibility() -> None: row = { "gate_result": '{"schema_version":"legacy_mcp_bridge_v1","policy_enforced":false}', "source_envelope": '{"adapter":"legacy_telegram_gateway"}', "plain_text": '{"not":"parsed"}', } cleaned = _clean_row(row) assert cleaned["gate_result"]["schema_version"] == "legacy_mcp_bridge_v1" assert cleaned["gate_result"]["policy_enforced"] is False assert cleaned["source_envelope"]["adapter"] == "legacy_telegram_gateway" assert cleaned["plain_text"] == '{"not":"parsed"}' def test_incident_fingerprints_reads_signal_labels() -> None: fingerprints = _incident_fingerprints({ "incident_id": "INC-1", "signals": [ {"labels": {"fingerprint": "fp-label"}}, {"fingerprint": "fp-direct", "labels": {}}, {"labels": {"fingerprint": "fp-label"}}, ], }) assert fingerprints == ["fp-direct", "fp-label"] def test_fetch_truth_chain_can_match_inbound_provider_event_id() -> None: source = inspect.getsource(fetch_truth_chain) assert "provider_event_id = :source_id" in source def test_fetch_truth_chain_returns_inbound_redacted_envelope_fields() -> None: source = inspect.getsource(fetch_truth_chain) assert "content_redacted" in source assert "source_envelope" in source assert "source_refs,event_ids" in source assert "source_refs,incident_ids" in source assert "source_refs,sentry_issue_ids" in source assert "source_refs,signoz_alerts" in source def test_truth_status_marks_no_action_approval_as_manual_required() -> None: status = _truth_status( incident={"incident_id": "INC-1", "status": "INVESTIGATING"}, approvals=[{"status": "APPROVED", "action": "未知操作 | NO_ACTION"}], evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}], automation_ops=[], drift=None, drift_repeat_count=0, gateway_mcp_total=0, legacy_mcp_total=8, outbound_visible_total=0, ) assert status["current_stage"] == "manual_required" assert status["stage_status"] == "blocked" assert status["needs_human"] is True assert "approval_resolved_no_action_without_execution" in status["blockers"] assert "all_evidence_sensors_failed" in status["blockers"] assert "awooop_mcp_gateway_audit_empty" in status["blockers"] def test_truth_status_marks_inbound_only_source_as_received() -> None: status = _truth_status( incident=None, approvals=[], evidence_rows=[], automation_ops=[], drift=None, drift_repeat_count=0, gateway_mcp_total=0, legacy_mcp_total=0, outbound_visible_total=0, inbound_visible_total=1, ) assert status["current_stage"] == "inbound_received" assert status["stage_status"] == "observed" assert status["needs_human"] is False assert "awooop_mcp_gateway_audit_empty" in status["blockers"] def test_truth_status_does_not_treat_no_action_audit_as_execution() -> None: status = _truth_status( incident={"incident_id": "INC-1", "status": "RESOLVED"}, approvals=[{"status": "EXECUTION_SUCCESS", "action": "未知操作 | NO_ACTION"}], evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 6}], automation_ops=[ { "operation_type": "playbook_executed", "status": "success", "actor": "approval_execution", "output_reason": "NO_ACTION", "output_action": "未知操作 | NO_ACTION", }, { "operation_type": "ansible_candidate_matched", "status": "dry_run", "output_not_used_reason": "Ansible check-mode is not wired yet", }, ], drift=None, drift_repeat_count=0, gateway_mcp_total=8, legacy_mcp_total=8, outbound_visible_total=1, ) assert status["current_stage"] == "manual_required" assert status["stage_status"] == "blocked" assert status["needs_human"] is True assert "approval_resolved_no_action_without_execution" in status["blockers"] def test_truth_status_marks_open_incident_after_successful_execution() -> None: status = _truth_status( incident={"incident_id": "INC-OPEN", "status": "INVESTIGATING"}, approvals=[{"status": "EXECUTION_SUCCESS", "action": "kubectl rollout restart deployment/app"}], evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 6}], automation_ops=[], drift=None, drift_repeat_count=0, gateway_mcp_total=3, legacy_mcp_total=2, outbound_visible_total=1, auto_repair_executions=[{"success": True}], ) assert status["current_stage"] == "execution_succeeded" assert status["stage_status"] == "success" assert status["needs_human"] is True assert "incident_open_after_successful_execution" in status["blockers"] def test_truth_status_marks_repeated_pending_drift_as_human_needed() -> None: status = _truth_status( incident=None, approvals=[], evidence_rows=[], automation_ops=[], drift={ "report_id": "7f858956", "status": "pending", "interpretation": {"confidence": 0.0}, }, drift_repeat_count=12, gateway_mcp_total=0, legacy_mcp_total=0, outbound_visible_total=0, ) assert status["current_stage"] == "dedup_or_repeat_updated" assert status["stage_status"] == "pending" assert status["needs_human"] is True assert "drift_report_pending_without_resolution" in status["blockers"] assert "drift_ai_confidence_zero" in status["blockers"] def test_gateway_summary_surfaces_first_class_approval_execution() -> None: summary = _summarize_gateway_mcp([ { "agent_id": "approval_executor", "tool_name": "ssh_docker_restart", "result_status": "failed", "block_gate": None, "gate_result": { "schema_version": "awooop_mcp_gateway_audit_v1", "gateway_path": "awooop_mcp_gateway", "policy_enforced": True, "required_scope": "write", "is_shadow": False, "gate5_approval": True, }, } ]) assert summary["total"] == 1 assert summary["first_class_total"] == 1 assert summary["legacy_bridge_total"] == 0 assert summary["policy_enforced_total"] == 1 assert summary["approval_executor_total"] == 1 assert summary["stage"] == "provider_failed_after_gateway" assert summary["stage_status"] == "failed" assert summary["needs_human"] is True assert summary["by_agent"][0]["agent_id"] == "approval_executor" assert summary["by_tool"][0]["tool_name"] == "ssh_docker_restart" assert summary["by_scope"][0]["required_scope"] == "write" assert summary["by_scope"][0]["failed"] == 1 def _drift_item( *, resource_name: str = "awoooi-api", field_path: str = "spec.template.spec.containers[0].image", actual_value: str = "api:hotfix", ) -> dict: return { "resource_kind": "Deployment", "resource_name": resource_name, "namespace": "awoooi-prod", "field_path": field_path, "git_value": "api:main", "actual_value": actual_value, "drift_level": "high", "is_allowlisted": False, } def test_drift_fingerprint_is_stable_across_item_order() -> None: item_a = _drift_item(resource_name="awoooi-api") item_b = _drift_item( resource_name="awoooi-worker", field_path="spec.template.spec.serviceAccountName", actual_value="awoooi-executor", ) first = build_drift_fingerprint("awoooi-prod", [item_a, item_b]) second = build_drift_fingerprint("awoooi-prod", [item_b, item_a]) changed = build_drift_fingerprint( "awoooi-prod", [item_a, {**item_b, "actual_value": "different-service-account"}], ) assert first == second assert first.startswith("dfp_") assert first != changed def test_drift_repeat_state_counts_matching_fingerprint_only() -> None: now = datetime(2026, 5, 13, 1, 0, tzinfo=timezone.utc) report = { "report_id": "drift-now", "namespace": "awoooi-prod", "status": "pending", "scanned_at": now, "created_at": now, "items": [_drift_item()], } recent = [ { **report, "report_id": "drift-prev", "scanned_at": now - timedelta(hours=1), "created_at": now - timedelta(hours=1), }, { **report, "report_id": "drift-different", "scanned_at": now - timedelta(hours=2), "created_at": now - timedelta(hours=2), "items": [_drift_item(actual_value="api:other")], }, { **report, "report_id": "drift-old", "scanned_at": now - timedelta(hours=13), "created_at": now - timedelta(hours=13), }, ] repeat_state = build_drift_repeat_state(report, recent) assert repeat_state["schema_version"] == "drift_repeat_state_v1" assert repeat_state["fingerprint"].startswith("dfp_") assert repeat_state["matching_strategy"] == "namespace_and_stable_items_v1" assert repeat_state["occurrences_12h"] == 2 assert repeat_state["operator_stage"] == "pending_human" assert [row["report_id"] for row in repeat_state["reports"]] == [ "drift-now", "drift-prev", ] def test_reconciliation_blocks_open_incident_after_no_action_approval() -> None: reconciliation = build_incident_reconciliation( incident={"incident_id": "INC-1", "status": "INVESTIGATING"}, approvals=[ { "id": "approval-1", "status": "APPROVED", "action": "未知操作 | NO_ACTION", "resolved_at": "2026-05-13T01:00:00+00:00", } ], evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}], automation_ops=[], timeline_events=[], ) codes = {row["code"] for row in reconciliation["mismatches"]} assert reconciliation["schema_version"] == "incident_reconciliation_v1" assert reconciliation["consistency_status"] == "blocked" assert reconciliation["operator_next_state"] == "manual_required" assert reconciliation["facts"]["incident_closed"] is False assert reconciliation["facts"]["automation_operation_records"] == 0 assert "incident_open_after_approval_resolved" in codes assert "approval_approved_without_execution_record" in codes assert "approval_no_action_without_execution" in codes assert "evidence_all_sensors_failed" in codes assert "timeline_missing_for_approval" in codes def test_automation_quality_marks_no_action_without_execution() -> None: quality = build_automation_quality( incident={"incident_id": "INC-1", "status": "INVESTIGATING"}, approvals=[{"status": "APPROVED", "action": "未知操作 | NO_ACTION"}], evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}], automation_ops=[], auto_repair_executions=[], gateway_mcp_summary={"total": 8}, legacy_mcp_summary={"total": 8}, outbound_rows=[{"message_id": "m1"}], km_entries=[], timeline_events=[], ) gates = {row["name"]: row["status"] for row in quality["gates"]} assert quality["schema_version"] == "automation_quality_v1" assert quality["verdict"] == "manual_required_no_action" assert quality["facts"]["auto_repair_execution_records"] == 0 assert gates["execution_recorded"] == "missing" assert gates["verification_recorded"] == "not_applicable" assert "execution_recorded" in quality["blockers"] def test_automation_quality_ignores_no_action_audit_rows_as_execution() -> None: quality = build_automation_quality( incident={"incident_id": "INC-1", "status": "RESOLVED"}, approvals=[{"status": "EXECUTION_SUCCESS", "action": "未知操作 | NO_ACTION"}], evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 6}], automation_ops=[ { "operation_type": "playbook_executed", "status": "success", "actor": "approval_execution", "output_reason": "NO_ACTION", "output_action": "未知操作 | NO_ACTION", }, { "operation_type": "ansible_candidate_matched", "status": "dry_run", "output_not_used_reason": "Ansible check-mode is not wired yet", }, ], auto_repair_executions=[], gateway_mcp_summary={"total": 8}, legacy_mcp_summary={"total": 8}, outbound_rows=[{"message_id": "m1"}], km_entries=[{"id": "km-1"}], timeline_events=[{"id": "tl-1"}], ) gates = {row["name"]: row["status"] for row in quality["gates"]} assert quality["verdict"] == "manual_required_no_action" assert quality["facts"]["automation_operation_records"] == 2 assert quality["facts"]["effective_execution_records"] == 0 assert quality["facts"]["noop_operation_records"] == 1 assert quality["facts"]["audit_only_operation_records"] == 1 assert gates["execution_recorded"] == "missing" assert gates["verification_recorded"] == "not_applicable" def test_automation_quality_marks_verified_auto_repair() -> None: quality = build_automation_quality( incident={ "incident_id": "INC-2", "status": "RESOLVED", "verification_result": "success", }, approvals=[], evidence_rows=[{"sensors_attempted": 3, "sensors_succeeded": 3}], automation_ops=[], auto_repair_executions=[ { "id": "repair-1", "success": True, "playbook_id": "pb-1", } ], gateway_mcp_summary={"total": 3}, legacy_mcp_summary={"total": 3}, outbound_rows=[{"message_id": "m1"}], km_entries=[{"id": "km-1"}], timeline_events=[{"id": "tl-1"}], ) gates = {row["name"]: row["status"] for row in quality["gates"]} assert quality["verdict"] == "auto_repaired_verified" assert quality["facts"]["verification_result"] == "success" assert quality["score"] == 100 assert gates["auto_repair_recorded"] == "passed" assert gates["verification_recorded"] == "passed" assert quality["blockers"] == [] def test_automation_quality_score_buckets_are_stable() -> None: assert _automation_quality_score_bucket(100) == "green" assert _automation_quality_score_bucket(85) == "green" assert _automation_quality_score_bucket(84) == "yellow" assert _automation_quality_score_bucket(60) == "yellow" assert _automation_quality_score_bucket(59) == "red" def test_automation_quality_summary_denies_full_claim_when_unverified() -> None: summary = summarize_automation_quality_records( project_id="awoooi", window_hours=24, limit=200, records=[ { "incident": { "incident_id": "INC-OK", "alertname": "container recovered", "severity": "P4", "status": "RESOLVED", "created_at": "2026-05-13T01:00:00+00:00", }, "truth_status": { "current_stage": "execution_succeeded", "stage_status": "success", "needs_human": False, }, "automation_quality": { "applicable": True, "verdict": "auto_repaired_verified", "score": 100, "gates": [ {"name": "verification_recorded", "status": "passed"}, ], "blockers": [], }, }, { "incident": { "incident_id": "INC-GAP", "alertname": "low risk action", "severity": "P4", "status": "INVESTIGATING", "created_at": "2026-05-13T02:00:00+00:00", }, "truth_status": { "current_stage": "execution_succeeded", "stage_status": "success", "needs_human": False, }, "automation_quality": { "applicable": True, "verdict": "execution_unverified", "score": 65, "gates": [ {"name": "verification_recorded", "status": "missing"}, {"name": "learning_recorded", "status": "missing"}, ], "blockers": ["verification_recorded", "learning_recorded"], }, }, ], ) assert summary["schema_version"] == "automation_quality_summary_v1" assert summary["incident_total"] == 2 assert summary["evaluated_total"] == 2 assert summary["verified_auto_repair_total"] == 1 assert summary["score_buckets"] == {"green": 1, "yellow": 1, "red": 0} assert summary["production_claim"]["can_claim_full_auto_repair"] is False assert summary["production_claim"]["reason"] == "some_incidents_are_not_auto_repaired_verified" assert {row["verdict"]: row["total"] for row in summary["by_verdict"]} == { "auto_repaired_verified": 1, "execution_unverified": 1, } assert {row["gate"]: row["total"] for row in summary["gate_failures"]} == { "learning_recorded": 1, "verification_recorded": 1, } assert summary["examples"][1]["incident_id"] == "INC-GAP" assert summary["examples"][1]["score_bucket"] == "yellow" def test_reconciliation_marks_consistent_resolved_execution() -> None: reconciliation = build_incident_reconciliation( incident={"incident_id": "INC-2", "status": "RESOLVED"}, approvals=[ { "id": "approval-2", "status": "APPROVED", "action": "restart service", "resolved_at": "2026-05-13T01:00:00+00:00", } ], evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 7}], automation_ops=[{"status": "success"}], timeline_events=[{"event_type": "executor", "status": "success"}], ) assert reconciliation["consistency_status"] == "consistent" assert reconciliation["operator_next_state"] == "continue" assert reconciliation["mismatches"] == [] def test_ansible_truth_surfaces_audited_check_mode_record() -> None: truth = build_ansible_truth( [ { "op_id": "op-ansible-1", "operation_type": "ansible_check_mode_executed", "status": "dry_run", "actor": "platform_operator", "input_playbook_path": "infra/ansible/playbooks/188-ai-web.yml", "input_check_mode": "true", "dry_run_result": {"changed": 1}, "tags": ["ansible", "check_mode"], "created_at": "2026-05-12T22:00:00+08:00", } ], incident={"incident_id": "INC-1", "alertname": "momo pg_backup failed on 188"}, drift=None, ) assert truth["considered"] is True assert truth["not_used_reason"] is None assert truth["records"][0]["playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml" assert truth["records"][0]["check_mode"] == "true" assert truth["records"][0]["dry_run_result"] == {"changed": 1} assert "ansible_check_mode_executed" in truth["audit_contract"]["operation_types"] assert truth["candidate_catalog"]["decision_effect"] == "none" assert truth["candidate_catalog"]["candidates"][0]["catalog_id"] == "ansible:188-ai-web" assert truth["candidate_catalog"]["candidates"][0]["auto_apply_enabled"] is False def test_ansible_truth_keeps_catalog_hint_separate_from_runtime_use() -> None: truth = build_ansible_truth( [], incident={"incident_id": "INC-2", "alertname": "nginx 502 upstream timeout"}, drift=None, ) assert truth["considered"] is False assert truth["records"] == [] assert truth["not_used_reason"].startswith("no automation_operation_log row") assert truth["candidate_catalog"]["candidates"][0]["catalog_id"] == "ansible:nginx-sync" assert truth["candidate_catalog"]["candidates"][0]["approval_required"] is True assert truth["candidate_catalog"]["decision_effect"] == "none" def test_ansible_decision_audit_payload_is_dry_run_only() -> None: incident = SimpleNamespace( incident_id="INC-DOCKER", project_id="awoooi", alert_category="infrastructure", notification_type="TYPE-3", severity=SimpleNamespace(value="P3"), affected_services=["bitan-pharmacy-bitan-1"], signals=[ SimpleNamespace( alert_name="DockerContainerUnhealthy", labels={"alertname": "DockerContainerUnhealthy", "container": "bitan-pharmacy-bitan-1"}, annotations={}, ) ], ) payload = build_ansible_decision_audit_payload( incident=incident, proposal_data={"source": "expert_system", "risk_level": "low", "action": "NO_ACTION"}, decision_path="manual_approval", not_used_reason="manual approval required; Ansible check-mode is not wired yet", ) assert payload is not None assert payload["operation_type"] == "ansible_candidate_matched" assert payload["status"] == "dry_run" assert payload["input"]["executor"] == "ansible" assert payload["input"]["check_mode"] is True assert payload["input"]["apply_enabled"] is False assert payload["input"]["approval_required"] is True assert payload["input"]["executor_candidates"] assert payload["output"]["decision_effect"] == "audit_only" assert payload["dry_run_result"]["check_mode_executed"] is False