108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
from src.services.controlled_alert_target_router import (
|
|
resolve_typed_alert_target,
|
|
)
|
|
from src.services.independent_verifier_registry import (
|
|
build_independent_verifier_registry,
|
|
)
|
|
|
|
|
|
def test_registry_covers_every_typed_domain_without_false_runtime_closure() -> None:
|
|
registry = build_independent_verifier_registry()
|
|
|
|
assert registry["status"] == (
|
|
"source_coverage_complete_runtime_receipts_pending"
|
|
)
|
|
assert registry["source_coverage_complete"] is True
|
|
assert registry["runtime_closed"] is False
|
|
assert registry["rollups"] == {
|
|
"domain_count": 8,
|
|
"mutating_domain_count": 6,
|
|
"read_only_domain_count": 2,
|
|
"source_blocker_count": 0,
|
|
"runtime_closed_domain_count": 0,
|
|
}
|
|
assert registry["ansible_catalog_coverage"] == {
|
|
"catalog_count": 17,
|
|
"check_mode_catalog_count": 16,
|
|
"auto_apply_catalog_count": 13,
|
|
"registered_postcondition_count": 17,
|
|
"missing_postcondition_catalog_ids": [],
|
|
"orphan_postcondition_catalog_ids": [],
|
|
}
|
|
by_domain = {
|
|
row["domain"]: row for row in registry["domain_contracts"]
|
|
}
|
|
assert set(by_domain) == {
|
|
"kubernetes_workload",
|
|
"host_systemd",
|
|
"docker_container",
|
|
"windows_vmware",
|
|
"database",
|
|
"backup_restore",
|
|
"unknown",
|
|
"control_plane_recovery",
|
|
}
|
|
assert all(
|
|
row["cross_domain_fallback_allowed"] is False
|
|
for row in by_domain.values()
|
|
)
|
|
assert all(
|
|
row["executor"] != row["verifier"]
|
|
for row in by_domain.values()
|
|
if row["mutating"] is True
|
|
)
|
|
assert by_domain["backup_restore"]["runtime_write_allowed"] is False
|
|
assert by_domain["backup_restore"]["executor"] == "Agent99"
|
|
assert by_domain["backup_restore"]["break_glass_executor"] == (
|
|
"backup_restore_break_glass"
|
|
)
|
|
assert by_domain["unknown"]["runtime_write_allowed"] is False
|
|
assert by_domain["unknown"]["executor"] is None
|
|
assert all(
|
|
row["runtime_status"] == "receipt_pending"
|
|
for row in registry["external_readbacks"]
|
|
)
|
|
|
|
|
|
def test_backup_readback_route_matches_independent_verifier_contract() -> None:
|
|
route = resolve_typed_alert_target(
|
|
alertname="BackupFreshnessMissing",
|
|
target_resource="backup_restore",
|
|
namespace="",
|
|
)
|
|
registry = build_independent_verifier_registry()
|
|
contract = next(
|
|
row
|
|
for row in registry["domain_contracts"]
|
|
if row["domain"] == route["target_kind"]
|
|
)
|
|
|
|
assert route["executor"] == contract["executor"] == "Agent99"
|
|
assert route["break_glass_executor"] == contract["break_glass_executor"]
|
|
assert route["verifier"] == contract["verifier"]
|
|
assert route["controlled_apply_allowed"] is False
|
|
assert contract["runtime_write_allowed"] is False
|
|
|
|
|
|
def test_registry_fails_closed_when_one_catalog_loses_its_postcondition() -> None:
|
|
catalog = [
|
|
{
|
|
"catalog_id": "ansible:one",
|
|
"supports_check_mode": True,
|
|
"auto_apply_enabled": True,
|
|
}
|
|
]
|
|
registry = build_independent_verifier_registry(
|
|
ansible_catalog=catalog,
|
|
ansible_registered_catalog_ids=(),
|
|
)
|
|
|
|
assert registry["source_coverage_complete"] is False
|
|
assert registry["status"] == "source_coverage_incomplete_fail_closed"
|
|
assert registry["ansible_catalog_coverage"][
|
|
"missing_postcondition_catalog_ids"
|
|
] == ["ansible:one"]
|
|
assert registry["source_blockers"] == [
|
|
"ansible_postcondition_missing:ansible:one"
|
|
]
|