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