test(deploy): execute migration audit preflight

infra/deploy-runner/test_device_control_core_migration_replay_audit_artifact.py

@@ -159,6 +159,122 @@ class DeviceControlCoreMigrationReplayAuditArtifactTest(unittest.TestCase):
         ):
             self.assertIn(marker, rendered)
 
+    def test_audit_preflight_resolves_recovery_from_canonical_inbox(self):
+        with tempfile.TemporaryDirectory(
+            prefix="nodedc-control-core-migration-replay-preflight-",
+        ) as directory:
+            root = Path(directory)
+            inbox = root / "inbox"
+            runner_tmp = root / "runner-tmp"
+            source = root / "device-plane"
+            inbox.mkdir()
+            runner_tmp.mkdir()
+            live_migration = (
+                source / RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_014_REL
+            )
+            live_migration.parent.mkdir(parents=True)
+            live_migration.write_bytes(b"migration-014-predecessor")
+            recovery_artifact = (
+                inbox
+                / RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_RECOVERY_ARTIFACT
+            )
+            recovery_artifact.write_bytes(b"reviewed-recovery-044")
+            core = {
+                "containerId": "c" * 64,
+                "imageId": RUNNER.DEVICE_PLANE_CONTROL_CORE_V3_PREAPPLY_IMAGE_ID,
+                "status": "running",
+                "health": "unhealthy",
+                "restartCount": 3,
+            }
+            database = {
+                "invalidCommandKindCount": 0,
+                "triggeringReceiptCount": 0,
+                "constraintValidated": True,
+                "constraintCoversFinalKinds": True,
+                "hostTelemetryTableAbsent": True,
+                "recovery044Ready": False,
+            }
+            failure_evidence = {
+                "firstBackup": root / "first-backup",
+                "secondBackup": root / "second-backup",
+            }
+            recovery_manifest = {
+                "id": RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_RECOVERY_PATCH_ID,
+                "component": "device-plane",
+                "type": "app-overlay",
+            }
+
+            with (
+                mock.patch.object(RUNNER, "INBOX", inbox),
+                mock.patch.object(RUNNER, "TMP_DIR", runner_tmp),
+                mock.patch.object(
+                    RUNNER,
+                    "DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_RECOVERY_ARTIFACT_SHA256",
+                    hashlib.sha256(recovery_artifact.read_bytes()).hexdigest(),
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "DEVICE_PLANE_CONTROL_CORE_MIGRATION_014_PREDECESSOR_SHA256",
+                    hashlib.sha256(live_migration.read_bytes()).hexdigest(),
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "validate_device_plane_control_core_migration_replay_audit_payload",
+                    return_value=(
+                        RUNNER.expected_device_plane_control_core_migration_replay_audit_descriptor()
+                    ),
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "validate_device_plane_control_core_double_failure_evidence",
+                    return_value=failure_evidence,
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "load_artifact",
+                    return_value=(
+                        recovery_manifest,
+                        RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_RECOVERY_ENTRIES,
+                        root / "recovery-payload",
+                    ),
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "validate_device_plane_control_core_migration_replay_recovery_payload",
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "load_state",
+                    return_value=[],
+                ) as load_state,
+                mock.patch.object(RUNNER, "component_root", return_value=source),
+                mock.patch.object(
+                    RUNNER,
+                    "validate_device_plane_control_core_migration_replay_preserved_runtime",
+                    return_value={"current": {"services": []}, "core": core},
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "collect_device_plane_control_core_migration_replay_database_evidence",
+                    return_value=database,
+                ) as collect_database,
+            ):
+                evidence = (
+                    RUNNER.validate_device_plane_control_core_migration_replay_audit_evidence(
+                        root / "audit-payload"
+                    )
+                )
+
+        self.assertEqual(evidence["recoveryArtifact"], recovery_artifact)
+        self.assertEqual(evidence["database"], database)
+        self.assertEqual(
+            [call.args for call in load_state.call_args_list],
+            [(RUNNER.STATE_FILE,), (RUNNER.FAILED_STATE_FILE,)],
+        )
+        collect_database.assert_called_once_with(
+            enforce_recovery_invariants=False,
+        )
+
     def test_plan_and_apply_paths_preserve_plan_only_boundary(self):
         plan_source = inspect.getsource(RUNNER.plan_artifact)
         apply_source = inspect.getsource(RUNNER.apply_artifact)