Add audit recovery exercise preflights

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa

tests/test_audit_core_recovery_preflight.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import importlib.util
+import unittest
+from datetime import UTC, datetime
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC = importlib.util.spec_from_file_location(
+    "audit_core_recovery_preflight",
+    ROOT / "scripts" / "audit-core-recovery-preflight.py",
+)
+assert SPEC and SPEC.loader
+module = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(module)
+
+
+class RecoveryPreflightTests(unittest.TestCase):
+    def test_condition_selects_named_condition(self) -> None:
+        resource = {
+            "status": {
+                "conditions": [
+                    {"type": "Ready", "status": "True"},
+                    {"type": "ContinuousArchiving", "status": "False"},
+                ]
+            }
+        }
+        self.assertEqual("True", module.condition(resource, "Ready"))
+        self.assertEqual("False", module.condition(resource, "ContinuousArchiving"))
+        self.assertIsNone(module.condition(resource, "Missing"))
+
+    def test_latest_backup_uses_completed_stop_time_and_reports_no_values(self) -> None:
+        backups = {
+            "items": [
+                {
+                    "metadata": {"name": "older"},
+                    "spec": {"method": "barmanObjectStore"},
+                    "status": {"phase": "completed", "stoppedAt": "2026-08-22T01:00:00Z"},
+                },
+                {
+                    "metadata": {"name": "newer"},
+                    "spec": {"method": "barmanObjectStore"},
+                    "status": {"phase": "completed", "stoppedAt": "2026-08-22T02:00:00Z"},
+                },
+                {
+                    "metadata": {"name": "running"},
+                    "status": {"phase": "running", "stoppedAt": "2026-08-22T03:00:00Z"},
+                },
+            ]
+        }
+        result = module.latest_completed_backup(
+            backups, datetime(2026, 8, 22, 4, 0, tzinfo=UTC)
+        )
+        self.assertEqual("newer", result["name"])
+        self.assertEqual(2.0, result["age_hours"])
+        self.assertNotIn("data", result)
+
+    def test_common_gate_requires_every_boundary(self) -> None:
+        state = {
+            "node": {"ready": True, "k3s_active": True},
+            "platform_pg": {
+                "ready_instances": 1,
+                "instances": 1,
+                "continuous_archiving": True,
+                "last_backup_succeeded": True,
+            },
+            "openbao": {"initialized": True, "sealed": False},
+            "external_secret_stores": {"a": {"ready": True}, "b": {"ready": True}},
+            "external_secrets": {"a": {"ready": True}, "b": {"ready": True}},
+            "audit_core": {
+                "image_matches_reviewed_digest": True,
+                "ready_replicas": 1,
+                "replicas": 1,
+            },
+        }
+        self.assertTrue(module.automated_common_pass(state))
+        state["openbao"]["sealed"] = True
+        self.assertFalse(module.automated_common_pass(state))
+
+    def test_expected_database_projection_is_exact_key_set(self) -> None:
+        self.assertEqual(
+            {"username", "password", "host", "port", "dbname"},
+            module.EXPECTED_DB_KEYS,
+        )
+
+    def test_secret_description_parser_reads_names_not_values(self) -> None:
+        description = """Name: audit-core-database
+Data
+====
+username:  20 bytes
+password:  36 bytes
+host:      50 bytes
+"""
+        self.assertEqual(
+            {"username", "password", "host"}, module.secret_key_names(description)
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()