#!/usr/bin/env python3 import hashlib import importlib.machinery import importlib.util import json import subprocess import tarfile import tempfile import unittest from contextlib import ExitStack from pathlib import Path from unittest import mock SCRIPT_DIR = Path(__file__).resolve().parent RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy" def load_runner(): loader = importlib.machinery.SourceFileLoader( "nodedc_module_foundry_runtime_recovery", str(RUNNER_PATH), ) spec = importlib.util.spec_from_loader(loader.name, loader) module = importlib.util.module_from_spec(spec) loader.exec_module(module) return module RUNNER = load_runner() def runtime(container_char="a", image_char="b"): return { "schemaVersion": "nodedc.module-foundry.runtime-inventory.v1", "composeProject": "nodedc-module-foundry", "service": RUNNER.MODULE_FOUNDRY_SERVICE, "containerId": container_char * 64, "imageId": f"sha256:{image_char * 64}", "imageRef": "nodedc-module-foundry-nodedc-module-foundry:latest", "health": "healthy", } class ModuleFoundryRuntimeRecoveryTest(unittest.TestCase): def test_healthchecks_wait_for_compose_health_before_http(self): events = [] with ( mock.patch.object( RUNNER, "healthcheck_compose_service", side_effect=lambda component, service: events.append( ("compose", component, service) ), ), mock.patch.object( RUNNER, "component_healthchecks", return_value=("http://foundry/healthz",), ), mock.patch.object( RUNNER, "healthcheck_url", side_effect=lambda check: events.append(("http", check)), ), ): RUNNER.run_healthchecks( "module-foundry", ("package.json",), (RUNNER.MODULE_FOUNDRY_SERVICE,), ) self.assertEqual( events, [ ( "compose", "module-foundry", RUNNER.MODULE_FOUNDRY_SERVICE, ), ("http", "http://foundry/healthz"), ], ) def test_build_retries_before_any_runtime_recreate(self): first_failure = subprocess.CalledProcessError(17, ["docker", "compose"]) completed = subprocess.CompletedProcess([], 0) with ( mock.patch.object( RUNNER, "component_compose_root", return_value=Path("/compose"), ), mock.patch.object( RUNNER, "compose_base_cmd", return_value=["docker", "compose"], ), mock.patch.object( RUNNER.subprocess, "run", side_effect=[first_failure, completed], ) as run, mock.patch.object(RUNNER.time, "sleep") as sleep, mock.patch.object(RUNNER, "run_module_foundry_compose_up") as up, ): RUNNER.run_module_foundry_compose( (RUNNER.MODULE_FOUNDRY_SERVICE,) ) expected = [ "docker", "compose", "build", RUNNER.MODULE_FOUNDRY_SERVICE, ] self.assertEqual(run.call_count, 2) self.assertEqual(run.call_args_list[0].args[0], expected) self.assertEqual(run.call_args_list[1].args[0], expected) sleep.assert_called_once_with(2) up.assert_called_once_with((RUNNER.MODULE_FOUNDRY_SERVICE,)) def test_runtime_recreate_uses_built_image_without_build_or_pull(self): completed = subprocess.CompletedProcess([], 0) with ( mock.patch.object( RUNNER, "component_compose_root", return_value=Path("/compose"), ), mock.patch.object( RUNNER, "compose_base_cmd", return_value=["docker", "compose"], ), mock.patch.object( RUNNER.subprocess, "run", return_value=completed, ) as run, ): RUNNER.run_module_foundry_compose_up( (RUNNER.MODULE_FOUNDRY_SERVICE,) ) up = run.call_args_list[0].args[0] self.assertEqual( up, [ "docker", "compose", "up", "-d", "--force-recreate", "--pull", "never", "--no-deps", RUNNER.MODULE_FOUNDRY_SERVICE, ], ) self.assertNotIn("--build", up) self.assertEqual(run.call_args_list[1].args[0], ["docker", "compose", "ps"]) def test_rollback_restores_source_and_keeps_unchanged_runtime(self): with tempfile.TemporaryDirectory( prefix="module-foundry-rollback-unchanged-" ) as directory: work = Path(directory) root = work / "source" backup = work / "backup" tmp = work / "tmp" root.mkdir() backup.mkdir() tmp.mkdir() entries = ("old.txt", "new.txt") (root / "old.txt").write_text("old\n", encoding="utf-8") RUNNER.create_backup(root, backup, entries, False) (root / "old.txt").write_text("candidate\n", encoding="utf-8") (root / "new.txt").write_text("candidate\n", encoding="utf-8") before = runtime() (backup / RUNNER.MODULE_FOUNDRY_RUNTIME_BEFORE_FILE).write_text( json.dumps(before), encoding="utf-8", ) with ( mock.patch.object(RUNNER, "TMP_DIR", tmp), mock.patch.object( RUNNER, "module_foundry_runtime_inventory", return_value=before, ), mock.patch.object(RUNNER, "run_healthchecks") as health, mock.patch.object(RUNNER.subprocess, "run") as process, ): result = RUNNER.rollback_module_foundry_apply( root, backup, entries, "20260809-000000", (RUNNER.MODULE_FOUNDRY_SERVICE,), ) self.assertEqual( result, "source-restored-runtime-unchanged:2", ) self.assertEqual( (root / "old.txt").read_text(encoding="utf-8"), "old\n", ) self.assertFalse((root / "new.txt").exists()) process.assert_not_called() health.assert_called_once() def test_rollback_retags_and_recreates_changed_runtime(self): with tempfile.TemporaryDirectory( prefix="module-foundry-rollback-changed-" ) as directory: work = Path(directory) root = work / "source" backup = work / "backup" tmp = work / "tmp" root.mkdir() backup.mkdir() tmp.mkdir() entries = ("old.txt",) (root / "old.txt").write_text("old\n", encoding="utf-8") RUNNER.create_backup(root, backup, entries, False) (root / "old.txt").write_text("candidate\n", encoding="utf-8") before = runtime() candidate = runtime("c", "d") (backup / RUNNER.MODULE_FOUNDRY_RUNTIME_BEFORE_FILE).write_text( json.dumps(before), encoding="utf-8", ) with ( mock.patch.object(RUNNER, "TMP_DIR", tmp), mock.patch.object( RUNNER, "module_foundry_runtime_inventory", side_effect=[candidate, before], ) as inventory, mock.patch.object( RUNNER, "inspect_local_image", return_value=before["imageId"], ), mock.patch.object(RUNNER, "run_module_foundry_compose_up") as up, mock.patch.object(RUNNER, "run_healthchecks") as health, mock.patch.object(RUNNER.subprocess, "run") as process, ): result = RUNNER.rollback_module_foundry_apply( root, backup, entries, "20260809-000000", (RUNNER.MODULE_FOUNDRY_SERVICE,), ) self.assertEqual(result, "source+runtime-restored:1") self.assertEqual( inventory.call_args_list[0].kwargs, {"required": False, "require_healthy": False}, ) process.assert_called_once_with( [ str(RUNNER.DOCKER), "image", "tag", before["imageId"], before["imageRef"], ], check=True, ) up.assert_called_once_with((RUNNER.MODULE_FOUNDRY_SERVICE,)) health.assert_called_once() def test_candidate_must_replace_both_container_and_image(self): before = runtime() for candidate in ( runtime("a", "c"), runtime("c", "b"), ): with mock.patch.object( RUNNER, "module_foundry_runtime_inventory", return_value=candidate, ): with self.assertRaises(RUNNER.DeployError): RUNNER.validate_module_foundry_candidate_runtime(before) def test_preapply_runtime_rejects_digest_only_image_reference(self): with tempfile.TemporaryDirectory( prefix="module-foundry-runtime-reference-" ) as directory: backup = Path(directory) before = runtime() before["imageRef"] = before["imageId"] (backup / RUNNER.MODULE_FOUNDRY_RUNTIME_BEFORE_FILE).write_text( json.dumps(before), encoding="utf-8", ) with self.assertRaises(RUNNER.DeployError): RUNNER.read_module_foundry_runtime_before(backup) def test_recovery_accepts_only_exact_failed_partial_predecessor(self): with tempfile.TemporaryDirectory( prefix="module-foundry-map-evidence-" ) as directory: work = Path(directory) backups = work / "backups" failed = work / "failed" state = work / "state" tmp = work / "tmp" installed = work / "installed" candidate = work / "candidate" for path in (backups, failed, state, tmp, installed, candidate): path.mkdir() recovery_entries = ( "apps/catalog/src", "scripts", "server/map-grid-persistence.test.mjs", ) missing_entries = ("server/map-grid-persistence.test.mjs",) failed_entries = ( "apps/catalog/src/MapFixturePreview.tsx", "scripts/map-sector-grid.test.mjs", "server/map-grid-persistence.test.mjs", ) candidate_files = { "apps/catalog/src/MapFixturePreview.tsx": "sector-current\n", "apps/catalog/src/CesiumMapRenderer.tsx": "renderer-next\n", "scripts/map-sector-grid.test.mjs": "sector-test-current\n", "scripts/map-grid-lod.test.mjs": "grid-test-next\n", "server/map-grid-persistence.test.mjs": "server-test-next\n", } installed_files = { "apps/catalog/src/MapFixturePreview.tsx": "sector-current\n", "apps/catalog/src/CesiumMapRenderer.tsx": "renderer-old\n", "scripts/map-sector-grid.test.mjs": "sector-test-current\n", "scripts/map-grid-lod.test.mjs": "grid-test-old\n", } for root, values in ( (candidate, candidate_files), (installed, installed_files), ): for rel, value in values.items(): path = root / rel path.parent.mkdir(parents=True, exist_ok=True) path.write_text(value, encoding="utf-8") failed_patch_id = "module-foundry-map-failed-test" failed_artifact_name = "module-foundry-map-failed-test.tgz.1" failed_backup_id = "module-foundry-map-failed-test-backup" recovery_patch_id = "module-foundry-map-recovery-test" failed_message = "test build failed" backup = backups / failed_backup_id backup.mkdir() backup_names = { "existing-files.txt", "files.txt", "manifest.env", "missing-files.txt", "module-foundry-runtime-before.json", "source-before.tgz", } for name in backup_names: (backup / name).write_text(f"{name}\n", encoding="utf-8") backup_hashes = { name: hashlib.sha256((backup / name).read_bytes()).hexdigest() for name in backup_names } archive_source = work / "archive-source" (archive_source / "payload").mkdir(parents=True) (archive_source / "manifest.env").write_text( "id=" f"{failed_patch_id}\n" "component=module-foundry\n" "type=app-overlay\n", encoding="utf-8", ) (archive_source / "files.txt").write_text( "\n".join(failed_entries) + "\n", encoding="utf-8", ) for rel in failed_entries: source = candidate / rel target = archive_source / "payload" / rel target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(source.read_bytes()) failed_artifact = failed / failed_artifact_name with tarfile.open(failed_artifact, "w:gz") as archive: archive.add( archive_source / "manifest.env", arcname="manifest.env", ) archive.add( archive_source / "files.txt", arcname="files.txt", ) archive.add(archive_source / "payload", arcname="payload") failed_sha256 = hashlib.sha256( failed_artifact.read_bytes() ).hexdigest() journal = state / "failed.jsonl" journal.write_text( json.dumps( { "artifact": failed_artifact_name, "backup_id": failed_backup_id, "component": "module-foundry", "id": failed_patch_id, "message": failed_message, "rollback_status": "failed:DeployError", "sha256": failed_sha256, "started_apply": True, "status": "failed", } ) + "\n", encoding="utf-8", ) manifest = { "id": recovery_patch_id, "component": "module-foundry", "type": "app-overlay", } candidate_tree = RUNNER.exact_file_map_sha256( RUNNER.collect_exact_files( candidate, recovery_entries, "candidate", ) ) installed_tree = RUNNER.exact_file_map_sha256( RUNNER.collect_exact_files( installed, recovery_entries[:-1], "installed", ) ) patches = { "BACKUPS_DIR": backups, "FAILED_DIR": failed, "FAILED_STATE_FILE": journal, "TMP_DIR": tmp, "MODULE_FOUNDRY_MAP_RECOVERY_PATCH_ID": recovery_patch_id, "MODULE_FOUNDRY_MAP_RECOVERY_ENTRIES": recovery_entries, "MODULE_FOUNDRY_MAP_RECOVERY_MISSING_PREDECESSOR_ENTRIES": ( missing_entries ), "MODULE_FOUNDRY_MAP_FAILED_INSTALLED_ENTRIES": ( failed_entries[:-1] ), "MODULE_FOUNDRY_MAP_RECOVERY_INSTALLED_TREE_SHA256": ( installed_tree ), "MODULE_FOUNDRY_MAP_RECOVERY_CANDIDATE_TREE_SHA256": ( candidate_tree ), "MODULE_FOUNDRY_MAP_FAILED_ENTRIES": failed_entries, "MODULE_FOUNDRY_MAP_FAILED_PATCH_ID": failed_patch_id, "MODULE_FOUNDRY_MAP_FAILED_ARTIFACT": failed_artifact_name, "MODULE_FOUNDRY_MAP_FAILED_ARTIFACT_SHA256": failed_sha256, "MODULE_FOUNDRY_MAP_FAILED_BACKUP_ID": failed_backup_id, "MODULE_FOUNDRY_MAP_FAILED_MESSAGE": failed_message, "MODULE_FOUNDRY_MAP_FAILED_ROLLBACK_STATUS": ( "failed:DeployError" ), "MODULE_FOUNDRY_MAP_FAILED_BACKUP_SHA256": backup_hashes, } with ExitStack() as stack: for name, value in patches.items(): stack.enter_context(mock.patch.object(RUNNER, name, value)) stack.enter_context( mock.patch.object( RUNNER, "component_root", return_value=installed, ) ) stack.enter_context( mock.patch.object( RUNNER, "module_foundry_runtime_inventory", return_value=runtime(), ) ) result = RUNNER.validate_module_foundry_map_recovery_evidence( manifest, recovery_entries, candidate, ) self.assertEqual( result["mode"], "failed-map-overlay-source+runtime-reconciliation", ) (installed / failed_entries[0]).write_text( "drift\n", encoding="utf-8", ) with self.assertRaises(RUNNER.DeployError): RUNNER.validate_module_foundry_map_recovery_evidence( manifest, recovery_entries, candidate, ) if __name__ == "__main__": unittest.main(verbosity=2)