import hashlib import importlib.machinery import importlib.util import json import os from pathlib import Path import shutil import subprocess import tarfile import tempfile import unittest from unittest import mock SCRIPT_DIR = Path(__file__).resolve().parent RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy" BUILDER_PATH = SCRIPT_DIR / "build-engine-l2-closed-loop-artifact.mjs" ENGINE_ROOT = SCRIPT_DIR.parent.parent.parent / "NODEDC_ENGINE_INFRA" PATCH_ID = "engine-l2-closed-loop-20991231-999" SOURCE_COMMIT = "dda405cff27977622af0c218abb4d4420c408536" BASELINE_ARTIFACT = ( SCRIPT_DIR.parent / "deploy-artifacts" / "nodedc-engine-provider-authority-diagnostics-20260723-029.tgz" ) def load_runner(): loader = importlib.machinery.SourceFileLoader( "nodedc_engine_l2_closed_loop", 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() class EngineL2ClosedLoopArtifactTest(unittest.TestCase): def require_target_commit(self): current = subprocess.run( ["git", "-C", str(ENGINE_ROOT), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ).stdout.strip() if current != SOURCE_COMMIT: self.skipTest("historical Engine L2 closed-loop source has advanced") def build(self, artifact_dir): self.require_target_commit() environment = os.environ.copy() environment["NODEDC_ENGINE_SOURCE_ROOT"] = str(ENGINE_ROOT) environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir) completed = subprocess.run( ["node", str(BUILDER_PATH), PATCH_ID], check=True, capture_output=True, text=True, env=environment, ) return json.loads(completed.stdout) def load_built_payload(self, root): built = self.build(root / "artifact") artifact = Path(built["artifact"]) manifest, entries, payload = RUNNER.load_artifact( artifact, root / "loaded", ) return built, manifest, entries, payload def materialize_partial_predecessor(self, payload, entries, root): for relative_path in entries: source = payload / relative_path destination = root / relative_path destination.parent.mkdir(parents=True, exist_ok=True) if source.is_dir(): shutil.copytree(source, destination) else: shutil.copy2(source, destination) with tarfile.open(BASELINE_ARTIFACT, "r:gz") as archive: member = archive.extractfile( "payload/nodedc-source/services/node-intelligence/activation.json" ) self.assertIsNotNone(member) (root / RUNNER.ENGINE_L2_CLOSED_LOOP_DESCRIPTOR_REL).write_bytes( member.read() ) shutil.copytree( root / "nodedc-source/dist", root / "nginx-html", ) def test_builder_is_commit_bound_and_byte_deterministic(self): with tempfile.TemporaryDirectory(prefix="nodedc-engine-l2-closed-loop-") as directory: root = Path(directory) first = self.build(root / "first") second = self.build(root / "second") first_artifact = Path(first["artifact"]) second_artifact = Path(second["artifact"]) self.assertEqual(first["sourceCommit"], SOURCE_COMMIT) self.assertEqual(first_artifact.read_bytes(), second_artifact.read_bytes()) self.assertEqual( first["sha256"], hashlib.sha256(first_artifact.read_bytes()).hexdigest(), ) self.assertEqual(first["services"], ["nodedc-backend", "app"]) self.assertEqual(first["mcpVersion"], "0.7.0") def test_failed_030_identity_is_terminal_and_cannot_be_retried(self): with self.assertRaisesRegex( RUNNER.DeployError, "terminal failed", ): RUNNER.reject_terminal_engine_l2_failed_artifact( { "id": RUNNER.ENGINE_L2_CLOSED_LOOP_FAILED_PATCH_ID, }, "0" * 64, ) with self.assertRaisesRegex( RUNNER.DeployError, "terminal failed", ): RUNNER.reject_terminal_engine_l2_failed_artifact( {"id": "different-successor"}, RUNNER.ENGINE_L2_CLOSED_LOOP_FAILED_ARTIFACT_SHA256, ) RUNNER.reject_terminal_engine_l2_failed_artifact( {"id": "engine-l2-closed-loop-20260723-031"}, "1" * 64, ) def test_artifact_is_a_safe_exact_engine_overlay(self): with tempfile.TemporaryDirectory(prefix="nodedc-engine-l2-overlay-") as directory: root = Path(directory) built = self.build(root / "artifact") artifact = Path(built["artifact"]) with tarfile.open(artifact, "r:gz") as archive: members = archive.getmembers() for member in members: self.assertIn(member.type, (tarfile.REGTYPE, tarfile.DIRTYPE)) self.assertEqual(member.uid, 0) self.assertEqual(member.gid, 0) self.assertEqual(member.mtime, 0) self.assertNotIn(".DS_Store", member.name) self.assertNotIn("__MACOSX", member.name) archive.extractall(root / "extract", filter="data") manifest, entries, payload = RUNNER.load_artifact( artifact, root / "loaded", ) self.assertEqual(manifest["component"], "engine") self.assertEqual(manifest["type"], "app-overlay") self.assertEqual(tuple(entries), tuple(built["entries"])) self.assertEqual( RUNNER.component_services("engine", entries), ("nodedc-backend", "app"), ) self.assertEqual( RUNNER.component_healthchecks( "engine", entries, ("nodedc-backend", "app"), ), ( "http://127.0.0.1:8080/", "http://127.0.0.1:3001/health", ), ) self.assertTrue(RUNNER.component_publish_dist("engine", entries)) self.assertEqual(RUNNER.component_builds("engine", entries), ()) payload_files = { path.relative_to(payload).as_posix() for path in payload.rglob("*") if path.is_file() } self.assertEqual(payload_files, set(built["targetSha256"])) for relative_path, expected in built["targetSha256"].items(): self.assertEqual( hashlib.sha256((payload / relative_path).read_bytes()).hexdigest(), expected, ) self.assertTrue(all("/tests/" not in path for path in payload_files)) self.assertTrue(all("/data/" not in path for path in payload_files)) self.assertTrue(all("/storage/" not in path for path in payload_files)) self.assertTrue(all("/logs/" not in path for path in payload_files)) self.assertTrue(all(not path.endswith(".env") for path in payload_files)) def test_preflight_accepts_only_the_exact_failed_030_partial_state(self): with tempfile.TemporaryDirectory(prefix="nodedc-engine-l2-preflight-") as directory: workspace = Path(directory) _built, _manifest, entries, payload = self.load_built_payload( workspace / "build" ) live = workspace / "live" self.materialize_partial_predecessor(payload, entries, live) recovery = workspace / RUNNER.ENGINE_L2_CLOSED_LOOP_RECOVERY_BACKUP_ID backend = { "mode": "verified-derived-retry", "container_id": "a" * 64, } with ( mock.patch.object(RUNNER, "component_root", return_value=live), mock.patch.object( RUNNER, "validate_engine_l2_closed_loop_recovery_evidence", return_value=recovery, ), mock.patch.object( RUNNER, "validate_installed_engine_node_intelligence_source", ) as validate_source, mock.patch.object( RUNNER, "preflight_engine_credential_backend_runtime", return_value=backend, ) as validate_backend, mock.patch.object( RUNNER, "engine_compose_service_container_id_for_gateway", return_value="b" * 64, ) as app_lookup, ): result = RUNNER.preflight_engine_l2_closed_loop_predecessor( payload ) self.assertEqual( result["mode"], "failed-030-partial-source-reconciliation", ) self.assertEqual(result["backend_container_id"], "a" * 64) self.assertEqual(result["app_container_id"], "b" * 64) self.assertEqual( result["target_gateway_sha256"], RUNNER.ENGINE_L2_CLOSED_LOOP_TARGET_SHA256[ RUNNER.ENGINE_NODE_INTELLIGENCE_GATEWAY_REL ], ) self.assertEqual( validate_source.call_args.kwargs[ "expected_gateway_sha256" ], RUNNER.ENGINE_L2_CLOSED_LOOP_PARTIAL_SHA256[ RUNNER.ENGINE_NODE_INTELLIGENCE_GATEWAY_REL ], ) self.assertEqual( validate_backend.call_args.kwargs[ "expected_node_intelligence_gateway_sha256" ], RUNNER.ENGINE_L2_CLOSED_LOOP_PARTIAL_SHA256[ RUNNER.ENGINE_NODE_INTELLIGENCE_GATEWAY_REL ], ) app_lookup.assert_called_once_with( "app", RUNNER.ENGINE_L2_CLOSED_LOOP_PARTIAL_SHA256[ RUNNER.ENGINE_NODE_INTELLIGENCE_GATEWAY_REL ], ) index_path = live / "nodedc-source/dist/index.html" index_path.write_bytes(index_path.read_bytes() + b"\n") with ( mock.patch.object(RUNNER, "component_root", return_value=live), mock.patch.object( RUNNER, "validate_engine_l2_closed_loop_recovery_evidence", return_value=recovery, ), mock.patch.object( RUNNER, "preflight_engine_credential_backend_runtime", ) as unreachable_backend, ): with self.assertRaisesRegex( RUNNER.DeployError, "partial predecessor drift", ): RUNNER.preflight_engine_l2_closed_loop_predecessor(payload) unreachable_backend.assert_not_called() def test_reconciliation_rollback_uses_failed_030_backup_and_restores_runtime(self): entries = RUNNER.ENGINE_L2_CLOSED_LOOP_ARTIFACT_ENTRIES services = ("nodedc-backend", "app") with tempfile.TemporaryDirectory(prefix="nodedc-engine-l2-rollback-") as directory: root = Path(directory) candidate = root / "candidate-backup" recovery = root / "recovery-backup" candidate_entries = [*entries, "nginx-html"] original_entries = [ rel for rel in entries if rel != RUNNER.ENGINE_L2_CLOSED_LOOP_DESCRIPTOR_REL ] original_entries.append("nginx-html") def backup_paths(path): if path.parent == candidate: return candidate_entries if path.name == "existing-files.txt" else [] if path.parent == recovery: if path.name == "existing-files.txt": return [ rel for rel in original_entries if rel != "nodedc-source/server/l2/graphRepository.js" ] return ["nodedc-source/server/l2/graphRepository.js"] raise AssertionError(path) with ( mock.patch.object( RUNNER, "validate_engine_l2_closed_loop_recovery_evidence", return_value=recovery, ), mock.patch.object( RUNNER, "read_backup_path_list", side_effect=backup_paths, ), mock.patch.object( RUNNER, "restore_platform_overlay", side_effect=(len(candidate_entries), len(original_entries)), ) as restore, mock.patch.object( RUNNER, "validate_engine_l2_closed_loop_stable_source", ) as validate_stable, mock.patch.object(RUNNER, "run_component_runtime") as run_runtime, mock.patch.object(RUNNER, "healthcheck_compose_service") as check_service, mock.patch.object( RUNNER, "component_healthchecks", return_value=( "http://127.0.0.1:8080/", "http://127.0.0.1:3001/health", ), ), mock.patch.object(RUNNER, "healthcheck_url") as check_url, mock.patch.object( RUNNER, "preflight_engine_credential_backend_runtime", return_value={"mode": "verified-derived-retry"}, ), ): result = RUNNER.rollback_engine_l2_closed_loop_reconciliation( root, candidate, entries, "20991231-235959", True, services, ) self.assertTrue(result.startswith("stable-source+runtime-restored:")) self.assertEqual( [call.args[1] for call in restore.call_args_list], [candidate, recovery], ) validate_stable.assert_called_once_with(root) run_runtime.assert_called_once_with( "engine", tuple(RUNNER.ENGINE_L2_CLOSED_LOOP_STABLE_SHA256), services, ) self.assertEqual( [call.args for call in check_service.call_args_list], [("engine", "nodedc-backend"), ("engine", "app")], ) self.assertEqual(check_url.call_count, 2) if __name__ == "__main__": unittest.main(verbosity=2)