ca80972dc7
Some checks failed
Code Review / ai-code-review (push) Successful in 10s
run-migration / migrate (push) Failing after 9s
CD Pipeline / tests (push) Successful in 2m21s
CD Pipeline / build-and-deploy (push) Successful in 3m50s
CD Pipeline / post-deploy-checks (push) Successful in 1m19s
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from src.services.awooop_ansible_audit_service import build_ansible_truth
|
|
from src.services.awooop_truth_chain_service import _clean_row, _truth_status
|
|
|
|
|
|
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_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_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_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"
|