dbf8d0255e
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled
257 lines
8.6 KiB
Python
257 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
|
|
from src.services.awooop_ansible_audit_service import get_ansible_catalog_item
|
|
from src.services.awooop_ansible_check_mode_service import (
|
|
_apply_ansible_playbook_trust_gate,
|
|
_revalidate_claim_playbook_trust,
|
|
run_controlled_apply_for_claim,
|
|
)
|
|
|
|
|
|
def _candidate(
|
|
catalog_id: str,
|
|
*,
|
|
catalog_revision: str = "test-revision-v1",
|
|
) -> dict[str, object]:
|
|
catalog_item = get_ansible_catalog_item(catalog_id) or {}
|
|
return {
|
|
"catalog_id": catalog_id,
|
|
"playbook_path": f"infra/ansible/playbooks/{catalog_id.split(':', 1)[1]}.yml",
|
|
"inventory_hosts": ["host_188"],
|
|
"risk_level": "medium",
|
|
"catalog_revision": (
|
|
catalog_item.get("catalog_revision") or catalog_revision
|
|
),
|
|
}
|
|
|
|
|
|
def test_low_trust_failed_playbook_opens_claim_circuit() -> None:
|
|
candidate = _candidate("ansible:nginx-sync")
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-NGINX-SYNC",
|
|
"status": "approved",
|
|
"trust_score": 0.001417,
|
|
"success_count": 0,
|
|
"failure_count": 24,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": candidate["catalog_revision"],
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == []
|
|
assert gate["status"] == "circuit_open"
|
|
assert gate["blocked_candidate_count"] == 1
|
|
assert gate["decisions"][0]["reason"] == (
|
|
"playbook_trust_below_archive_threshold"
|
|
)
|
|
assert gate["next_controlled_action"] == (
|
|
"generate_or_verify_playbook_repair_candidate"
|
|
)
|
|
|
|
|
|
def test_low_trust_candidate_is_filtered_without_blocking_healthy_candidate() -> None:
|
|
failed = _candidate("ansible:nginx-sync")
|
|
healthy = _candidate("ansible:188-momo-backup-user")
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{
|
|
"executor_candidates": [
|
|
failed,
|
|
healthy,
|
|
]
|
|
},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-NGINX-SYNC",
|
|
"status": "approved",
|
|
"trust_score": 0.01,
|
|
"success_count": 0,
|
|
"failure_count": 15,
|
|
"review_required": True,
|
|
"symptom_pattern": {
|
|
"catalog_revision": failed["catalog_revision"],
|
|
},
|
|
},
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-188-MOMO-BACKUP-USER",
|
|
"status": "approved",
|
|
"trust_score": 0.65,
|
|
"success_count": 7,
|
|
"failure_count": 1,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": healthy["catalog_revision"],
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == [healthy]
|
|
assert gate["status"] == "degraded_candidates_filtered"
|
|
assert gate["eligible_candidate_count"] == 1
|
|
assert gate["blocked_candidate_count"] == 1
|
|
|
|
|
|
def test_repaired_catalog_revision_gets_one_bounded_canary() -> None:
|
|
candidate = _candidate(
|
|
"ansible:110-devops",
|
|
catalog_revision="host110-route-v3",
|
|
)
|
|
current_revision = str(candidate["catalog_revision"])
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-110-DEVOPS",
|
|
"status": "approved",
|
|
"trust_score": 0.024,
|
|
"success_count": 0,
|
|
"failure_count": 1,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": f"{current_revision}-previous",
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == [candidate]
|
|
assert gate["status"] == "closed"
|
|
decision = gate["decisions"][0]
|
|
assert decision["circuit_open"] is False
|
|
assert decision["bounded_repair_canary"] is True
|
|
assert decision["reason"] == "repaired_catalog_revision_canary_allowed"
|
|
assert decision["current_catalog_revision"] == current_revision
|
|
assert decision["catalog_revision_source"] == "allowlisted_catalog"
|
|
assert decision["observed_catalog_revision"] == (
|
|
f"{current_revision}-previous"
|
|
)
|
|
|
|
|
|
def test_failed_repaired_revision_reopens_circuit_for_same_revision() -> None:
|
|
candidate = _candidate(
|
|
"ansible:110-devops",
|
|
catalog_revision="host110-route-v3",
|
|
)
|
|
current_revision = str(candidate["catalog_revision"])
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-110-DEVOPS",
|
|
"status": "approved",
|
|
"trust_score": 0.0192,
|
|
"success_count": 0,
|
|
"failure_count": 2,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": current_revision,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == []
|
|
assert gate["status"] == "circuit_open"
|
|
decision = gate["decisions"][0]
|
|
assert decision["circuit_open"] is True
|
|
assert decision["bounded_repair_canary"] is False
|
|
assert decision["reason"] == "playbook_trust_below_archive_threshold"
|
|
|
|
|
|
def test_legacy_failed_record_without_revision_gets_migration_canary() -> None:
|
|
candidate = _candidate(
|
|
"ansible:110-devops",
|
|
catalog_revision="host110-route-v3",
|
|
)
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-110-DEVOPS",
|
|
"status": "approved",
|
|
"trust_score": 0.02,
|
|
"success_count": 0,
|
|
"failure_count": 3,
|
|
"review_required": False,
|
|
"symptom_pattern": {},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == [candidate]
|
|
decision = gate["decisions"][0]
|
|
assert decision["bounded_repair_canary"] is True
|
|
assert decision["observed_catalog_revision"] is None
|
|
|
|
|
|
def test_candidate_cannot_forge_revision_to_bypass_same_revision_circuit() -> None:
|
|
catalog_id = "ansible:110-devops"
|
|
current_revision = str(
|
|
(get_ansible_catalog_item(catalog_id) or {}).get("catalog_revision")
|
|
or ""
|
|
)
|
|
candidate = _candidate(catalog_id)
|
|
candidate["catalog_revision"] = "forged-revision-v999"
|
|
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-110-DEVOPS",
|
|
"status": "approved",
|
|
"trust_score": 0.02,
|
|
"success_count": 0,
|
|
"failure_count": 3,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": current_revision,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == []
|
|
decision = gate["decisions"][0]
|
|
assert decision["bounded_repair_canary"] is False
|
|
assert decision["candidate_catalog_revision_mismatch"] is True
|
|
assert decision["current_catalog_revision"] == current_revision
|
|
|
|
|
|
def test_owner_review_flag_does_not_become_low_medium_high_terminal_gate() -> None:
|
|
candidate = _candidate("ansible:188-momo-backup-user")
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-188-MOMO-BACKUP-USER",
|
|
"status": "approved",
|
|
"trust_score": 0.4,
|
|
"success_count": 2,
|
|
"failure_count": 1,
|
|
"review_required": True,
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == [candidate]
|
|
assert gate["status"] == "closed"
|
|
assert gate["owner_review_used_as_terminal"] is False
|
|
|
|
|
|
def test_controlled_apply_revalidates_trust_before_runtime_write() -> None:
|
|
apply_source = inspect.getsource(run_controlled_apply_for_claim)
|
|
revalidate_source = inspect.getsource(_revalidate_claim_playbook_trust)
|
|
|
|
assert "_revalidate_claim_playbook_trust" in apply_source
|
|
assert "ansible_controlled_apply_blocked_by_playbook_trust" in apply_source
|
|
assert "playbook_trust_circuit_open" in revalidate_source
|
|
assert "UPDATE automation_operation_log" in revalidate_source
|