f0f6eee7a5
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m32s
CD Pipeline / build-and-deploy (push) Successful in 6m50s
CD Pipeline / post-deploy-checks (push) Successful in 2m22s
2972 lines
113 KiB
Python
2972 lines
113 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
import os
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src.core.config import settings
|
|
from src.jobs.awooop_ansible_candidate_backfill_job import (
|
|
enqueue_ai_decision_ansible_candidate,
|
|
)
|
|
from src.services.awooop_ansible_audit_service import (
|
|
build_ansible_decision_audit_payload,
|
|
build_ansible_truth,
|
|
record_ansible_decision_audit,
|
|
)
|
|
from src.services.awooop_ansible_check_mode_service import (
|
|
_EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE,
|
|
AnsibleCheckModeClaim,
|
|
AnsibleRunResult,
|
|
_append_alert_lifecycle_receipt,
|
|
_append_runtime_stage_receipts_to_apply,
|
|
_automation_operation_log_incident_id,
|
|
_build_auto_repair_execution_receipt,
|
|
_claim_from_apply_operation_row,
|
|
_claim_from_stale_check_mode_row,
|
|
_execution_capability_timeout_seconds,
|
|
_expire_stale_ansible_execution_capabilities,
|
|
_issue_ansible_execution_capability,
|
|
_load_open_failed_apply_retry_row,
|
|
_load_pre_decision_context_runtime_stage_receipts,
|
|
_post_apply_km_path_type,
|
|
_post_apply_verification_result,
|
|
_record_auto_repair_execution_receipt,
|
|
_record_learning_writeback_receipt,
|
|
_record_retry_runtime_stage_receipt,
|
|
_record_runtime_stage_receipts,
|
|
_record_timeline_projection_receipt,
|
|
_resolve_execution_capability_operation_type,
|
|
_revoke_ansible_execution_capability,
|
|
_run_ansible_command,
|
|
_send_controlled_apply_telegram_receipt,
|
|
backfill_missing_auto_repair_execution_receipts_once,
|
|
build_ansible_apply_command,
|
|
build_ansible_check_mode_claim_input,
|
|
build_ansible_check_mode_command,
|
|
build_ansible_context_runtime_stage_receipts,
|
|
build_ansible_post_apply_runtime_stage_receipts,
|
|
build_ansible_pre_apply_runtime_stage_receipts,
|
|
build_ansible_timeline_runtime_stage_receipt,
|
|
claim_catalog_drift_failed_check_modes,
|
|
claim_pending_check_modes,
|
|
claim_stale_pending_check_modes,
|
|
detect_ansible_transport_blockers,
|
|
finalize_check_mode_claim,
|
|
recent_ansible_transport_blockers,
|
|
run_controlled_apply_for_claim,
|
|
run_failed_apply_check_mode_replay_once,
|
|
run_pending_check_modes_once,
|
|
)
|
|
from src.services.awooop_ansible_learning_writeback import (
|
|
_ensure_rag_chunk_fallback,
|
|
_playbook_trust_readback,
|
|
canonical_ansible_playbook_id,
|
|
ensure_ansible_rag_writeback,
|
|
record_ansible_playbook_trust_writeback,
|
|
resolve_ansible_playbook_identity,
|
|
)
|
|
from src.services.awooop_truth_chain_service import (
|
|
_ansible_playbook_roots,
|
|
_ansible_runtime_readiness,
|
|
_automation_quality_score_bucket,
|
|
_clean_row,
|
|
_execution_backend_summary,
|
|
_fetch_inbound_conversation_event_rows,
|
|
_incident_fingerprints,
|
|
_summarize_gateway_mcp,
|
|
_truth_status,
|
|
build_automation_quality,
|
|
build_incident_reconciliation,
|
|
fetch_automation_quality_summary,
|
|
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_inbound_conversation_event_rows)
|
|
|
|
assert "provider_event_id = :source_id" in source
|
|
|
|
|
|
def test_quality_summary_includes_recent_ansible_operation_incidents() -> None:
|
|
source = inspect.getsource(fetch_automation_quality_summary)
|
|
|
|
assert "FROM automation_operation_log" in source
|
|
assert "input ->> 'incident_id'" in source
|
|
assert "JOIN incidents ON incidents.incident_id = source_ids.incident_id" in source
|
|
assert "source_ids.recent_evidence_at DESC" in source
|
|
|
|
|
|
def test_quality_summary_uses_batched_truth_chain_inputs() -> None:
|
|
source = inspect.getsource(fetch_automation_quality_summary)
|
|
|
|
assert "fetch_truth_chain(" not in source
|
|
assert "approval_records" in source
|
|
assert "incident_evidence" in source
|
|
assert "awooop_outbound_message" in source
|
|
assert "_build_summary_quality_records" in source
|
|
|
|
|
|
def test_ansible_audit_keeps_external_incident_id_in_json_not_bigint_column() -> None:
|
|
decision_source = inspect.getsource(record_ansible_decision_audit)
|
|
claim_source = inspect.getsource(claim_pending_check_modes)
|
|
|
|
assert "operation_type, actor, status, incident_id" in decision_source
|
|
assert "coalesce(incident_id::text, input ->> 'incident_id')" in decision_source
|
|
assert "operation_type, actor, status, incident_id" in claim_source
|
|
assert "incident_db_id" in decision_source
|
|
assert "incident_db_id" in claim_source
|
|
assert "NULLIF(:incident_id, '')" not in decision_source
|
|
assert "NULLIF(:incident_id, '')" not in claim_source
|
|
assert _automation_operation_log_incident_id("INC-20260530-0E5C5C") is None
|
|
assert _automation_operation_log_incident_id("12345") == 12345
|
|
|
|
|
|
def test_ansible_transport_cooldown_uses_asyncpg_safe_interval_parameter() -> None:
|
|
source = inspect.getsource(recent_ansible_transport_blockers)
|
|
|
|
assert ":cooldown_seconds * INTERVAL '1 second'" in source
|
|
assert "CAST(:cooldown AS interval)" not in source
|
|
assert "include_repair_forced_command_blocker" in source
|
|
|
|
|
|
def test_fetch_truth_chain_returns_inbound_redacted_envelope_fields() -> None:
|
|
source = inspect.getsource(fetch_truth_chain)
|
|
helper_source = inspect.getsource(_fetch_inbound_conversation_event_rows)
|
|
|
|
assert "content_redacted" in source
|
|
assert "source_envelope" in helper_source
|
|
assert "source_refs,event_ids" in helper_source
|
|
assert "source_refs,incident_ids" in helper_source
|
|
assert "source_refs,sentry_issue_ids" in helper_source
|
|
assert "source_refs,signoz_alerts" in helper_source
|
|
|
|
|
|
def test_fetch_truth_chain_splits_inbound_lookup_into_index_friendly_branches() -> None:
|
|
source = inspect.getsource(fetch_truth_chain)
|
|
helper_source = inspect.getsource(_fetch_inbound_conversation_event_rows)
|
|
|
|
assert "await _fetch_inbound_conversation_event_rows(" in source
|
|
assert "run_id::text = :source_id" in helper_source
|
|
assert "provider_event_id = :source_id" in helper_source
|
|
assert "source_refs,event_ids" in helper_source
|
|
assert "source_refs,incident_ids" in helper_source
|
|
assert "content_preview ILIKE :source_needle" in helper_source
|
|
assert "len(rows_by_id) < branch_limit" in helper_source
|
|
assert "OR content_preview ILIKE :source_needle" not in helper_source
|
|
assert "OR coalesce(source_envelope #>" not in helper_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_keeps_verified_controlled_apply_autonomous_when_incident_open() -> None:
|
|
status = _truth_status(
|
|
incident={
|
|
"incident_id": "INC-OPEN-VERIFIED",
|
|
"status": "INVESTIGATING",
|
|
"verification_result": "success",
|
|
},
|
|
approvals=[{"status": "EXECUTION_SUCCESS", "action": "ansible controlled apply"}],
|
|
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 False
|
|
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=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_drift_repeat_state_can_group_semantic_shape_without_values() -> None:
|
|
now = datetime(2026, 5, 19, 1, 0, tzinfo=UTC)
|
|
report = {
|
|
"report_id": "drift-now",
|
|
"namespace": "awoooi-prod",
|
|
"status": "pending",
|
|
"scanned_at": now,
|
|
"created_at": now,
|
|
"items": [_drift_item(actual_value=["vol-a", "vol-b"])],
|
|
}
|
|
recent = [
|
|
{
|
|
**report,
|
|
"report_id": "drift-prev",
|
|
"scanned_at": now - timedelta(hours=1),
|
|
"created_at": now - timedelta(hours=1),
|
|
"items": [_drift_item(actual_value=["vol-a", "vol-b", "vol-c"])],
|
|
},
|
|
]
|
|
|
|
strict_state = build_drift_repeat_state(report, recent)
|
|
semantic_state = build_drift_repeat_state(
|
|
report,
|
|
recent,
|
|
include_values=False,
|
|
)
|
|
|
|
assert strict_state["occurrences_12h"] == 1
|
|
assert semantic_state["matching_strategy"] == "namespace_resource_field_level_v2"
|
|
assert semantic_state["occurrences_12h"] == 2
|
|
assert semantic_state["fingerprint"] != semantic_state["strict_fingerprint"]
|
|
|
|
|
|
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_reconciliation_accepts_approval_with_raw_timeline_event() -> None:
|
|
reconciliation = build_incident_reconciliation(
|
|
incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
|
|
approvals=[
|
|
{
|
|
"id": "approval-1",
|
|
"status": "PENDING",
|
|
"action": "kubectl rollout restart deployment/api",
|
|
}
|
|
],
|
|
evidence_rows=[],
|
|
automation_ops=[],
|
|
timeline_events=[
|
|
{
|
|
"event_type": "human",
|
|
"status": "warning",
|
|
"approval_id": "approval-1",
|
|
}
|
|
],
|
|
)
|
|
|
|
codes = {row["code"] for row in reconciliation["mismatches"]}
|
|
assert "timeline_missing_for_approval" not in codes
|
|
assert reconciliation["facts"]["timeline_events"] == 1
|
|
|
|
|
|
def test_reconciliation_counts_auto_repair_execution_as_real_execution() -> None:
|
|
reconciliation = build_incident_reconciliation(
|
|
incident={"incident_id": "INC-2", "status": "INVESTIGATING"},
|
|
approvals=[
|
|
{
|
|
"id": "approval-2",
|
|
"status": "APPROVED",
|
|
"action": "未知操作 | NO_ACTION",
|
|
"resolved_at": "2026-05-13T01:00:00+00:00",
|
|
}
|
|
],
|
|
evidence_rows=[
|
|
{
|
|
"sensors_attempted": 8,
|
|
"sensors_succeeded": 6,
|
|
"verification_result": "degraded",
|
|
}
|
|
],
|
|
automation_ops=[],
|
|
auto_repair_executions=[
|
|
{
|
|
"id": "repair-1",
|
|
"success": True,
|
|
"playbook_id": "PB-1",
|
|
}
|
|
],
|
|
timeline_events=[{"event_type": "executor", "status": "success"}],
|
|
)
|
|
|
|
codes = {row["code"] for row in reconciliation["mismatches"]}
|
|
assert reconciliation["consistency_status"] == "blocked"
|
|
assert reconciliation["facts"]["auto_repair_execution_records"] == 1
|
|
assert reconciliation["facts"]["successful_auto_repair_records"] == 1
|
|
assert reconciliation["facts"]["effective_execution_records"] == 1
|
|
assert "incident_open_after_successful_execution" in codes
|
|
assert "verification_degraded_after_auto_repair" in codes
|
|
assert "approval_approved_without_execution_record" not in codes
|
|
assert "approval_no_action_without_execution" not 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_marks_rejected_approval_closed_without_execution() -> None:
|
|
quality = build_automation_quality(
|
|
incident={"incident_id": "INC-REJECTED", "status": "INVESTIGATING"},
|
|
approvals=[{"status": "REJECTED", "action": "kubectl rollout restart deployment/api"}],
|
|
evidence_rows=[],
|
|
automation_ops=[],
|
|
auto_repair_executions=[],
|
|
gateway_mcp_summary={"total": 1},
|
|
legacy_mcp_summary={"total": 1},
|
|
outbound_rows=[{"message_id": "m1"}],
|
|
km_entries=[],
|
|
timeline_events=[],
|
|
)
|
|
|
|
gates = {row["name"]: row["status"] for row in quality["gates"]}
|
|
assert quality["verdict"] == "approval_rejected_no_execution"
|
|
assert gates["approval_state"] == "passed"
|
|
assert gates["execution_recorded"] == "missing"
|
|
|
|
|
|
def test_automation_quality_marks_expired_approval_for_ai_retry() -> None:
|
|
quality = build_automation_quality(
|
|
incident={"incident_id": "INC-EXPIRED", "status": "INVESTIGATING"},
|
|
approvals=[{"status": "EXPIRED", "action": "kubectl rollout restart deployment/api"}],
|
|
evidence_rows=[],
|
|
automation_ops=[],
|
|
auto_repair_executions=[],
|
|
gateway_mcp_summary={"total": 1},
|
|
legacy_mcp_summary={"total": 1},
|
|
outbound_rows=[{"message_id": "m1"}],
|
|
km_entries=[],
|
|
timeline_events=[],
|
|
)
|
|
|
|
gates = {row["name"]: row["status"] for row in quality["gates"]}
|
|
assert quality["verdict"] == "approval_expired_ai_retry"
|
|
assert gates["approval_state"] == "warning"
|
|
assert "approval_state" not in quality["blockers"]
|
|
|
|
|
|
def test_truth_status_marks_rejected_approval_as_closed() -> None:
|
|
status = _truth_status(
|
|
incident={"incident_id": "INC-REJECTED", "status": "INVESTIGATING"},
|
|
approvals=[{"status": "REJECTED", "action": "kubectl rollout restart deployment/api"}],
|
|
evidence_rows=[],
|
|
automation_ops=[],
|
|
drift=None,
|
|
drift_repeat_count=0,
|
|
gateway_mcp_total=1,
|
|
legacy_mcp_total=1,
|
|
outbound_visible_total=1,
|
|
auto_repair_executions=[],
|
|
)
|
|
|
|
assert status["current_stage"] == "approval_rejected"
|
|
assert status["stage_status"] == "closed"
|
|
assert status["needs_human"] is False
|
|
|
|
|
|
def test_truth_status_marks_expired_approval_as_manual_needed() -> None:
|
|
status = _truth_status(
|
|
incident={"incident_id": "INC-EXPIRED", "status": "INVESTIGATING"},
|
|
approvals=[{"status": "EXPIRED", "action": "kubectl rollout restart deployment/api"}],
|
|
evidence_rows=[],
|
|
automation_ops=[],
|
|
drift=None,
|
|
drift_repeat_count=0,
|
|
gateway_mcp_total=1,
|
|
legacy_mcp_total=1,
|
|
outbound_visible_total=1,
|
|
auto_repair_executions=[],
|
|
)
|
|
|
|
assert status["current_stage"] == "approval_expired"
|
|
assert status["stage_status"] == "expired"
|
|
assert status["needs_human"] is True
|
|
assert "approval_expired_without_operator_decision" in status["blockers"]
|
|
|
|
|
|
def test_reconciliation_ignores_no_action_audit_rows_as_execution() -> None:
|
|
reconciliation = build_incident_reconciliation(
|
|
incident={"incident_id": "INC-NOOP", "status": "INVESTIGATING"},
|
|
approvals=[
|
|
{
|
|
"id": "approval-noop",
|
|
"status": "APPROVED",
|
|
"action": "未知操作 | NO_ACTION",
|
|
"resolved_at": "2026-05-31T01:00:00+00:00",
|
|
}
|
|
],
|
|
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",
|
|
}
|
|
],
|
|
auto_repair_executions=[],
|
|
timeline_events=[{"event_type": "executor", "status": "success"}],
|
|
)
|
|
|
|
codes = {row["code"] for row in reconciliation["mismatches"]}
|
|
assert reconciliation["facts"]["executed_operation_records"] == 0
|
|
assert reconciliation["facts"]["effective_execution_records"] == 0
|
|
assert "approval_approved_without_execution_record" in codes
|
|
assert "approval_no_action_without_execution" in codes
|
|
assert "incident_open_after_successful_execution" not in codes
|
|
|
|
|
|
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_uses_approval_metadata_to_suppress_diagnostic_ops() -> None:
|
|
quality = build_automation_quality(
|
|
incident={"incident_id": "INC-DIAG", "status": "RESOLVED"},
|
|
approvals=[
|
|
{
|
|
"status": "EXECUTION_SUCCESS",
|
|
"action": "ssh 192.168.0.110 'df -h /data/minio'",
|
|
"extra_metadata": {
|
|
"execution_kind": "diagnostic",
|
|
"repair_executed": False,
|
|
},
|
|
}
|
|
],
|
|
evidence_rows=[
|
|
{
|
|
"sensors_attempted": 8,
|
|
"sensors_succeeded": 6,
|
|
"verification_result": "degraded",
|
|
}
|
|
],
|
|
automation_ops=[
|
|
{
|
|
"operation_type": "playbook_executed",
|
|
"status": "success",
|
|
"actor": "approval_execution",
|
|
"output_action": "ssh 192.168.0.110 'df -h /data/minio'",
|
|
}
|
|
],
|
|
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_diagnostic_only"
|
|
assert quality["facts"]["automation_operation_records"] == 1
|
|
assert quality["facts"]["effective_execution_records"] == 0
|
|
assert quality["facts"]["approval_repair_suppressed"] is True
|
|
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_marks_degraded_auto_repair_verification() -> None:
|
|
quality = build_automation_quality(
|
|
incident={
|
|
"incident_id": "INC-3",
|
|
"status": "INVESTIGATING",
|
|
},
|
|
approvals=[{"status": "APPROVED", "action": "未知操作 | NO_ACTION"}],
|
|
evidence_rows=[
|
|
{
|
|
"sensors_attempted": 8,
|
|
"sensors_succeeded": 6,
|
|
"verification_result": "degraded",
|
|
}
|
|
],
|
|
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=[],
|
|
timeline_events=[{"id": "tl-1"}],
|
|
)
|
|
|
|
gates = {row["name"]: row["status"] for row in quality["gates"]}
|
|
assert quality["verdict"] == "auto_repaired_verification_degraded"
|
|
assert quality["facts"]["auto_repair_execution_records"] == 1
|
|
assert quality["facts"]["verification_result"] == "degraded"
|
|
assert gates["execution_recorded"] == "passed"
|
|
assert gates["verification_recorded"] == "warning"
|
|
|
|
|
|
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,
|
|
"facts": {
|
|
"automation_operation_records": 1,
|
|
"effective_execution_records": 1,
|
|
"noop_operation_records": 0,
|
|
"audit_only_operation_records": 0,
|
|
"auto_repair_execution_records": 1,
|
|
},
|
|
"gates": [
|
|
{"name": "verification_recorded", "status": "passed"},
|
|
],
|
|
"blockers": [],
|
|
},
|
|
"execution": {
|
|
"automation_operation_log": [
|
|
{
|
|
"operation_type": "playbook_executed",
|
|
"status": "success",
|
|
}
|
|
],
|
|
"auto_repair_executions": [{"success": True}],
|
|
"ansible": {
|
|
"considered": False,
|
|
"records": [],
|
|
"candidate_catalog": {"candidates": []},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"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,
|
|
"facts": {
|
|
"automation_operation_records": 2,
|
|
"effective_execution_records": 0,
|
|
"noop_operation_records": 0,
|
|
"audit_only_operation_records": 2,
|
|
"auto_repair_execution_records": 0,
|
|
},
|
|
"gates": [
|
|
{"name": "verification_recorded", "status": "missing"},
|
|
{"name": "learning_recorded", "status": "missing"},
|
|
],
|
|
"blockers": ["verification_recorded", "learning_recorded"],
|
|
},
|
|
"execution": {
|
|
"automation_operation_log": [
|
|
{
|
|
"operation_type": "ansible_candidate_matched",
|
|
"status": "dry_run",
|
|
},
|
|
{
|
|
"operation_type": "ansible_check_mode_executed",
|
|
"status": "dry_run",
|
|
},
|
|
],
|
|
"auto_repair_executions": [],
|
|
"ansible": {
|
|
"considered": True,
|
|
"records": [
|
|
{
|
|
"operation_type": "ansible_candidate_matched",
|
|
"status": "dry_run",
|
|
},
|
|
{
|
|
"operation_type": "ansible_check_mode_executed",
|
|
"status": "dry_run",
|
|
},
|
|
],
|
|
"candidate_catalog": {
|
|
"candidates": [
|
|
{"catalog_id": "ansible:188-ai-web"},
|
|
{"catalog_id": "ansible:nginx-sync"},
|
|
]
|
|
},
|
|
},
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
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["execution_backend_summary"] == {
|
|
"operation_records_total": 3,
|
|
"effective_execution_records_total": 1,
|
|
"noop_operation_records_total": 0,
|
|
"audit_only_operation_records_total": 2,
|
|
"auto_repair_execution_records_total": 1,
|
|
"ansible_considered_total": 1,
|
|
"ansible_audit_record_total": 2,
|
|
"ansible_candidate_total": 2,
|
|
"ansible_check_mode_total": 1,
|
|
"ansible_apply_total": 0,
|
|
"ansible_rollback_total": 0,
|
|
"ansible_pending_check_mode_total": 1,
|
|
}
|
|
assert summary["ansible_runtime"]["playbook_root_present"] is True
|
|
assert summary["ansible_runtime"]["inventory_present"] is True
|
|
assert summary["ansible_runtime"]["playbook_count"] >= 1
|
|
assert "ansible_playbook_binary_present" in summary["ansible_runtime"]
|
|
assert summary["examples"][1]["incident_id"] == "INC-GAP"
|
|
assert summary["examples"][1]["score_bucket"] == "yellow"
|
|
|
|
|
|
def test_automation_quality_summary_builds_operator_flow_gates() -> None:
|
|
def quality(
|
|
*,
|
|
incident_id: str,
|
|
verdict: str,
|
|
score: int,
|
|
gates: list[dict[str, str]],
|
|
blockers: list[str] | None = None,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"incident": {
|
|
"incident_id": incident_id,
|
|
"alertname": "container restart loop",
|
|
"severity": "P3",
|
|
"status": "INVESTIGATING",
|
|
},
|
|
"truth_status": {
|
|
"current_stage": "execution_succeeded",
|
|
"stage_status": "success",
|
|
"needs_human": verdict != "auto_repaired_verified",
|
|
},
|
|
"automation_quality": {
|
|
"applicable": True,
|
|
"verdict": verdict,
|
|
"score": score,
|
|
"facts": {
|
|
"automation_operation_records": 1,
|
|
"effective_execution_records": 1 if verdict == "auto_repaired_verified" else 0,
|
|
"auto_repair_execution_records": 1 if verdict == "auto_repaired_verified" else 0,
|
|
},
|
|
"gates": gates,
|
|
"blockers": blockers or [],
|
|
},
|
|
"execution": {
|
|
"automation_operation_log": [],
|
|
"auto_repair_executions": [],
|
|
"ansible": {"considered": False, "records": [], "candidate_catalog": {"candidates": []}},
|
|
},
|
|
}
|
|
|
|
passed_gates = [
|
|
{"name": "source_persisted", "status": "passed"},
|
|
{"name": "outbound_recorded", "status": "passed"},
|
|
{"name": "evidence_collected", "status": "passed"},
|
|
{"name": "mcp_gateway_observed", "status": "passed"},
|
|
{"name": "approval_state", "status": "not_applicable"},
|
|
{"name": "execution_recorded", "status": "passed"},
|
|
{"name": "auto_repair_recorded", "status": "passed"},
|
|
{"name": "verification_recorded", "status": "passed"},
|
|
{"name": "learning_recorded", "status": "passed"},
|
|
{"name": "timeline_recorded", "status": "passed"},
|
|
]
|
|
blocked_gates = [
|
|
{"name": "source_persisted", "status": "passed"},
|
|
{"name": "outbound_recorded", "status": "passed"},
|
|
{"name": "evidence_collected", "status": "missing"},
|
|
{"name": "mcp_gateway_observed", "status": "missing"},
|
|
{"name": "approval_state", "status": "warning"},
|
|
{"name": "execution_recorded", "status": "missing"},
|
|
{"name": "auto_repair_recorded", "status": "missing"},
|
|
{"name": "verification_recorded", "status": "not_applicable"},
|
|
{"name": "learning_recorded", "status": "not_applicable"},
|
|
{"name": "timeline_recorded", "status": "missing"},
|
|
]
|
|
|
|
summary = summarize_automation_quality_records(
|
|
project_id="awoooi",
|
|
window_hours=24,
|
|
limit=2,
|
|
records=[
|
|
quality(
|
|
incident_id="INC-PASS",
|
|
verdict="auto_repaired_verified",
|
|
score=100,
|
|
gates=passed_gates,
|
|
),
|
|
quality(
|
|
incident_id="INC-BLOCKED",
|
|
verdict="manual_required_no_action",
|
|
score=40,
|
|
gates=blocked_gates,
|
|
blockers=["mcp_gateway_observed", "execution_recorded"],
|
|
),
|
|
],
|
|
)
|
|
|
|
flow = summary["automation_flow_gates"]
|
|
gates = {row["gate"]: row for row in flow["gates"]}
|
|
|
|
assert flow["schema_version"] == "automation_flow_gate_summary_v1"
|
|
assert flow["overall_status"] == "blocked"
|
|
assert gates["alert_intake"]["status"] == "passed"
|
|
assert gates["mcp_investigation"]["missing_total"] == 1
|
|
assert gates["approval_policy"]["warning_total"] == 1
|
|
assert gates["execution_recorded"]["missing_total"] == 1
|
|
assert gates["verification_recorded"]["missing_total"] == 1
|
|
assert gates["knowledge_recorded"]["missing_total"] == 1
|
|
assert gates["operator_visible"]["missing_total"] == 1
|
|
assert gates["mcp_investigation"]["examples"][0]["incident_id"] == "INC-BLOCKED"
|
|
|
|
|
|
def test_ansible_runtime_readiness_reports_check_mode_blockers() -> None:
|
|
readiness = _ansible_runtime_readiness()
|
|
|
|
assert readiness["playbook_root_present"] is True
|
|
assert readiness["inventory_present"] is True
|
|
assert readiness["playbook_count"] >= 1
|
|
assert isinstance(readiness["can_run_check_mode"], bool)
|
|
assert isinstance(readiness["blockers"], list)
|
|
assert "check_mode_transport_profile" in readiness
|
|
assert "check_mode_ssh_key_readable" in readiness
|
|
assert "check_mode_known_hosts_readable" in readiness
|
|
assert "repair_ssh_key_readable" in readiness
|
|
assert "repair_known_hosts_readable" in readiness
|
|
|
|
|
|
def test_ansible_runtime_readiness_requires_check_mode_ssh_material(tmp_path: Path) -> None:
|
|
missing_key = tmp_path / "missing-id-ed25519"
|
|
missing_known_hosts = tmp_path / "missing-known-hosts"
|
|
|
|
readiness = _ansible_runtime_readiness(
|
|
check_mode_ssh_key_path=missing_key,
|
|
check_mode_known_hosts_path=missing_known_hosts,
|
|
)
|
|
|
|
assert readiness["check_mode_ssh_key_present"] is False
|
|
assert readiness["check_mode_ssh_key_readable"] is False
|
|
assert readiness["check_mode_known_hosts_present"] is False
|
|
assert readiness["check_mode_known_hosts_readable"] is False
|
|
assert "ansible_check_mode_ssh_key_missing" in readiness["blockers"]
|
|
assert "ansible_check_mode_known_hosts_missing" in readiness["blockers"]
|
|
|
|
|
|
def test_ansible_runtime_readiness_accepts_readable_check_mode_ssh_material(tmp_path: Path) -> None:
|
|
key_path = tmp_path / "id_ed25519"
|
|
known_hosts_path = tmp_path / "known_hosts"
|
|
key_path.write_text("test-key", encoding="utf-8")
|
|
known_hosts_path.write_text("192.168.0.110 ssh-ed25519 AAAATEST", encoding="utf-8")
|
|
key_path.chmod(0o400)
|
|
known_hosts_path.chmod(0o400)
|
|
|
|
readiness = _ansible_runtime_readiness(
|
|
check_mode_ssh_key_path=key_path,
|
|
check_mode_known_hosts_path=known_hosts_path,
|
|
)
|
|
|
|
assert readiness["check_mode_ssh_key_present"] is True
|
|
assert readiness["check_mode_ssh_key_readable"] is True
|
|
assert readiness["check_mode_known_hosts_present"] is True
|
|
assert readiness["check_mode_known_hosts_readable"] is True
|
|
assert "ansible_check_mode_ssh_key_missing" not in readiness["blockers"]
|
|
assert "ansible_check_mode_known_hosts_missing" not in readiness["blockers"]
|
|
|
|
|
|
def test_ansible_playbook_roots_supports_flat_container_module_path() -> None:
|
|
roots = _ansible_playbook_roots(Path("/app/src/services/awooop_truth_chain_service.py"))
|
|
|
|
assert Path("/app/infra/ansible") in roots
|
|
assert Path("/app/src/infra/ansible") in roots
|
|
|
|
|
|
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_catalog_id": "ansible:188-momo-backup-user",
|
|
"input_execution_mode": "check_mode",
|
|
"input_playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
|
|
"input_check_mode": "true",
|
|
"dry_run_result": {
|
|
"changed": 1,
|
|
"check_mode_executed": True,
|
|
"apply_executed": False,
|
|
"returncode": 0,
|
|
},
|
|
"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]["catalog_id"] == "ansible:188-momo-backup-user"
|
|
assert truth["records"][0]["execution_mode"] == "check_mode"
|
|
assert truth["records"][0]["returncode"] == 0
|
|
assert truth["records"][0]["dry_run_result"]["changed"] == 1
|
|
assert truth["summary"]["check_mode_total"] == 1
|
|
assert truth["summary"]["apply_total"] == 0
|
|
assert truth["summary"]["latest_catalog_id"] == "ansible:188-momo-backup-user"
|
|
assert truth["summary"]["latest_returncode"] == 0
|
|
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-momo-backup-user"
|
|
assert truth["candidate_catalog"]["candidates"][0]["auto_apply_enabled"] is True
|
|
assert (
|
|
truth["candidate_catalog"]["candidates"][0]["check_mode_playbook_path"]
|
|
== "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
)
|
|
|
|
|
|
def test_ansible_truth_marks_worker_apply_as_controlled_apply() -> None:
|
|
truth = build_ansible_truth(
|
|
[
|
|
{
|
|
"op_id": "apply-op-1",
|
|
"operation_type": "ansible_apply_executed",
|
|
"status": "success",
|
|
"actor": "ansible_controlled_apply_worker",
|
|
"input_catalog_id": "ansible:188-momo-backup-user",
|
|
"input_execution_mode": "controlled_apply",
|
|
"input_playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
|
|
"input_apply_enabled": "true",
|
|
"output_apply_executed": "true",
|
|
"output_returncode": "0",
|
|
"tags": ["ansible", "controlled_apply", "low", "ai_agent_auto_execution"],
|
|
},
|
|
{
|
|
"op_id": "check-op-1",
|
|
"operation_type": "ansible_check_mode_executed",
|
|
"status": "success",
|
|
"actor": "ansible_check_mode_worker",
|
|
"input_catalog_id": "ansible:188-momo-backup-user",
|
|
"input_execution_mode": "check_mode",
|
|
"input_playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
|
|
"dry_run_result": {"check_mode_executed": True, "apply_executed": False},
|
|
"tags": ["ansible", "check_mode", "controlled_apply_allowed"],
|
|
},
|
|
],
|
|
incident={"incident_id": "INC-MOMO", "alertname": "MomoPostgresBackupFailed"},
|
|
drift=None,
|
|
)
|
|
|
|
summary = truth["summary"]
|
|
assert summary["check_mode_total"] == 1
|
|
assert summary["apply_total"] == 1
|
|
assert summary["applied"] is True
|
|
assert summary["controlled_apply"] is True
|
|
assert summary["latest_actor"] == "ansible_controlled_apply_worker"
|
|
assert summary["latest_execution_mode"] == "controlled_apply"
|
|
assert summary["latest_apply_executed"] is True
|
|
|
|
|
|
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]["check_mode_playbook_path"] == (
|
|
"infra/ansible/playbooks/nginx-sync-readonly.yml"
|
|
)
|
|
assert truth["candidate_catalog"]["candidates"][0]["approval_required"] is False
|
|
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"]["execution_priority"] == 100
|
|
assert payload["input"]["executor_candidates"]
|
|
assert payload["output"]["decision_effect"] == "audit_only"
|
|
assert payload["dry_run_result"]["check_mode_executed"] is False
|
|
|
|
|
|
def test_ansible_decision_audit_payload_marks_repair_candidate_queue_claimable() -> None:
|
|
incident = SimpleNamespace(
|
|
incident_id="INC-NODE-188",
|
|
project_id="awoooi",
|
|
alert_category="infrastructure",
|
|
notification_type="TYPE-3",
|
|
severity=SimpleNamespace(value="P3"),
|
|
affected_services=["node-exporter-188"],
|
|
signals=[
|
|
SimpleNamespace(
|
|
alert_name="NodeExporterDown",
|
|
labels={"alertname": "NodeExporterDown", "instance": "node-exporter-188"},
|
|
annotations={},
|
|
)
|
|
],
|
|
)
|
|
|
|
payload = build_ansible_decision_audit_payload(
|
|
incident=incident,
|
|
proposal_data={
|
|
"source": "repair_candidate_controlled_queue",
|
|
"risk_level": "medium",
|
|
"execution_priority": 0,
|
|
"action": "systemctl restart node-exporter-188",
|
|
},
|
|
decision_path="repair_candidate_controlled_queue",
|
|
not_used_reason="repair candidate controlled queue ready",
|
|
)
|
|
|
|
assert payload is not None
|
|
assert payload["operation_type"] == "ansible_candidate_matched"
|
|
assert payload["status"] == "dry_run"
|
|
assert payload["output"]["decision_effect"] == "check_mode_queue_ready"
|
|
assert payload["output"]["next_required_step"] == (
|
|
"awooop_ansible_check_mode_worker_claims_candidate"
|
|
)
|
|
assert payload["input"]["executor_candidates"]
|
|
assert payload["dry_run_result"]["check_mode_executed"] is False
|
|
assert payload["input"]["project_id"] == "awoooi"
|
|
assert payload["input"]["execution_priority"] == 0
|
|
|
|
|
|
def test_ansible_canary_candidate_preserves_approval_and_controlled_apply() -> None:
|
|
approval_id = "00000000-0000-0000-0000-000000000044"
|
|
incident = SimpleNamespace(
|
|
incident_id="INC-CONTROLLED-CANARY",
|
|
project_id="awoooi",
|
|
alert_category="infrastructure",
|
|
notification_type="TYPE-3",
|
|
severity=SimpleNamespace(value="P2"),
|
|
affected_services=["awoooi-auto-repair-canary"],
|
|
signals=[
|
|
SimpleNamespace(
|
|
alert_name="AwoooPAutoRepairCanaryT16",
|
|
labels={
|
|
"alertname": "AwoooPAutoRepairCanaryT16",
|
|
"deployment": "awoooi-auto-repair-canary",
|
|
"namespace": "awoooi-prod",
|
|
},
|
|
annotations={},
|
|
)
|
|
],
|
|
)
|
|
|
|
payload = build_ansible_decision_audit_payload(
|
|
incident=incident,
|
|
proposal_data={
|
|
"source": "alert_webhook_controlled_router",
|
|
"risk_level": "medium",
|
|
"action": "awoooi-auto-repair-canary",
|
|
"approval_id": approval_id,
|
|
},
|
|
decision_path="repair_candidate_controlled_queue",
|
|
not_used_reason="queue bounded no-traffic canary",
|
|
)
|
|
|
|
assert payload is not None
|
|
assert payload["input"]["approval_id"] == approval_id
|
|
candidate = payload["input"]["executor_candidates"][0]
|
|
assert candidate["catalog_id"] == "ansible:awoooi-auto-repair-canary"
|
|
assert candidate["inventory_hosts"] == ["host_121"]
|
|
assert candidate["approval_required"] is False
|
|
claim = build_ansible_check_mode_claim_input(
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000045",
|
|
candidate_input=payload["input"],
|
|
)
|
|
assert claim["approval_id"] == approval_id
|
|
assert claim["controlled_apply_allowed"] is True
|
|
assert claim["apply_playbook_path"] == (
|
|
"infra/ansible/playbooks/awoooi-auto-repair-canary.yml"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_lifecycle_receipt_uses_same_run_and_project_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from src.repositories import alert_operation_log_repository as repo_module
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeRepository:
|
|
async def append(self, event_type: str, **kwargs):
|
|
captured["event_type"] = event_type
|
|
captured.update(kwargs)
|
|
return SimpleNamespace(id="receipt-1")
|
|
|
|
monkeypatch.setattr(
|
|
repo_module,
|
|
"get_alert_operation_log_repository",
|
|
lambda: FakeRepository(),
|
|
)
|
|
claim = AnsibleCheckModeClaim(
|
|
op_id="00000000-0000-0000-0000-000000000046",
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000047",
|
|
incident_id="INC-CONTROLLED-CANARY",
|
|
catalog_id="ansible:awoooi-auto-repair-canary",
|
|
playbook_path="infra/ansible/playbooks/awoooi-auto-repair-canary.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/awoooi-auto-repair-canary.yml",
|
|
inventory_hosts=("host_121",),
|
|
risk_level="low",
|
|
input_payload={
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000047",
|
|
"approval_id": "00000000-0000-0000-0000-000000000048",
|
|
},
|
|
)
|
|
|
|
written = await _append_alert_lifecycle_receipt(
|
|
claim,
|
|
"EXECUTION_STARTED",
|
|
apply_op_id="00000000-0000-0000-0000-000000000049",
|
|
success=True,
|
|
action_detail="controlled_apply_started",
|
|
project_id="awoooi",
|
|
)
|
|
|
|
assert written is True
|
|
assert captured["event_type"] == "EXECUTION_STARTED"
|
|
assert captured["project_id"] == "awoooi"
|
|
assert captured["approval_id"] == "00000000-0000-0000-0000-000000000048"
|
|
assert captured["context"]["automation_run_id"] == (
|
|
"00000000-0000-0000-0000-000000000047"
|
|
)
|
|
|
|
|
|
def test_controlled_apply_bridges_terminal_projection_and_telegram_receipt() -> None:
|
|
source = inspect.getsource(run_controlled_apply_for_claim)
|
|
|
|
assert '"EXECUTION_STARTED"' in source
|
|
assert '"EXECUTION_COMPLETED"' in source
|
|
assert '"TELEGRAM_RESULT_SENT"' in source
|
|
assert "_finalize_controlled_approval_projection" in source
|
|
|
|
|
|
def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags(
|
|
monkeypatch,
|
|
) -> None:
|
|
monkeypatch.setenv("AWOOOI_BUILD_COMMIT_SHA", "source-sha-123")
|
|
incident = SimpleNamespace(
|
|
incident_id="INC-MOMO",
|
|
project_id="awoooi",
|
|
alert_category="database",
|
|
notification_type="TYPE-3",
|
|
severity=SimpleNamespace(value="P3"),
|
|
affected_services=["momo"],
|
|
signals=[
|
|
SimpleNamespace(
|
|
alert_name="MomoPostgresBackupFailed",
|
|
labels={"alertname": "MomoPostgresBackupFailed", "instance": "188"},
|
|
annotations={},
|
|
)
|
|
],
|
|
)
|
|
|
|
payload = build_ansible_decision_audit_payload(
|
|
incident=incident,
|
|
proposal_data={"source": "expert_system", "risk_level": "low"},
|
|
decision_path="manual_approval",
|
|
not_used_reason="candidate audit",
|
|
)
|
|
|
|
candidate = payload["input"]["executor_candidates"][0]
|
|
assert candidate["catalog_id"] == "ansible:188-momo-backup-user"
|
|
assert candidate["supports_check_mode"] is True
|
|
assert candidate["playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
assert candidate["check_mode_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
assert candidate["auto_apply_enabled"] is True
|
|
assert candidate["approval_required"] is False
|
|
assert candidate["risk_level"] == "low"
|
|
assert payload["input"]["router_source_sha"] == "source-sha-123"
|
|
|
|
|
|
def test_ansible_check_mode_claim_input_authorizes_controlled_apply() -> None:
|
|
candidate_input = {
|
|
"incident_id": "INC-MOMO",
|
|
"automation_run_id": "untrusted-upstream-run-id",
|
|
"executor": "ansible",
|
|
"decision_path": "repair_candidate_controlled_queue",
|
|
"idempotency_key": "ansible-candidate:awoooi:INC-MOMO",
|
|
"router_source_sha": "source-sha-123",
|
|
"target_selector": {"catalog_ids": ["ansible:188-ai-web"]},
|
|
"source_truth_diff": {"required_before_apply": True},
|
|
"executor_candidates": [
|
|
{
|
|
"catalog_id": "ansible:188-ai-web",
|
|
"playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
|
|
"inventory_hosts": ["host_188"],
|
|
"risk_level": "medium",
|
|
}
|
|
],
|
|
}
|
|
|
|
claim = build_ansible_check_mode_claim_input(
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000001",
|
|
candidate_input=candidate_input,
|
|
)
|
|
|
|
assert claim["execution_mode"] == "check_mode"
|
|
assert claim["automation_run_id"] == "00000000-0000-0000-0000-000000000001"
|
|
assert claim["check_mode"] is True
|
|
assert claim["diff"] is True
|
|
assert claim["apply_enabled"] is True
|
|
assert claim["transport_profile"]
|
|
assert claim["approval_required_before_apply"] is False
|
|
assert claim["controlled_apply_allowed"] is True
|
|
assert claim["controlled_apply_blocker"] is None
|
|
assert claim["single_writer_executor"] == "awoooi-ansible-executor-broker"
|
|
assert claim["candidate_idempotency_key"] == (
|
|
"ansible-candidate:awoooi:INC-MOMO"
|
|
)
|
|
assert claim["apply_idempotency_key"] == (
|
|
"ansible-apply:00000000-0000-0000-0000-000000000001:"
|
|
"ansible:188-ai-web"
|
|
)
|
|
assert claim["router_source_sha"] == "source-sha-123"
|
|
assert claim["target_selector"] == {
|
|
"catalog_ids": ["ansible:188-ai-web"]
|
|
}
|
|
assert claim["source_truth_diff"]["required_before_apply"] is True
|
|
assert claim["catalog_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
|
|
assert claim["apply_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
|
|
assert claim["source_candidate_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
|
|
assert claim["check_mode_playbook_path"] == "infra/ansible/playbooks/188-ai-web-readonly.yml"
|
|
assert claim["playbook_path"] == "infra/ansible/playbooks/188-ai-web-readonly.yml"
|
|
|
|
|
|
def test_ansible_check_mode_claim_input_accepts_momo_backup_user_catalog() -> None:
|
|
candidate_input = {
|
|
"incident_id": "INC-MOMO",
|
|
"executor": "ansible",
|
|
"executor_candidates": [
|
|
{
|
|
"catalog_id": "ansible:188-momo-backup-user",
|
|
"playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
|
|
"inventory_hosts": ["host_188"],
|
|
"risk_level": "low",
|
|
}
|
|
],
|
|
}
|
|
|
|
claim = build_ansible_check_mode_claim_input(
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000003",
|
|
candidate_input=candidate_input,
|
|
)
|
|
|
|
assert claim["execution_mode"] == "check_mode"
|
|
assert claim["automation_run_id"] == "00000000-0000-0000-0000-000000000003"
|
|
assert claim["apply_enabled"] is True
|
|
assert claim["controlled_apply_allowed"] is True
|
|
assert claim["approval_required_before_apply"] is False
|
|
assert claim["catalog_id"] == "ansible:188-momo-backup-user"
|
|
assert claim["risk_level"] == "low"
|
|
assert claim["catalog_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
assert claim["apply_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
assert claim["source_candidate_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
assert claim["check_mode_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
assert claim["playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
|
|
|
|
def test_ansible_check_mode_claim_rejects_non_check_mode_catalog() -> None:
|
|
candidate_input = {
|
|
"incident_id": "INC-SSH",
|
|
"executor": "ansible",
|
|
"executor_candidates": [
|
|
{
|
|
"catalog_id": "ansible:restore-password-auth",
|
|
"playbook_path": "infra/ansible/playbooks/restore-password-auth.yml",
|
|
"inventory_hosts": ["host_188"],
|
|
"risk_level": "high",
|
|
}
|
|
],
|
|
}
|
|
|
|
try:
|
|
build_ansible_check_mode_claim_input(
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000002",
|
|
candidate_input=candidate_input,
|
|
)
|
|
except ValueError as exc:
|
|
assert str(exc) == "no_safe_check_mode_candidate"
|
|
else:
|
|
raise AssertionError("non-check-mode catalog should be rejected")
|
|
|
|
|
|
def test_ansible_check_mode_command_uses_check_diff_and_selected_ssh_transport(tmp_path: Path) -> None:
|
|
playbook_root = tmp_path / "infra" / "ansible"
|
|
playbook_dir = playbook_root / "playbooks"
|
|
inventory_dir = playbook_root / "inventory"
|
|
playbook_dir.mkdir(parents=True)
|
|
inventory_dir.mkdir(parents=True)
|
|
(playbook_dir / "188-ai-web.yml").write_text("---\n- hosts: host_188\n tasks: []\n")
|
|
(inventory_dir / "hosts.yml").write_text("all: {}\n")
|
|
repair_key = tmp_path / "id_ed25519"
|
|
known_hosts = tmp_path / "known_hosts"
|
|
repair_key.write_text("key")
|
|
known_hosts.write_text("host key")
|
|
|
|
spec = build_ansible_check_mode_command(
|
|
playbook_path="infra/ansible/playbooks/188-ai-web.yml",
|
|
inventory_hosts=("host_188",),
|
|
playbook_root=playbook_root,
|
|
check_mode_ssh_key_path=repair_key,
|
|
check_mode_known_hosts_path=known_hosts,
|
|
)
|
|
|
|
assert "--check" in spec.command
|
|
assert "--diff" in spec.command
|
|
assert "--limit" in spec.command
|
|
assert "host_188" in spec.command
|
|
assert "ansible_ssh_private_key_file" in spec.command[-1]
|
|
assert str(repair_key) in spec.command[-1]
|
|
assert str(known_hosts) in spec.command[-1]
|
|
assert "apply" not in " ".join(spec.command)
|
|
assert str((playbook_root / "roles").resolve()) in spec.env[
|
|
"ANSIBLE_ROLES_PATH"
|
|
].split(os.pathsep)
|
|
|
|
|
|
def test_ansible_apply_command_uses_controlled_apply_without_check(tmp_path: Path) -> None:
|
|
playbook_root = tmp_path / "infra" / "ansible"
|
|
playbook_dir = playbook_root / "playbooks"
|
|
inventory_dir = playbook_root / "inventory"
|
|
playbook_dir.mkdir(parents=True)
|
|
inventory_dir.mkdir(parents=True)
|
|
(playbook_dir / "188-ai-web.yml").write_text("---\n- hosts: host_188\n tasks: []\n")
|
|
(inventory_dir / "hosts.yml").write_text("all: {}\n")
|
|
repair_key = tmp_path / "id_ed25519"
|
|
known_hosts = tmp_path / "known_hosts"
|
|
repair_key.write_text("key")
|
|
known_hosts.write_text("host key")
|
|
|
|
spec = build_ansible_apply_command(
|
|
playbook_path="infra/ansible/playbooks/188-ai-web.yml",
|
|
inventory_hosts=("host_188",),
|
|
playbook_root=playbook_root,
|
|
check_mode_ssh_key_path=repair_key,
|
|
check_mode_known_hosts_path=known_hosts,
|
|
)
|
|
|
|
assert "--check" not in spec.command
|
|
assert "--diff" in spec.command
|
|
assert "--limit" in spec.command
|
|
assert "host_188" in spec.command
|
|
assert "ansible_ssh_private_key_file" in spec.command[-1]
|
|
assert str(repair_key) in spec.command[-1]
|
|
assert str(known_hosts) in spec.command[-1]
|
|
assert str((playbook_root / "roles").resolve()) in spec.env[
|
|
"ANSIBLE_ROLES_PATH"
|
|
].split(os.pathsep)
|
|
|
|
|
|
def test_ansible_controlled_apply_builds_auto_repair_receipt() -> None:
|
|
claim = AnsibleCheckModeClaim(
|
|
op_id="check-op-1",
|
|
source_candidate_op_id="candidate-op-1",
|
|
incident_id="INC-20260627-NODE110",
|
|
catalog_id="ansible:110-devops",
|
|
playbook_path="infra/ansible/playbooks/110-devops.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/110-devops.yml",
|
|
inventory_hosts=("host_110",),
|
|
risk_level="medium",
|
|
input_payload={"controlled_apply_allowed": True},
|
|
)
|
|
result = AnsibleRunResult(
|
|
returncode=0,
|
|
stdout="ok",
|
|
stderr="",
|
|
duration_ms=1234,
|
|
post_verifier_passed=True,
|
|
)
|
|
|
|
receipt = _build_auto_repair_execution_receipt(
|
|
claim,
|
|
result,
|
|
apply_op_id="apply-op-1",
|
|
)
|
|
|
|
assert receipt["incident_id"] == "INC-20260627-NODE110"
|
|
assert receipt["playbook_id"] == "ansible:110-devops"
|
|
assert receipt["success"] is True
|
|
assert receipt["triggered_by"] == "ansible_controlled_apply"
|
|
assert "automation_run_id:candidate-op-1" in receipt["executed_steps"]
|
|
assert "apply:apply-op-1" in receipt["executed_steps"]
|
|
assert receipt["risk_level"] == "medium"
|
|
assert receipt["execution_time_ms"] == 1234
|
|
|
|
|
|
def test_ansible_auto_repair_receipt_never_trusts_executor_only() -> None:
|
|
claim = AnsibleCheckModeClaim(
|
|
op_id="check-op-1",
|
|
source_candidate_op_id="candidate-op-1",
|
|
incident_id="INC-20260627-NODE110",
|
|
catalog_id="ansible:110-devops",
|
|
playbook_path="infra/ansible/playbooks/110-devops.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/110-devops.yml",
|
|
inventory_hosts=("host_110",),
|
|
risk_level="medium",
|
|
input_payload={"controlled_apply_allowed": True},
|
|
)
|
|
executor_only = AnsibleRunResult(
|
|
returncode=0,
|
|
stdout="ok",
|
|
stderr="",
|
|
duration_ms=1234,
|
|
)
|
|
|
|
receipt = _build_auto_repair_execution_receipt(
|
|
claim,
|
|
executor_only,
|
|
apply_op_id="apply-op-1",
|
|
)
|
|
|
|
assert receipt["success"] is False
|
|
assert receipt["error_message"] == "independent_post_verifier_failed"
|
|
assert "post_verifier_passed:false" in receipt["executed_steps"]
|
|
|
|
|
|
def test_ansible_apply_operation_row_can_backfill_auto_repair_receipt() -> None:
|
|
reconstructed = _claim_from_apply_operation_row({
|
|
"op_id": "apply-op-1",
|
|
"parent_op_id": "check-op-1",
|
|
"incident_id": "INC-20260627-NODE110",
|
|
"status": "success",
|
|
"input": {
|
|
"incident_id": "INC-20260627-NODE110",
|
|
"catalog_id": "ansible:110-devops",
|
|
"source_candidate_op_id": "candidate-op-1",
|
|
"check_mode_op_id": "check-op-1",
|
|
"playbook_path": "infra/ansible/playbooks/110-devops.yml",
|
|
"apply_playbook_path": "infra/ansible/playbooks/110-devops.yml",
|
|
"inventory_hosts": ["host_110"],
|
|
"risk_level": "medium",
|
|
},
|
|
"output": {"returncode": 0, "stdout_tail": "ok"},
|
|
"dry_run_result": {"apply_executed": True},
|
|
"duration_ms": 456,
|
|
})
|
|
|
|
assert reconstructed is not None
|
|
claim, result = reconstructed
|
|
assert claim.incident_id == "INC-20260627-NODE110"
|
|
assert claim.catalog_id == "ansible:110-devops"
|
|
assert claim.apply_playbook_path == "infra/ansible/playbooks/110-devops.yml"
|
|
assert result.returncode == 0
|
|
assert result.duration_ms == 456
|
|
|
|
|
|
def test_ansible_apply_operation_row_reconstructs_from_input_without_physical_columns() -> None:
|
|
reconstructed = _claim_from_apply_operation_row({
|
|
"op_id": "apply-op-2",
|
|
"parent_op_id": "check-op-2",
|
|
"incident_id": "INC-20260629-231F8E",
|
|
"status": "success",
|
|
"input": {
|
|
"incident_id": "INC-20260629-231F8E",
|
|
"catalog_id": "ansible:188-momo-backup-user",
|
|
"source_candidate_op_id": "candidate-op-2",
|
|
"check_mode_op_id": "check-op-2",
|
|
"playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
|
|
"risk_level": "low",
|
|
},
|
|
"output": {"returncode": 0, "stdout_tail": "ok"},
|
|
"dry_run_result": {"apply_executed": True},
|
|
"duration_ms": 7727,
|
|
})
|
|
|
|
assert reconstructed is not None
|
|
claim, result = reconstructed
|
|
assert claim.catalog_id == "ansible:188-momo-backup-user"
|
|
assert claim.inventory_hosts == ("host_188",)
|
|
assert claim.apply_playbook_path == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
assert claim.risk_level == "low"
|
|
assert result.returncode == 0
|
|
|
|
|
|
def test_stale_check_mode_row_is_revalidated_with_canonical_run_id() -> None:
|
|
source_candidate_op_id = "00000000-0000-0000-0000-000000000091"
|
|
claim = _claim_from_stale_check_mode_row({
|
|
"op_id": "00000000-0000-0000-0000-000000000092",
|
|
"parent_op_id": source_candidate_op_id,
|
|
"incident_id": "INC-20260709-46478E",
|
|
"input": {
|
|
"automation_run_id": "untrusted-old-run-id",
|
|
"incident_id": "INC-20260709-46478E",
|
|
"source_candidate_op_id": source_candidate_op_id,
|
|
"catalog_id": "ansible:188-momo-backup-user",
|
|
"catalog_playbook_path": (
|
|
"infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
),
|
|
"inventory_hosts": ["host_188"],
|
|
"risk_level": "low",
|
|
},
|
|
})
|
|
|
|
assert claim is not None
|
|
assert claim.op_id == "00000000-0000-0000-0000-000000000092"
|
|
assert claim.source_candidate_op_id == source_candidate_op_id
|
|
assert claim.input_payload["automation_run_id"] == source_candidate_op_id
|
|
assert claim.input_payload["controlled_apply_allowed"] is True
|
|
assert claim.playbook_path == "infra/ansible/playbooks/188-momo-backup-user.yml"
|
|
|
|
|
|
def test_failed_nginx_check_mode_revalidates_to_readonly_catalog_path() -> None:
|
|
source_candidate_op_id = "00000000-0000-0000-0000-000000000093"
|
|
claim = _claim_from_stale_check_mode_row({
|
|
"op_id": "00000000-0000-0000-0000-000000000094",
|
|
"parent_op_id": source_candidate_op_id,
|
|
"incident_id": "INC-20260710-F87642",
|
|
"input": {
|
|
"incident_id": "INC-20260710-F87642",
|
|
"source_candidate_op_id": source_candidate_op_id,
|
|
"catalog_id": "ansible:nginx-sync",
|
|
"catalog_playbook_path": "infra/ansible/playbooks/nginx-sync.yml",
|
|
"check_mode_playbook_path": "infra/ansible/playbooks/nginx-sync.yml",
|
|
"inventory_hosts": ["host_110", "host_188"],
|
|
"risk_level": "high",
|
|
},
|
|
})
|
|
|
|
assert claim is not None
|
|
assert claim.playbook_path == "infra/ansible/playbooks/nginx-sync-readonly.yml"
|
|
assert claim.apply_playbook_path == "infra/ansible/playbooks/nginx-sync.yml"
|
|
|
|
|
|
def test_stale_check_mode_reclaim_uses_lease_lock_and_current_policy() -> None:
|
|
source = inspect.getsource(claim_stale_pending_check_modes)
|
|
run_source = inspect.getsource(run_pending_check_modes_once)
|
|
|
|
assert "FOR UPDATE SKIP LOCKED" in source
|
|
assert "reclaimed_at" in source
|
|
assert "ORDER BY check_mode.created_at DESC" in source
|
|
assert "stale_pending_reclaim_rejected_by_current_policy" in source
|
|
assert "_claim_from_stale_check_mode_row" in source
|
|
assert "claim_stale_pending_check_modes" in run_source
|
|
assert "effective_timeout_seconds + 120" in run_source
|
|
|
|
|
|
def test_ansible_execution_broker_issues_expires_and_revokes_bounded_capabilities() -> None:
|
|
issue_source = inspect.getsource(_issue_ansible_execution_capability)
|
|
expire_source = inspect.getsource(_expire_stale_ansible_execution_capabilities)
|
|
revoke_source = inspect.getsource(_revoke_ansible_execution_capability)
|
|
resolver_source = inspect.getsource(_resolve_execution_capability_operation_type)
|
|
run_source = inspect.getsource(run_pending_check_modes_once)
|
|
|
|
assert "execution_capability_catalog_not_allowlisted" in issue_source
|
|
assert "execution_capability_inventory_outside_allowlist" in issue_source
|
|
assert "ansible_executor_capability_issued" in issue_source
|
|
assert "execution_capability_active_lease_exists" in issue_source
|
|
assert "WHERE NOT EXISTS" in issue_source
|
|
assert "raw_credential_exposed_to_requester" in issue_source
|
|
assert "expires_at" in issue_source
|
|
assert "ansible_executor_capability_expired" in expire_source
|
|
assert "FOR UPDATE SKIP LOCKED" in expire_source
|
|
assert "ansible_executor_capability_revoked" in revoke_source
|
|
assert "further_execution_allowed" in revoke_source
|
|
assert "automation_operation_log_type_valid" in resolver_source
|
|
assert "semantic_operation_type" in resolver_source
|
|
assert _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE == "remediation_executed"
|
|
assert "_issue_ansible_execution_capability" in run_source
|
|
assert "_revoke_ansible_execution_capability" in run_source
|
|
assert "_expire_stale_ansible_execution_capabilities" in run_source
|
|
|
|
|
|
def test_ansible_execution_capability_enforces_scope_mode_and_expiry() -> None:
|
|
now = datetime(2026, 7, 11, 1, 0, tzinfo=UTC)
|
|
run_id = "00000000-0000-0000-0000-000000000201"
|
|
check_op_id = "00000000-0000-0000-0000-000000000202"
|
|
scope = {
|
|
"catalog_id": "ansible:110-devops",
|
|
"playbook_path": "infra/ansible/playbooks/110-devops.yml",
|
|
"apply_playbook_path": "infra/ansible/playbooks/110-devops.yml",
|
|
"inventory_hosts": ["host_110"],
|
|
"risk_level": "medium",
|
|
"allowed_execution_modes": ["check_mode", "controlled_apply"],
|
|
}
|
|
|
|
def _claim(*, capability_scope: dict, expires_at: datetime) -> AnsibleCheckModeClaim:
|
|
return AnsibleCheckModeClaim(
|
|
op_id=check_op_id,
|
|
source_candidate_op_id=run_id,
|
|
incident_id="INC-CAPABILITY",
|
|
catalog_id="ansible:110-devops",
|
|
playbook_path="infra/ansible/playbooks/110-devops.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/110-devops.yml",
|
|
inventory_hosts=("host_110",),
|
|
risk_level="medium",
|
|
input_payload={
|
|
"automation_run_id": run_id,
|
|
"controlled_apply_allowed": True,
|
|
"execution_capability": {
|
|
"capability_op_id": (
|
|
"00000000-0000-0000-0000-000000000203"
|
|
),
|
|
"automation_run_id": run_id,
|
|
"check_mode_op_id": check_op_id,
|
|
"issued_at": (now - timedelta(seconds=1)).isoformat(),
|
|
"expires_at": expires_at.isoformat(),
|
|
"scope": capability_scope,
|
|
},
|
|
},
|
|
)
|
|
|
|
claim = _claim(
|
|
capability_scope=scope,
|
|
expires_at=now + timedelta(seconds=100),
|
|
)
|
|
assert _execution_capability_timeout_seconds(
|
|
claim,
|
|
execution_mode="controlled_apply",
|
|
requested_timeout_seconds=180,
|
|
now=now,
|
|
) == 95
|
|
|
|
with pytest.raises(ValueError, match="execution_capability_expired"):
|
|
_execution_capability_timeout_seconds(
|
|
_claim(capability_scope=scope, expires_at=now),
|
|
execution_mode="check_mode",
|
|
requested_timeout_seconds=180,
|
|
now=now,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="execution_capability_mode_not_allowed"):
|
|
_execution_capability_timeout_seconds(
|
|
_claim(
|
|
capability_scope={
|
|
**scope,
|
|
"allowed_execution_modes": ["check_mode"],
|
|
},
|
|
expires_at=now + timedelta(seconds=100),
|
|
),
|
|
execution_mode="controlled_apply",
|
|
requested_timeout_seconds=180,
|
|
now=now,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="execution_capability_scope_mismatch"):
|
|
_execution_capability_timeout_seconds(
|
|
_claim(
|
|
capability_scope={**scope, "inventory_hosts": ["host_188"]},
|
|
expires_at=now + timedelta(seconds=100),
|
|
),
|
|
execution_mode="check_mode",
|
|
requested_timeout_seconds=180,
|
|
now=now,
|
|
)
|
|
|
|
|
|
def test_failed_check_mode_catalog_drift_replays_once() -> None:
|
|
source = inspect.getsource(claim_catalog_drift_failed_check_modes)
|
|
run_source = inspect.getsource(run_pending_check_modes_once)
|
|
|
|
assert "replay_of_check_mode_op_id" in source
|
|
assert "catalog_drift_replay" in source
|
|
assert "FOR UPDATE SKIP LOCKED" in source
|
|
assert "claim_catalog_drift_failed_check_modes" in run_source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from src.services import awooop_ansible_check_mode_service as service
|
|
|
|
fresh_claim_attempted = False
|
|
|
|
async def no_expired_capabilities(**_kwargs):
|
|
return 0
|
|
|
|
async def no_transport_blockers(**_kwargs):
|
|
return []
|
|
|
|
async def no_stale_claims(**_kwargs):
|
|
return []
|
|
|
|
async def no_failed_apply_retry(**_kwargs):
|
|
return {
|
|
"scanned": 0,
|
|
"replayed": 0,
|
|
"check_mode_passed": 0,
|
|
"check_mode_failed": 0,
|
|
"runtime_stage_receipt_written": 0,
|
|
"blockers": [],
|
|
"error": None,
|
|
}
|
|
|
|
async def failed_catalog_replay(**_kwargs):
|
|
raise RuntimeError("catalog replay query canceled")
|
|
|
|
async def fresh_claims_continue(**_kwargs):
|
|
nonlocal fresh_claim_attempted
|
|
fresh_claim_attempted = True
|
|
return []
|
|
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_expire_stale_ansible_execution_capabilities",
|
|
no_expired_capabilities,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"backfill_missing_auto_repair_execution_receipts_once",
|
|
AsyncMock(return_value={"scanned": 0, "written": 0, "error": None}),
|
|
)
|
|
monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
|
|
monkeypatch.setattr(
|
|
service,
|
|
"recent_ansible_transport_blockers",
|
|
no_transport_blockers,
|
|
)
|
|
monkeypatch.setattr(service, "claim_stale_pending_check_modes", no_stale_claims)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"run_failed_apply_check_mode_replay_once",
|
|
no_failed_apply_retry,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_catalog_drift_failed_check_modes",
|
|
failed_catalog_replay,
|
|
)
|
|
monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claims_continue)
|
|
|
|
result = await service.run_pending_check_modes_once(limit=1)
|
|
|
|
assert fresh_claim_attempted is True
|
|
assert result["claimed"] == 0
|
|
assert result["catalog_replayed"] == 0
|
|
assert result["catalog_replay_error"] == "RuntimeError"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_claims(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from src.services import awooop_ansible_check_mode_service as service
|
|
|
|
claim_attempted = False
|
|
|
|
async def no_expired_capabilities(**_kwargs):
|
|
return 0
|
|
|
|
async def no_transport_blockers(**_kwargs):
|
|
return []
|
|
|
|
async def replay_open_retry(**_kwargs):
|
|
return {
|
|
"scanned": 1,
|
|
"replayed": 1,
|
|
"check_mode_passed": 1,
|
|
"check_mode_failed": 0,
|
|
"runtime_stage_receipt_written": 1,
|
|
"blockers": [],
|
|
"error": None,
|
|
}
|
|
|
|
async def candidate_claim_must_not_run(**_kwargs):
|
|
nonlocal claim_attempted
|
|
claim_attempted = True
|
|
return []
|
|
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_expire_stale_ansible_execution_capabilities",
|
|
no_expired_capabilities,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"backfill_missing_auto_repair_execution_receipts_once",
|
|
AsyncMock(return_value={"scanned": 0, "written": 0, "error": None}),
|
|
)
|
|
monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
|
|
monkeypatch.setattr(
|
|
service,
|
|
"recent_ansible_transport_blockers",
|
|
no_transport_blockers,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"run_failed_apply_check_mode_replay_once",
|
|
replay_open_retry,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_stale_pending_check_modes",
|
|
candidate_claim_must_not_run,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_catalog_drift_failed_check_modes",
|
|
candidate_claim_must_not_run,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"claim_pending_check_modes",
|
|
candidate_claim_must_not_run,
|
|
)
|
|
|
|
result = await service.run_pending_check_modes_once(limit=1)
|
|
|
|
assert claim_attempted is False
|
|
assert result["claimed"] == 0
|
|
assert result["failed_apply_retry_scanned"] == 1
|
|
assert result["failed_apply_retry_replayed"] == 1
|
|
assert result["failed_apply_retry_check_mode_passed"] == 1
|
|
assert result["failed_apply_retry_stage_receipt_written"] == 1
|
|
assert result["failed_apply_retry_priority_tick"] is True
|
|
|
|
|
|
def test_check_mode_worker_error_backoff_is_short_and_bounded() -> None:
|
|
from src.jobs import awooop_ansible_check_mode_job as job
|
|
|
|
delay = job._error_backoff_seconds()
|
|
|
|
assert job._ERROR_BACKOFF_MIN_SECONDS <= delay <= job._ERROR_BACKOFF_MAX_SECONDS
|
|
assert job._ERROR_BACKOFF_MAX_SECONDS < 60
|
|
|
|
|
|
def test_ansible_apply_receipt_backfill_queries_existing_apply_rows() -> None:
|
|
source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once)
|
|
|
|
assert "operation_type = 'ansible_apply_executed'" in source
|
|
assert "auto_repair_executions existing" in source
|
|
assert "executed_steps::text LIKE" in source
|
|
assert "apply.catalog_id" not in source
|
|
assert "apply.playbook_path" not in source
|
|
assert "apply.risk_level" not in source
|
|
|
|
|
|
def test_ansible_auto_repair_receipt_insert_casts_asyncpg_parameters() -> None:
|
|
source = inspect.getsource(_record_auto_repair_execution_receipt)
|
|
|
|
assert "CAST(:incident_id AS varchar(30))" in source
|
|
assert "CAST(:playbook_id AS varchar(36))" in source
|
|
assert "CAST(:triggered_by AS varchar(50))" in source
|
|
assert "CAST(:apply_op_id_needle AS text)" in source
|
|
|
|
|
|
def test_ansible_apply_receipt_backfill_includes_verifier_and_km_gaps() -> None:
|
|
source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once)
|
|
|
|
assert "FROM incident_evidence evidence" in source
|
|
assert "evidence.post_execution_state ->> 'apply_op_id' = apply.op_id::text" in source
|
|
assert "FROM knowledge_entries km" in source
|
|
assert "km.path_type = 'ansible_apply_receipt:' || left(apply.op_id::text, 8)" in source
|
|
assert "ansible_learning_writeback_recorded" in source
|
|
assert "_record_post_apply_verifier_and_learning" in source
|
|
assert "runtime_stage_receipts" in source
|
|
assert "_record_runtime_stage_receipts" in source
|
|
assert "playbook_trust" in source
|
|
assert "rag_writeback" in source
|
|
assert "INTERVAL '5 minutes'" in source
|
|
|
|
|
|
def test_ansible_learning_writeback_uses_canonical_durable_service() -> None:
|
|
source = inspect.getsource(_record_learning_writeback_receipt)
|
|
durable_source = inspect.getsource(record_ansible_playbook_trust_writeback)
|
|
receipt_source = inspect.getsource(_playbook_trust_readback)
|
|
|
|
assert "record_ansible_playbook_trust_writeback" in source
|
|
assert "resolve_ansible_playbook_identity" in durable_source
|
|
assert "FOR UPDATE" in durable_source
|
|
assert "await db.refresh(playbook)" in durable_source
|
|
assert '"operation_receipt_readback_verified": True' in receipt_source
|
|
assert "operation_readback" in durable_source
|
|
assert 'operation_output.get("learning_recorded") is True' in durable_source
|
|
assert 'operation_output.get("trust_updated") is True' in durable_source
|
|
assert '"durable_write_acknowledged": True' in receipt_source
|
|
|
|
|
|
def test_ansible_catalog_ids_map_to_stable_playbook_primary_keys() -> None:
|
|
expected = {
|
|
"ansible:110-devops": "PB-ANSIBLE-110-DEVOPS",
|
|
"ansible:188-momo-backup-user": "PB-ANSIBLE-188-MOMO-BACKUP-USER",
|
|
"ansible:188-ai-web": "PB-ANSIBLE-188-AI-WEB",
|
|
"ansible:nginx-sync": "PB-ANSIBLE-NGINX-SYNC",
|
|
"ansible:restore-password-auth": (
|
|
"PB-ANSIBLE-RESTORE-PASSWORD-AUTH"
|
|
),
|
|
}
|
|
|
|
assert {
|
|
catalog_id: canonical_ansible_playbook_id(catalog_id)
|
|
for catalog_id in expected
|
|
} == expected
|
|
assert all(len(playbook_id) <= 32 for playbook_id in expected.values())
|
|
for catalog_id, playbook_id in expected.items():
|
|
identity = resolve_ansible_playbook_identity(catalog_id)
|
|
assert identity is not None
|
|
assert identity["schema_version"] == "ansible_playbook_identity_v1"
|
|
assert identity["canonical_playbook_id"] == playbook_id
|
|
assert len(identity["identity_fingerprint"]) == 64
|
|
assert resolve_ansible_playbook_identity("ansible:not-allowlisted") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ansible_learning_writeback_rejects_catalog_path_mismatch() -> None:
|
|
receipt = await record_ansible_playbook_trust_writeback(
|
|
project_id="awoooi",
|
|
automation_run_id="run-path-mismatch",
|
|
incident_id="INC-PATH-MISMATCH",
|
|
incident_db_id=None,
|
|
catalog_id="ansible:188-momo-backup-user",
|
|
playbook_path="infra/ansible/playbooks/not-allowlisted.yml",
|
|
apply_op_id="11111111-1111-4111-8111-111111111111",
|
|
verification_result="success",
|
|
)
|
|
|
|
assert receipt is None
|
|
|
|
|
|
def test_ansible_rag_writeback_requires_embedding_readback() -> None:
|
|
source = inspect.getsource(ensure_ansible_rag_writeback)
|
|
fallback_source = inspect.getsource(_ensure_rag_chunk_fallback)
|
|
|
|
assert "ensure_entry_embedding" in source
|
|
assert "embedding IS NOT NULL" in source
|
|
assert "durable_write_acknowledged" in source
|
|
assert "_ensure_rag_chunk_fallback" in source
|
|
assert "index_document" in fallback_source
|
|
assert "count_chunks_by_source_id" in fallback_source
|
|
assert "rag_chunk_count" in fallback_source
|
|
assert "chunk_index_verified" in fallback_source
|
|
assert "raw_log_payload_stored" in fallback_source
|
|
assert "secret_value_stored" in fallback_source
|
|
assert "ansible_rag_km_embedding_copy_failed" in fallback_source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ansible_rag_fallback_keeps_verified_chunks_when_km_copy_fails(
|
|
monkeypatch,
|
|
) -> None:
|
|
from src.repositories import rag_chunk_repository
|
|
from src.services import awooop_ansible_learning_writeback as learning_writeback
|
|
|
|
requested_projects: list[str | None] = []
|
|
|
|
async def count_chunks(
|
|
_source_id: str,
|
|
project_id: str | None = None,
|
|
) -> int:
|
|
requested_projects.append(project_id)
|
|
return 2
|
|
|
|
class FailingCopyDb:
|
|
async def execute(self, *_args, **_kwargs):
|
|
raise RuntimeError("copy unavailable")
|
|
|
|
class FailingCopyContext:
|
|
async def __aenter__(self):
|
|
return FailingCopyDb()
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
monkeypatch.setattr(
|
|
rag_chunk_repository,
|
|
"count_chunks_by_source_id",
|
|
count_chunks,
|
|
)
|
|
monkeypatch.setattr(
|
|
learning_writeback,
|
|
"get_db_context",
|
|
lambda _project_id: FailingCopyContext(),
|
|
)
|
|
|
|
receipt = await _ensure_rag_chunk_fallback(
|
|
project_id="awoooi",
|
|
incident_id="INC-RAG-FALLBACK",
|
|
apply_op_id="apply-op-1",
|
|
entry_id="km-entry-1",
|
|
title="Verified repair",
|
|
content="Durable public-safe learning summary",
|
|
)
|
|
|
|
assert requested_projects == ["awoooi"]
|
|
assert receipt is not None
|
|
assert receipt["embedding_persisted"] is False
|
|
assert receipt["source_table"] == "rag_chunks"
|
|
assert receipt["rag_chunk_count"] == 2
|
|
assert receipt["chunk_index_verified"] is True
|
|
assert receipt["durable_write_acknowledged"] is True
|
|
|
|
|
|
def test_ansible_post_apply_km_writeback_is_idempotent_for_learning_backfill() -> None:
|
|
from src.services.awooop_ansible_check_mode_service import (
|
|
_record_post_apply_verifier_and_learning,
|
|
)
|
|
|
|
source = inspect.getsource(_record_post_apply_verifier_and_learning)
|
|
|
|
assert "SELECT id" in source
|
|
assert "FROM knowledge_entries" in source
|
|
assert "path_type = :path_type" in source
|
|
assert "KnowledgeDBRepository" in source
|
|
assert "stage_id=\"rag_writeback\"" in source
|
|
assert "stage_id=\"playbook_trust\"" in source
|
|
assert "stage_id=\"km_playbook_writeback\"" in source
|
|
assert "repository_readback_verified" in source
|
|
assert "km_row_version" in source
|
|
assert '"knowledge_entries:"' in source
|
|
assert '"rag_chunks:"' in source
|
|
|
|
|
|
def test_ansible_learning_writeback_operation_type_has_schema_migration() -> None:
|
|
migrations_dir = Path(__file__).resolve().parents[1] / "migrations"
|
|
migration = Path(
|
|
migrations_dir / "adr090e_ansible_learning_writeback_operation_type.sql"
|
|
).read_text()
|
|
down = Path(
|
|
migrations_dir / "adr090e_ansible_learning_writeback_operation_type_down.sql"
|
|
).read_text()
|
|
|
|
assert "ansible_learning_writeback_recorded" in migration
|
|
assert "automation_operation_log_type_valid" in migration
|
|
assert "DROP CONSTRAINT IF EXISTS automation_operation_log_type_valid" in migration
|
|
assert "cannot remove ansible_learning_writeback_recorded" in down
|
|
|
|
|
|
def test_log_controlled_writeback_dispatch_operation_type_has_schema_migration() -> None:
|
|
migrations_dir = Path(__file__).resolve().parents[1] / "migrations"
|
|
migration = Path(
|
|
migrations_dir / "adr090f_log_controlled_writeback_dispatch_operation_type.sql"
|
|
).read_text()
|
|
down = Path(
|
|
migrations_dir / "adr090f_log_controlled_writeback_dispatch_operation_type_down.sql"
|
|
).read_text()
|
|
|
|
assert "log_controlled_writeback_dispatched" in migration
|
|
assert "automation_operation_log_type_valid" in migration
|
|
assert "DROP CONSTRAINT IF EXISTS automation_operation_log_type_valid" in migration
|
|
assert "cannot remove log_controlled_writeback_dispatched" in down
|
|
|
|
|
|
def test_ansible_live_controlled_apply_sends_telegram_receipt_but_backfill_does_not() -> None:
|
|
live_source = inspect.getsource(run_controlled_apply_for_claim)
|
|
backfill_source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once)
|
|
|
|
assert "_send_controlled_apply_telegram_receipt" in live_source
|
|
assert "telegram_receipt_sent" in live_source
|
|
assert "_send_controlled_apply_telegram_receipt" not in backfill_source
|
|
assert inspect.iscoroutinefunction(_send_controlled_apply_telegram_receipt)
|
|
|
|
|
|
def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None:
|
|
source = inspect.getsource(run_failed_apply_check_mode_replay_once)
|
|
preflight_source = inspect.getsource(_load_open_failed_apply_retry_row)
|
|
|
|
assert "FOR UPDATE SKIP LOCKED" in preflight_source
|
|
assert "controlled_retry_check_mode_replay" in source
|
|
assert "build_ansible_check_mode_command" in source
|
|
assert "verifier.incident_id = coalesce(" in preflight_source
|
|
assert "apply.input ->> 'incident_id'" in preflight_source
|
|
assert "controlled_apply_allowed=True" in source
|
|
assert '"approval_required_before_apply": False' in source
|
|
assert '"owner_review_required": False' in source
|
|
assert '"runtime_apply_executed": False' in source
|
|
assert "queue_ai_playbook_or_transport_repair_candidate" in source
|
|
assert "retry_requires_repair_and_new_apply_gate" not in source
|
|
assert "_record_retry_runtime_stage_receipt" in source
|
|
assert "NOT EXISTS" in preflight_source
|
|
assert "failed_apply_retry_already_claimed" in source
|
|
assert "run_controlled_apply_for_claim" not in source
|
|
|
|
receipt_source = inspect.getsource(_record_retry_runtime_stage_receipt)
|
|
append_source = inspect.getsource(_append_runtime_stage_receipts_to_apply)
|
|
assert "_runtime_stage_receipt" in receipt_source
|
|
assert 'stage_id="retry_or_rollback"' in receipt_source
|
|
assert "jsonb_array_elements" in append_source
|
|
assert "ansible_controlled_retry_terminal_v2" in receipt_source
|
|
assert "ansible_retry_terminal_receipt_not_written_unverified" in receipt_source
|
|
assert "runtime_apply_executed" in receipt_source
|
|
|
|
|
|
def test_ansible_check_and_apply_rows_persist_canonical_automation_run_id() -> None:
|
|
finalize_source = inspect.getsource(finalize_check_mode_claim)
|
|
apply_source = inspect.getsource(run_controlled_apply_for_claim)
|
|
|
|
assert "'{automation_run_id}'" in finalize_source
|
|
assert '"automation_run_id": str(' in apply_source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ai_decision_handoff_queues_evidence_backed_single_writer_candidate() -> None:
|
|
incident = SimpleNamespace(
|
|
incident_id="INC-AIA-P0-002",
|
|
project_id="awoooi",
|
|
severity=SimpleNamespace(value="low"),
|
|
alertname="MomoPostgresBackupFailed",
|
|
alert_category="backup_failure",
|
|
notification_type="TYPE-3",
|
|
affected_services=["momo-postgres"],
|
|
signals=[
|
|
SimpleNamespace(
|
|
alert_name="MomoPostgresBackupFailed",
|
|
labels={"alertname": "MomoPostgresBackupFailed", "instance": "188"},
|
|
annotations={},
|
|
)
|
|
],
|
|
)
|
|
recorded: list[dict] = []
|
|
|
|
async def collect(**_kwargs):
|
|
return SimpleNamespace(snapshot_id="evidence-1")
|
|
|
|
async def verify(**_kwargs):
|
|
return True
|
|
|
|
async def record(**kwargs):
|
|
recorded.append(kwargs)
|
|
return True
|
|
|
|
handoff = await enqueue_ai_decision_ansible_candidate(
|
|
incident=incident,
|
|
proposal_data={
|
|
"source": "decision_manager",
|
|
"risk_level": "low",
|
|
"action": "systemctl restart momo-backup",
|
|
},
|
|
recorder=record,
|
|
evidence_collector=collect,
|
|
evidence_verifier=verify,
|
|
)
|
|
|
|
assert handoff["status"] == "controlled_check_mode_queued"
|
|
assert handoff["queued"] is True
|
|
assert handoff["side_effect_performed"] is False
|
|
assert handoff["single_writer_executor"] == "awoooi-ansible-executor-broker"
|
|
assert handoff["source_truth_diff"]["required_before_apply"] is True
|
|
assert handoff["risk_policy_decision"]["risk_level"] == "low"
|
|
assert recorded[0]["decision_path"] == "repair_candidate_controlled_queue"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ai_decision_handoff_blocks_critical_before_evidence_or_queue() -> None:
|
|
calls = 0
|
|
|
|
async def must_not_run(**_kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
return True
|
|
|
|
handoff = await enqueue_ai_decision_ansible_candidate(
|
|
incident=SimpleNamespace(incident_id="INC-CRITICAL"),
|
|
proposal_data={"risk_level": "critical", "action": "restart"},
|
|
recorder=must_not_run,
|
|
evidence_collector=must_not_run,
|
|
evidence_verifier=must_not_run,
|
|
)
|
|
|
|
assert handoff["status"] == "critical_break_glass_required"
|
|
assert handoff["queued"] is False
|
|
assert handoff["side_effect_performed"] is False
|
|
assert calls == 0
|
|
|
|
|
|
def test_controlled_apply_uses_durable_idempotency_guard() -> None:
|
|
source = inspect.getsource(run_controlled_apply_for_claim)
|
|
broker_source = inspect.getsource(run_pending_check_modes_once)
|
|
|
|
assert "pg_advisory_xact_lock" in source
|
|
assert "apply_idempotency_key" in source
|
|
assert "ansible_controlled_apply_duplicate_suppressed" in source
|
|
assert "apply_duplicate_suppressed" in broker_source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_controlled_apply_duplicate_never_runs_subprocess(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from src.services import awooop_ansible_check_mode_service as service
|
|
|
|
class ExistingResult:
|
|
def mappings(self):
|
|
return self
|
|
|
|
def first(self):
|
|
return {
|
|
"op_id": "00000000-0000-0000-0000-000000000103",
|
|
"status": "success",
|
|
}
|
|
|
|
class FakeDb:
|
|
async def execute(self, *_args, **_kwargs):
|
|
return ExistingResult()
|
|
|
|
class FakeContext:
|
|
async def __aenter__(self):
|
|
return FakeDb()
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
subprocess_started = False
|
|
|
|
async def must_not_run(*_args, **_kwargs):
|
|
nonlocal subprocess_started
|
|
subprocess_started = True
|
|
raise AssertionError("duplicate apply must not execute")
|
|
|
|
monkeypatch.setattr(
|
|
service,
|
|
"get_db_context",
|
|
lambda _project_id: FakeContext(),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_execution_capability_timeout_seconds",
|
|
lambda *_args, **_kwargs: 30,
|
|
)
|
|
monkeypatch.setattr(service, "_run_ansible_command", must_not_run)
|
|
|
|
claim = AnsibleCheckModeClaim(
|
|
op_id="00000000-0000-0000-0000-000000000102",
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000101",
|
|
incident_id="INC-IDEMPOTENT",
|
|
catalog_id="ansible:188-momo-backup-user",
|
|
playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml",
|
|
inventory_hosts=("host_188",),
|
|
risk_level="low",
|
|
input_payload={
|
|
"controlled_apply_allowed": True,
|
|
"apply_idempotency_key": (
|
|
"ansible-apply:00000000-0000-0000-0000-000000000101:"
|
|
"ansible:188-momo-backup-user"
|
|
),
|
|
},
|
|
)
|
|
|
|
result = await service.run_controlled_apply_for_claim(
|
|
claim,
|
|
timeout_seconds=30,
|
|
)
|
|
|
|
assert result is None
|
|
assert subprocess_started is False
|
|
assert claim.input_payload["apply_duplicate_suppressed"] is True
|
|
assert claim.input_payload["existing_apply_status"] == "success"
|
|
|
|
|
|
def test_ansible_subprocess_is_terminated_when_worker_task_is_cancelled() -> None:
|
|
source = inspect.getsource(_run_ansible_command)
|
|
|
|
assert "except asyncio.CancelledError" in source
|
|
assert "process.kill()" in source
|
|
assert "await process.communicate()" in source
|
|
|
|
|
|
def test_controlled_apply_shutdown_marks_pending_row_before_cancelling() -> None:
|
|
source = inspect.getsource(run_controlled_apply_for_claim)
|
|
|
|
assert "except asyncio.CancelledError" in source
|
|
assert "ansible_controlled_apply_interrupted_by_worker_shutdown" in source
|
|
assert '"status": status' in source
|
|
assert '"op_id": apply_op_id' in source
|
|
assert "ansible_controlled_apply_interrupted" in source
|
|
|
|
|
|
def test_ansible_post_apply_verifier_helpers_are_deterministic() -> None:
|
|
assert _post_apply_km_path_type("03ca6836-1b76-4da2-8e3e-6d3b6df9254a") == (
|
|
"ansible_apply_receipt:03ca6836"
|
|
)
|
|
successful_apply = AnsibleRunResult(
|
|
returncode=0,
|
|
stdout="",
|
|
stderr="",
|
|
duration_ms=1,
|
|
)
|
|
assert _post_apply_verification_result(successful_apply) == "failed"
|
|
assert _post_apply_verification_result(
|
|
successful_apply,
|
|
{
|
|
"verification_result": "success",
|
|
"all_postconditions_passed": True,
|
|
"executor_returncode_trusted": False,
|
|
},
|
|
) == "success"
|
|
assert _post_apply_verification_result(
|
|
AnsibleRunResult(returncode=1, stdout="", stderr="", duration_ms=1)
|
|
) == "failed"
|
|
assert _post_apply_verification_result(
|
|
AnsibleRunResult(returncode=0, stdout="", stderr="", duration_ms=1, timed_out=True)
|
|
) == "timeout"
|
|
|
|
|
|
def test_ansible_runtime_stage_receipts_are_same_run_and_public_safe() -> None:
|
|
claim = AnsibleCheckModeClaim(
|
|
op_id="00000000-0000-0000-0000-000000000102",
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000101",
|
|
incident_id="INC-RUNTIME-STAGE",
|
|
catalog_id="ansible:188-momo-backup-user",
|
|
playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml",
|
|
inventory_hosts=("host_188",),
|
|
risk_level="low",
|
|
input_payload={
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000101",
|
|
},
|
|
)
|
|
result = AnsibleRunResult(
|
|
returncode=0,
|
|
stdout="changed=0",
|
|
stderr="",
|
|
duration_ms=42,
|
|
)
|
|
|
|
pre = build_ansible_pre_apply_runtime_stage_receipts(claim)
|
|
post = build_ansible_post_apply_runtime_stage_receipts(
|
|
claim,
|
|
result,
|
|
apply_op_id="00000000-0000-0000-0000-000000000103",
|
|
verifier_ready=True,
|
|
)
|
|
|
|
assert {receipt["stage_id"] for receipt in pre} == {
|
|
"normalized_asset_identity",
|
|
"source_truth_diff",
|
|
"risk_policy_decision",
|
|
}
|
|
assert {receipt["stage_id"] for receipt in post} == {
|
|
"executor_log_projection",
|
|
"retry_or_rollback",
|
|
}
|
|
assert all(receipt["durable_receipt"] is True for receipt in [*pre, *post])
|
|
assert all(receipt["raw_log_payload_stored"] is False for receipt in [*pre, *post])
|
|
assert all(receipt["secret_value_stored"] is False for receipt in [*pre, *post])
|
|
assert all(
|
|
receipt["automation_run_id"]
|
|
== "00000000-0000-0000-0000-000000000101"
|
|
for receipt in [*pre, *post]
|
|
)
|
|
assert inspect.iscoroutinefunction(_record_runtime_stage_receipts)
|
|
receipt_writer_source = inspect.getsource(_record_runtime_stage_receipts)
|
|
assert "jsonb_array_elements" in receipt_writer_source
|
|
assert "WITH ORDINALITY" in receipt_writer_source
|
|
assert "DISTINCT ON" in receipt_writer_source
|
|
|
|
|
|
def test_ansible_context_receipts_reference_pre_decision_evidence_without_content() -> None:
|
|
claim = AnsibleCheckModeClaim(
|
|
op_id="00000000-0000-0000-0000-000000000112",
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000111",
|
|
incident_id="INC-RUNTIME-CONTEXT",
|
|
catalog_id="ansible:188-ai-web",
|
|
playbook_path="infra/ansible/playbooks/188-ai-web-readonly.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/188-ai-web.yml",
|
|
inventory_hosts=("host_188",),
|
|
risk_level="medium",
|
|
input_payload={
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000111",
|
|
},
|
|
)
|
|
receipts = build_ansible_context_runtime_stage_receipts(
|
|
claim,
|
|
{
|
|
"id": "evidence-111",
|
|
"recent_logs": "sanitized service log summary",
|
|
"mcp_health": {"prometheus": True, "kubernetes": False},
|
|
"sensors_attempted": 2,
|
|
"sensors_succeeded": 1,
|
|
},
|
|
)
|
|
|
|
assert {receipt["stage_id"] for receipt in receipts} == {
|
|
"mcp_context",
|
|
"service_log_evidence",
|
|
}
|
|
assert all(receipt["automation_run_id"] == claim.source_candidate_op_id for receipt in receipts)
|
|
assert all(receipt["raw_log_payload_stored"] is False for receipt in receipts)
|
|
service_log = next(
|
|
receipt for receipt in receipts if receipt["stage_id"] == "service_log_evidence"
|
|
)
|
|
assert service_log["detail"]["content_in_receipt"] is False
|
|
assert "sanitized service log summary" not in str(service_log)
|
|
|
|
loader_source = inspect.getsource(_load_pre_decision_context_runtime_stage_receipts)
|
|
assert "evidence.collected_at <= candidate.created_at" in loader_source
|
|
assert "evidence.post_execution_state IS NULL" in loader_source
|
|
assert "CAST(evidence.post_execution_state AS jsonb)" in loader_source
|
|
assert "= 'null'::jsonb" in loader_source
|
|
|
|
|
|
def test_ansible_context_receipt_rejects_zero_success_mcp_attempts() -> None:
|
|
claim = AnsibleCheckModeClaim(
|
|
op_id="00000000-0000-0000-0000-000000000132",
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000131",
|
|
incident_id="INC-RUNTIME-CONTEXT-FAILED",
|
|
catalog_id="ansible:188-ai-web",
|
|
playbook_path="infra/ansible/playbooks/188-ai-web-readonly.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/188-ai-web.yml",
|
|
inventory_hosts=("host_188",),
|
|
risk_level="medium",
|
|
input_payload={},
|
|
)
|
|
|
|
receipts = build_ansible_context_runtime_stage_receipts(
|
|
claim,
|
|
{
|
|
"id": "evidence-failed",
|
|
"recent_logs": None,
|
|
"mcp_health": {"prometheus": False, "kubernetes": False},
|
|
"sensors_attempted": 2,
|
|
"sensors_succeeded": 0,
|
|
},
|
|
)
|
|
|
|
assert receipts == []
|
|
|
|
|
|
def test_ansible_timeline_projection_receipt_is_same_run_and_idempotent() -> None:
|
|
claim = AnsibleCheckModeClaim(
|
|
op_id="00000000-0000-0000-0000-000000000122",
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000121",
|
|
incident_id="INC-RUNTIME-TIMELINE",
|
|
catalog_id="ansible:188-ai-web",
|
|
playbook_path="infra/ansible/playbooks/188-ai-web-readonly.yml",
|
|
apply_playbook_path="infra/ansible/playbooks/188-ai-web.yml",
|
|
inventory_hosts=("host_188",),
|
|
risk_level="medium",
|
|
input_payload={
|
|
"automation_run_id": "00000000-0000-0000-0000-000000000121",
|
|
},
|
|
)
|
|
result = AnsibleRunResult(
|
|
returncode=2,
|
|
stdout="",
|
|
stderr="apply failed",
|
|
duration_ms=42,
|
|
)
|
|
receipt = build_ansible_timeline_runtime_stage_receipt(
|
|
claim,
|
|
result,
|
|
apply_op_id="00000000-0000-0000-0000-000000000123",
|
|
timeline_event_id="timeline-123",
|
|
)
|
|
|
|
assert receipt["stage_id"] == "timeline_projection"
|
|
assert receipt["automation_run_id"] == claim.source_candidate_op_id
|
|
assert receipt["evidence_ref"] == "timeline_events:timeline-123"
|
|
assert receipt["detail"]["projection_status"] == "error"
|
|
writer_source = inspect.getsource(_record_timeline_projection_receipt)
|
|
assert "WITH existing AS" in writer_source
|
|
assert "gen_random_uuid()::text" in writer_source
|
|
assert "WHERE NOT EXISTS (SELECT 1 FROM existing)" in writer_source
|
|
assert inspect.iscoroutinefunction(_record_timeline_projection_receipt)
|
|
|
|
|
|
def test_ansible_claim_query_limits_recent_candidate_backlog() -> None:
|
|
source = inspect.getsource(claim_pending_check_modes)
|
|
|
|
assert "candidate.created_at >= NOW() - (:candidate_max_age_hours * INTERVAL '1 hour')" in source
|
|
assert "AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS" in source
|
|
|
|
|
|
def test_ansible_claim_query_honors_bounded_execution_priority_before_fifo() -> None:
|
|
source = inspect.getsource(claim_pending_check_modes)
|
|
|
|
assert "candidate.input ->> 'work_item_id' LIKE 'P0-%'" in source
|
|
assert "alert_webhook_controlled_router" in source
|
|
assert "ELSE 30" in source
|
|
assert "candidate.input ->> 'execution_priority'" in source
|
|
assert "LEAST(" in source
|
|
assert "100" in source
|
|
assert source.index("execution_priority") < source.index("candidate.created_at ASC")
|
|
|
|
|
|
def test_ansible_transport_blocker_detects_repair_forced_command_denial() -> None:
|
|
blockers = detect_ansible_transport_blockers(
|
|
"fatal: host unreachable REPAIR_DENIED:invalid_command",
|
|
)
|
|
|
|
assert blockers == ["ansible_repair_ssh_forced_command_denies_ansible_bootstrap"]
|
|
|
|
|
|
def test_execution_backend_summary_subtracts_completed_check_mode_parent() -> None:
|
|
summary = _execution_backend_summary([
|
|
{
|
|
"execution": {
|
|
"ansible": {
|
|
"considered": True,
|
|
"candidate_catalog": {"candidates": [{"catalog_id": "ansible:188-ai-web"}]},
|
|
"records": [
|
|
{
|
|
"op_id": "candidate-1",
|
|
"operation_type": "ansible_candidate_matched",
|
|
"status": "dry_run",
|
|
},
|
|
{
|
|
"op_id": "check-1",
|
|
"parent_op_id": "candidate-1",
|
|
"operation_type": "ansible_check_mode_executed",
|
|
"status": "success",
|
|
},
|
|
],
|
|
},
|
|
"automation_operation_log": [],
|
|
"auto_repair_executions": [],
|
|
},
|
|
"automation_quality": {"facts": {}},
|
|
}
|
|
])
|
|
|
|
assert summary["ansible_check_mode_total"] == 1
|
|
assert summary["ansible_pending_check_mode_total"] == 0
|
|
|
|
|
|
def test_quality_summary_marks_forced_command_denial_as_runtime_blocker(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
settings,
|
|
"AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH",
|
|
"/etc/repair-ssh/id_ed25519",
|
|
)
|
|
summary = summarize_automation_quality_records(
|
|
project_id="awoooi",
|
|
window_hours=24,
|
|
limit=20,
|
|
records=[
|
|
{
|
|
"incident": {"incident_id": "INC-1", "alertname": "DockerContainerUnhealthy"},
|
|
"truth_status": {},
|
|
"automation_quality": {"applicable": True, "score": 50, "verdict": "observed"},
|
|
"execution": {
|
|
"automation_operation_log": [],
|
|
"auto_repair_executions": [],
|
|
"ansible": {
|
|
"considered": True,
|
|
"candidate_catalog": {"candidates": [{"catalog_id": "ansible:110-devops"}]},
|
|
"records": [
|
|
{
|
|
"op_id": "check-1",
|
|
"operation_type": "ansible_check_mode_executed",
|
|
"status": "failed",
|
|
"dry_run_result": {
|
|
"stdout_tail": "REPAIR_DENIED:invalid_command",
|
|
},
|
|
}
|
|
],
|
|
},
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert summary["ansible_runtime"]["can_run_check_mode"] is False
|
|
assert (
|
|
"ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
|
|
in summary["ansible_runtime"]["blockers"]
|
|
)
|
|
|
|
|
|
def test_quality_summary_keeps_repair_forced_blocker_historical_for_ssh_mcp(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
settings,
|
|
"AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH",
|
|
"/run/secrets/ssh_mcp_key",
|
|
)
|
|
summary = summarize_automation_quality_records(
|
|
project_id="awoooi",
|
|
window_hours=24,
|
|
limit=20,
|
|
records=[
|
|
{
|
|
"incident": {"incident_id": "INC-1", "alertname": "DockerContainerUnhealthy"},
|
|
"truth_status": {},
|
|
"automation_quality": {"applicable": True, "score": 50, "verdict": "observed"},
|
|
"execution": {
|
|
"automation_operation_log": [],
|
|
"auto_repair_executions": [],
|
|
"ansible": {
|
|
"considered": True,
|
|
"candidate_catalog": {"candidates": [{"catalog_id": "ansible:110-devops"}]},
|
|
"records": [
|
|
{
|
|
"op_id": "check-1",
|
|
"operation_type": "ansible_check_mode_executed",
|
|
"status": "failed",
|
|
"dry_run_result": {
|
|
"stdout_tail": "REPAIR_DENIED:invalid_command",
|
|
},
|
|
}
|
|
],
|
|
},
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert (
|
|
"ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
|
|
in summary["ansible_runtime"]["historical_transport_blockers"]
|
|
)
|
|
assert (
|
|
"ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
|
|
not in summary["ansible_runtime"]["blockers"]
|
|
)
|