227 lines
7.3 KiB
Python
227 lines
7.3 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from src.services.backup_restore_evidence_verifier import (
|
|
build_backup_restore_evidence_gate,
|
|
)
|
|
|
|
RUN_ID = "1df4455e-e49b-4f42-9a03-a5cb3c48bd35"
|
|
TRACE_ID = "00-12345678901234567890123456789012-1234567890123456-01"
|
|
WORK_ITEM_ID = "agent99-dispatch:awoooi:INC-BACKUP:BackupCheck"
|
|
VERIFIED_AT = "2026-07-11T20:00:00+08:00"
|
|
|
|
|
|
def _receipt(
|
|
dimension: str,
|
|
schema_version: str,
|
|
source: str,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"schema_version": schema_version,
|
|
"receipt_id": f"backup-metadata:{dimension}:1",
|
|
"source": source,
|
|
"run_id": RUN_ID,
|
|
"trace_id": TRACE_ID,
|
|
"work_item_id": WORK_ITEM_ID,
|
|
"status": "verified",
|
|
"durable_readback_ack": True,
|
|
"read_only": True,
|
|
"independent": True,
|
|
"runtime_mutation_performed": False,
|
|
"secret_values_included": False,
|
|
"observed_at": "2026-07-11T19:55:00+08:00",
|
|
"max_age_seconds": 900,
|
|
}
|
|
|
|
|
|
def valid_backup_evidence_bundle() -> dict[str, object]:
|
|
backup_status = _receipt(
|
|
"backup-status",
|
|
"backup_status_evidence_receipt_v1",
|
|
"backup_health_exporter",
|
|
)
|
|
backup_status.update(
|
|
{
|
|
"backup_succeeded": True,
|
|
"repository_integrity_verified": True,
|
|
}
|
|
)
|
|
freshness = _receipt(
|
|
"freshness",
|
|
"backup_freshness_evidence_receipt_v1",
|
|
"backup_freshness_exporter",
|
|
)
|
|
freshness.update(
|
|
{
|
|
"backup_age_seconds": 600,
|
|
"rpo_seconds": 3600,
|
|
}
|
|
)
|
|
offsite = _receipt(
|
|
"offsite",
|
|
"offsite_verify_evidence_receipt_v1",
|
|
"offsite_readonly_verifier",
|
|
)
|
|
offsite.update(
|
|
{
|
|
"independent_copy_verified": True,
|
|
"immutable_copy_verified": True,
|
|
}
|
|
)
|
|
escrow = _receipt(
|
|
"escrow",
|
|
"escrow_metadata_evidence_receipt_v1",
|
|
"escrow_metadata_readonly_verifier",
|
|
)
|
|
escrow.update(
|
|
{
|
|
"metadata_only": True,
|
|
"dual_control_verified": True,
|
|
"required_item_count": 5,
|
|
"verified_item_count": 5,
|
|
"missing_item_count": 0,
|
|
"evidence_age_seconds": 3600,
|
|
"max_evidence_age_seconds": 2678400,
|
|
}
|
|
)
|
|
restore = _receipt(
|
|
"restore",
|
|
"restore_drill_evidence_receipt_v1",
|
|
"isolated_restore_drill_verifier",
|
|
)
|
|
restore.update(
|
|
{
|
|
"isolated_target": True,
|
|
"production_target_used": False,
|
|
"destructive_restore_performed": False,
|
|
"integrity_verified": True,
|
|
"restored_object_count": 5,
|
|
"drill_age_seconds": 86400,
|
|
"max_drill_age_seconds": 2678400,
|
|
"rto_seconds": 600,
|
|
"rto_target_seconds": 3600,
|
|
}
|
|
)
|
|
return {
|
|
"schema_version": "backup_restore_evidence_bundle_v1",
|
|
"receipt_id": "backup-metadata-bundle:1",
|
|
"status": "verified",
|
|
"run_id": RUN_ID,
|
|
"trace_id": TRACE_ID,
|
|
"work_item_id": WORK_ITEM_ID,
|
|
"durable_readback_ack": True,
|
|
"read_only": True,
|
|
"runtime_mutation_performed": False,
|
|
"secret_values_included": False,
|
|
"receipts": {
|
|
"backup_status": backup_status,
|
|
"freshness": freshness,
|
|
"offsite_verify": offsite,
|
|
"escrow": escrow,
|
|
"restore_drill": restore,
|
|
},
|
|
}
|
|
|
|
|
|
def _gate(bundle: dict[str, object]) -> dict[str, object]:
|
|
return build_backup_restore_evidence_gate(
|
|
run_id=RUN_ID,
|
|
trace_id=TRACE_ID,
|
|
work_item_id=WORK_ITEM_ID,
|
|
verified_at=VERIFIED_AT,
|
|
evidence_bundle=bundle,
|
|
)
|
|
|
|
|
|
def test_complete_same_run_metadata_builds_verified_dr_scorecard() -> None:
|
|
result = _gate(valid_backup_evidence_bundle())
|
|
|
|
assert result["status"] == "verified"
|
|
assert result["verified"] is True
|
|
assert result["durable_readback_ack"] is True
|
|
assert result["same_run_receipts"] is True
|
|
assert result["active_blockers"] == []
|
|
assert result["runtime_mutation_performed"] is False
|
|
assert result["production_restore_performed"] is False
|
|
assert result["secret_boundary_violated"] is False
|
|
assert result["secret_values_recorded"] is False
|
|
assert result["read_only"] is True
|
|
assert result["dr_scorecard"] == {
|
|
"schema_version": "backup_restore_dr_scorecard_v1",
|
|
"status": "verified",
|
|
"score_percent": 100,
|
|
"passed_dimension_count": 5,
|
|
"required_dimension_count": 5,
|
|
"rpo_met": True,
|
|
"rto_met": True,
|
|
"escrow_complete": True,
|
|
"restore_drill_verified": True,
|
|
"active_blockers": [],
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("dimension", ["escrow", "restore_drill"])
|
|
def test_missing_critical_metadata_receipt_stays_blocked(dimension: str) -> None:
|
|
bundle = valid_backup_evidence_bundle()
|
|
del bundle["receipts"][dimension] # type: ignore[index]
|
|
|
|
result = _gate(bundle)
|
|
|
|
assert result["status"] == "blocked"
|
|
assert result["verified"] is False
|
|
assert f"{dimension}_receipt_missing" in result["active_blockers"]
|
|
assert result["dr_scorecard"]["score_percent"] < 100 # type: ignore[index]
|
|
|
|
|
|
def test_stale_or_cross_run_receipt_fails_closed() -> None:
|
|
stale = valid_backup_evidence_bundle()
|
|
stale["receipts"]["escrow"]["observed_at"] = ( # type: ignore[index]
|
|
"2026-07-11T19:00:00+08:00"
|
|
)
|
|
stale_result = _gate(stale)
|
|
assert "escrow_readback_stale" in stale_result["active_blockers"]
|
|
|
|
cross_run = valid_backup_evidence_bundle()
|
|
cross_run["receipts"]["restore_drill"]["run_id"] = ( # type: ignore[index]
|
|
"foreign-run"
|
|
)
|
|
cross_run_result = _gate(cross_run)
|
|
assert cross_run_result["same_run_receipts"] is False
|
|
assert "restore_drill_run_id_mismatch" in cross_run_result["active_blockers"]
|
|
|
|
|
|
def test_secret_shaped_or_production_restore_metadata_never_false_greens() -> None:
|
|
secret_bundle = valid_backup_evidence_bundle()
|
|
secret_bundle["secret"] = "must-not-be-projected"
|
|
secret_result = _gate(secret_bundle)
|
|
assert secret_result["verified"] is False
|
|
assert secret_result["secret_boundary_violated"] is True
|
|
assert secret_result["secret_values_recorded"] is False
|
|
assert (
|
|
"evidence_bundle_forbidden_secret_key_present"
|
|
in secret_result["active_blockers"]
|
|
)
|
|
assert "must-not-be-projected" not in str(secret_result)
|
|
|
|
production_restore = valid_backup_evidence_bundle()
|
|
production_restore["receipts"]["restore_drill"]["production_target_used"] = True # type: ignore[index]
|
|
restore_result = _gate(production_restore)
|
|
assert restore_result["verified"] is False
|
|
assert restore_result["production_restore_performed"] is True
|
|
assert "restore_drill_postcondition_unverified" in restore_result["active_blockers"]
|
|
|
|
|
|
def test_gate_receipt_is_deterministic_and_binds_scorecard_metadata() -> None:
|
|
bundle = valid_backup_evidence_bundle()
|
|
first = _gate(bundle)
|
|
second = _gate(deepcopy(bundle))
|
|
changed = deepcopy(bundle)
|
|
changed["receipts"]["freshness"]["backup_age_seconds"] = 601 # type: ignore[index]
|
|
changed_result = _gate(changed)
|
|
|
|
assert first["receipt_id"] == second["receipt_id"]
|
|
assert first["receipt_id"] != changed_result["receipt_id"]
|