#!/usr/bin/env python3 import hashlib import importlib.machinery import importlib.util import json import os import shutil import subprocess import tarfile import tempfile import unittest from pathlib import Path from unittest import mock SCRIPT_DIR = Path(__file__).resolve().parent RECOVERY_BUILDER = ( SCRIPT_DIR / "build-device-plane-foundation-recovery-artifact.mjs" ) FOUNDATION_BUILDER = SCRIPT_DIR / "build-device-plane-artifact.mjs" PREDECESSOR_COMPOSE = ( SCRIPT_DIR / "fixtures/device-plane-foundation-internal-only-v1.yml" ) RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy" RECOVERY_ENTRIES = [ ".dockerignore", "package.json", "package-lock.json", "docker-compose.device-plane.yml", "packages/device-protocol-contract", "packages/arusnavi-b2-adapter", "services/device-control-core", "services/device-gateway", "deployment/device-plane-foundation-recovery-v1.json", ] def load_runner(): loader = importlib.machinery.SourceFileLoader( "nodedc_device_plane_recovery_runner_under_test", 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 DevicePlaneFoundationRecoveryArtifactTest(unittest.TestCase): def build(self, builder, artifact_dir, patch_id): environment = os.environ.copy() environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir) result = subprocess.run( ["node", str(builder), patch_id], check=True, capture_output=True, text=True, env=environment, ) return json.loads(result.stdout) def test_recovery_artifact_is_source_only_exact_and_deterministic(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-plane-recovery-artifact-", ) as directory: artifact_dir = Path(directory) first = self.build( RECOVERY_BUILDER, artifact_dir, "device-plane-foundation-recovery-unit-002", ) artifact = Path(first["artifact"]) first_bytes = artifact.read_bytes() second = self.build( RECOVERY_BUILDER, artifact_dir, "device-plane-foundation-recovery-unit-002", ) second_bytes = Path(second["artifact"]).read_bytes() self.assertEqual(first["component"], "device-plane") self.assertEqual(first["entries"], RECOVERY_ENTRIES) self.assertEqual(first["build"], []) self.assertEqual(first["services"], []) self.assertEqual( first["transition"], "failed-foundation-live-runtime-adoption", ) self.assertEqual(first_bytes, second_bytes) self.assertEqual( first["sha256"], hashlib.sha256(first_bytes).hexdigest(), ) with tarfile.open(artifact, "r:gz") as archive: names = {member.name for member in archive.getmembers()} files = ( archive.extractfile("files.txt") .read() .decode("utf-8") .splitlines() ) descriptor = json.loads( archive.extractfile( "payload/deployment/" "device-plane-foundation-recovery-v1.json" ) .read() .decode("utf-8") ) self.assertEqual(files, RECOVERY_ENTRIES) self.assertIn( "payload/services/device-control-core/src/server.mjs", names, ) self.assertIn( "payload/services/device-gateway/src/server.mjs", names, ) self.assertFalse(any( "/test/" in name or "/node_modules/" in name or Path(name).name.startswith(".env") or name.startswith("payload/runtime/") or name.startswith("payload/secrets/") for name in names )) self.assertEqual( descriptor, RUNNER.expected_device_plane_foundation_recovery_descriptor(), ) self.assertEqual( RUNNER.component_services( "device-plane", tuple(RECOVERY_ENTRIES), ), (), ) self.assertEqual( RUNNER.component_builds( "device-plane", tuple(RECOVERY_ENTRIES), ), (), ) def test_preflight_requires_exact_failed_evidence_partial_source_and_runtime( self, ): with tempfile.TemporaryDirectory( prefix="nodedc-device-plane-recovery-preflight-", ) as directory: workspace = Path(directory) artifacts = workspace / "artifacts" failed_root = workspace / "failed" backups_root = workspace / "backups" state_root = workspace / "state" temp_root = workspace / "tmp" live_root = workspace / "live" for path in ( artifacts, failed_root, backups_root, state_root, temp_root, live_root / "deployment", ): path.mkdir(parents=True, exist_ok=True) failed_build = self.build( FOUNDATION_BUILDER, artifacts, RUNNER.DEVICE_PLANE_FOUNDATION_FAILED_PATCH_ID, ) self.assertEqual( failed_build["sha256"], hashlib.sha256( Path(failed_build["artifact"]).read_bytes() ).hexdigest(), ) failed_artifact = ( failed_root / RUNNER.DEVICE_PLANE_FOUNDATION_FAILED_ARTIFACT ) shutil.copy2(failed_build["artifact"], failed_artifact) recovery_build = self.build( RECOVERY_BUILDER, artifacts, "device-plane-foundation-recovery-unit-002", ) recovery_extract = workspace / "recovery-extract" recovery_extract.mkdir() _manifest, entries, payload = RUNNER.load_artifact( Path(recovery_build["artifact"]), recovery_extract, ) self.assertEqual(entries, RECOVERY_ENTRIES) backup = ( backups_root / RUNNER.DEVICE_PLANE_FOUNDATION_RECOVERY_BACKUP_ID ) backup.mkdir() for name in ( "manifest.env", "files.txt", "existing-files.txt", "missing-files.txt", "source-before.tgz", ): (backup / name).write_text( f"fixture:{name}\n", encoding="utf-8", ) backup_hashes = { name: hashlib.sha256((backup / name).read_bytes()).hexdigest() for name in ( "manifest.env", "files.txt", "existing-files.txt", "missing-files.txt", "source-before.tgz", ) } (state_root / "failed.jsonl").write_text( json.dumps({ "artifact": RUNNER.DEVICE_PLANE_FOUNDATION_FAILED_ARTIFACT, "backup_id": RUNNER.DEVICE_PLANE_FOUNDATION_RECOVERY_BACKUP_ID, "component": "device-plane", "id": RUNNER.DEVICE_PLANE_FOUNDATION_FAILED_PATCH_ID, "message": "healthcheck failed for " "http://127.0.0.1:18120/healthz: " "", "rollback_status": "failed:DeployError", "sha256": failed_build["sha256"], "started_apply": True, "status": "failed", }) + "\n", encoding="utf-8", ) source_root = SCRIPT_DIR.parent.parent / "device-plane" shutil.copy2( PREDECESSOR_COMPOSE, live_root / "docker-compose.device-plane.yml", ) shutil.copy2( source_root / "deployment/device-postgres-bootstrap-v1.json", live_root / "deployment/device-postgres-bootstrap-v1.json", ) runtime = { service: { "containerId": service, "imageId": RUNNER.DEVICE_PLANE_FOUNDATION_RECOVERY_IMAGE_IDS[ service ], "health": "healthy", "restartCount": 0, } for service in ( "device-control-core", "device-gateway", "device-postgres", ) } with ( mock.patch.object(RUNNER, "BACKUPS_DIR", backups_root), mock.patch.object(RUNNER, "FAILED_DIR", failed_root), mock.patch.object( RUNNER, "FAILED_STATE_FILE", state_root / "failed.jsonl", ), mock.patch.object(RUNNER, "TMP_DIR", temp_root), mock.patch.object( RUNNER, "component_root", return_value=live_root, ), mock.patch.object( RUNNER, "DEVICE_PLANE_FOUNDATION_FAILED_ARTIFACT_SHA256", failed_build["sha256"], ), mock.patch.dict( RUNNER.DEVICE_PLANE_FOUNDATION_RECOVERY_BACKUP_SHA256, backup_hashes, clear=True, ), mock.patch.object( RUNNER, "validate_device_plane_foundation_runtime", return_value=runtime, ), ): result = ( RUNNER .validate_device_plane_foundation_recovery_evidence( payload ) ) self.assertEqual( result["mode"], "failed-foundation-live-runtime-adoption", ) self.assertEqual(result["runtime"], runtime) def test_recovery_health_acceptance_never_mutates_runtime(self): entries = tuple(RECOVERY_ENTRIES) checks = ("core-health", "gateway-health") with ( mock.patch.object( RUNNER, "healthcheck_compose_service", ) as compose_health, mock.patch.object( RUNNER, "component_healthchecks", return_value=checks, ), mock.patch.object(RUNNER, "healthcheck_url") as url_health, mock.patch.object( RUNNER, "validate_device_plane_foundation_installed_source", ) as source_acceptance, mock.patch.object( RUNNER, "validate_device_plane_foundation_runtime", ) as runtime_acceptance, mock.patch.object(RUNNER, "run_compose") as compose_mutation, mock.patch.object(RUNNER, "run_build") as build_mutation, ): RUNNER.run_healthchecks("device-plane", entries, ()) self.assertEqual( [call.args for call in compose_health.call_args_list], [ ("device-plane", "device-control-core"), ("device-plane", "device-gateway"), ("device-plane", "device-postgres"), ], ) self.assertEqual( [call.args[0] for call in url_health.call_args_list], list(checks), ) source_acceptance.assert_called_once_with() runtime_acceptance.assert_called_once_with() compose_mutation.assert_not_called() build_mutation.assert_not_called() def test_recovery_runtime_phase_has_no_runtime_mutation(self): entries = tuple(RECOVERY_ENTRIES) with ( mock.patch.object( RUNNER, "prepare_component_runtime", ) as prepare, mock.patch.object(RUNNER, "run_compose") as compose_mutation, mock.patch.object(RUNNER, "run_build") as build_mutation, ): RUNNER.run_component_runtime("device-plane", entries, ()) prepare.assert_not_called() compose_mutation.assert_not_called() build_mutation.assert_not_called() if __name__ == "__main__": unittest.main(verbosity=2)