#!/usr/bin/env python3 import hashlib import importlib.machinery import importlib.util import json import os import subprocess import tarfile import tempfile import unittest from pathlib import Path from unittest import mock SCRIPT_DIR = Path(__file__).resolve().parent PLATFORM_ROOT = SCRIPT_DIR.parent.parent RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy" def load_runner(): loader = importlib.machinery.SourceFileLoader( "nodedc_device_manager_deploy_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() LAUNCHER_HUB_SERVICE_TRUST_UI_ENTRIES = ( "src/app/LauncherApp.tsx", "src/styles/globals.css", "src/widgets/admin-overlay/AdminOverlay.tsx", ) def healthy_device_plane_inventory(*, include_manager=False): services = [ ("device-control-core", "a"), ("device-gateway", "b"), ("device-postgres", "c"), ] if include_manager: services.append(("device-manager", "d")) return { "schemaVersion": "nodedc.device-plane.runtime-inventory.v1", "composeProject": "nodedc-device-plane", "services": [ { "service": service, "containerId": character * 64, "imageId": f"sha256:{character * 64}", "status": "running", "running": True, "health": "healthy", "restartCount": 0, } for service, character in services ], } def device_manager_release_descriptor( release_id="device-manager-release-unit-001", *, action="activate", predecessor_kind="reconciliation", predecessor_patch="device-manager-reconciliation-unit-001", predecessor_sha="a" * 64, ): return { "schemaVersion": "nodedc.device-plane.device-manager-release.v2", "releaseId": release_id, "action": action, "predecessor": { "kind": predecessor_kind, "patchId": predecessor_patch, "artifactSha256": predecessor_sha, }, **RUNNER.expected_device_plane_manager_release_v2_boundaries(), } def device_manager_release_v1_descriptor( release_id="device-manager-release-v1-unit-001", *, action="upgrade", predecessor_kind="release", predecessor_patch="device-manager-release-v1-unit-000", predecessor_sha="a" * 64, ): return { "schemaVersion": "nodedc.device-plane.device-manager-release.v1", "releaseId": release_id, "action": action, "predecessor": { "kind": predecessor_kind, "patchId": predecessor_patch, "artifactSha256": predecessor_sha, }, **RUNNER.expected_device_plane_manager_release_v1_boundaries(), } class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase): def build(self, script, patch_id, artifact_dir): environment = os.environ.copy() environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir) completed = subprocess.run( ["node", str(SCRIPT_DIR / script), patch_id], cwd=PLATFORM_ROOT, env=environment, check=True, capture_output=True, text=True, ) return json.loads(completed.stdout) def assert_deterministic_artifact(self, script, patch_id, expected_entries): with tempfile.TemporaryDirectory(prefix="nodedc-device-manager-artifact-") as directory: root = Path(directory) first = self.build(script, patch_id, root / "first") second = self.build(script, patch_id, root / "second") first_artifact = Path(first["artifact"]) second_artifact = Path(second["artifact"]) self.assertEqual(first_artifact.read_bytes(), second_artifact.read_bytes()) self.assertEqual( first["sha256"], hashlib.sha256(first_artifact.read_bytes()).hexdigest(), ) self.assertEqual(tuple(first["entries"]), tuple(expected_entries)) extracted = root / "extracted" extracted.mkdir() manifest, entries, payload = RUNNER.load_artifact(first_artifact, extracted) self.assertEqual(tuple(entries), tuple(expected_entries)) with tarfile.open(first_artifact, "r:gz") as archive: members = archive.getmembers() names = [member.name for member in members] bytes_joined = b"\n".join( archive.extractfile(member).read() for member in members if member.isfile() ) self.assertFalse(any(Path(name).name.startswith("._") for name in names)) self.assertFalse(any("/node_modules/" in name or "/.git/" in name for name in names)) self.assertNotIn(b"-----BEGIN PRIVATE KEY-----", bytes_joined) return manifest, entries, names, first def test_platform_hub_trust_artifact_is_exact_and_build_free(self): manifest, entries, _names, result = self.assert_deterministic_artifact( "build-platform-device-core-hub-trust-artifact.mjs", "platform-device-core-hub-trust-unit-001", RUNNER.PLATFORM_DEVICE_CORE_HUB_TRUST_ENTRIES, ) self.assertEqual(manifest["component"], "platform") self.assertEqual(RUNNER.component_services("platform", entries), ("launcher",)) self.assertEqual(RUNNER.component_builds("platform", entries), ()) self.assertEqual(result["services"], ["launcher"]) def test_launcher_session_artifact_is_exact(self): manifest, entries, _names, result = self.assert_deterministic_artifact( "build-launcher-device-core-artifact.mjs", "launcher-device-core-session-unit-001", RUNNER.LAUNCHER_DEVICE_CORE_SESSION_ENTRIES, ) self.assertEqual(manifest["component"], "launcher") self.assertEqual(RUNNER.component_services("launcher", entries), ("launcher",)) self.assertEqual(len(RUNNER.component_builds("launcher", entries)), 1) self.assertEqual(result["services"], ["launcher"]) def test_launcher_hub_service_trust_ui_artifact_is_exact(self): manifest, entries, _names, result = self.assert_deterministic_artifact( "build-launcher-hub-service-trust-ui-artifact.mjs", "launcher-hub-service-trust-ui-unit-001", LAUNCHER_HUB_SERVICE_TRUST_UI_ENTRIES, ) self.assertEqual(manifest["component"], "launcher") self.assertEqual(RUNNER.component_services("launcher", entries), ("launcher",)) self.assertEqual(len(RUNNER.component_builds("launcher", entries)), 1) self.assertEqual(result["services"], ["launcher"]) def test_device_manager_artifact_selects_only_core_and_manager(self): manifest, entries, names, result = self.assert_deterministic_artifact( "build-device-manager-control-plane-artifact.mjs", "device-manager-control-plane-unit-001", RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V1_SUCCESSOR_ENTRIES, ) self.assertEqual(manifest["component"], "device-plane") self.assertEqual( RUNNER.component_services("device-plane", entries), ("device-control-core", "device-manager"), ) builds = RUNNER.component_builds("device-plane", entries) self.assertEqual(len(builds), 2) self.assertIn(RUNNER.DEVICE_PLANE_CONTROL_CORE_IMAGE, builds[0][1]) self.assertIn(RUNNER.DEVICE_PLANE_MANAGER_IMAGE, builds[1][1]) self.assertIn("payload/services/device-manager/dist/index.html", names) self.assertIn( "payload/services/device-manager/server/device-manager-server.mjs", names, ) self.assertIn( "payload/services/device-control-core/src/credential-reference.mjs", names, ) self.assertIn( "payload/packages/device-edge-channel-contract/src/index.mjs", names, ) self.assertIn("packages/device-edge-channel-contract", entries) self.assertIn( "payload/services/device-control-core/package-lock.json", names, ) self.assertIn( "payload/services/device-control-core/src/device-gateway-core-runtime.mjs", names, ) self.assertFalse(any( name.startswith("payload/packages/device-adapter-runtime/") or name.startswith("payload/packages/device-adapter-catalog/") or name.startswith("payload/services/device-gateway-core/") for name in names )) core_dockerfile = ( PLATFORM_ROOT / "device-plane/services/device-control-core/Dockerfile" ).read_text(encoding="utf-8") self.assertIn( "COPY services/device-control-core/package.json " "services/device-control-core/package-lock.json ./", core_dockerfile, ) self.assertNotIn("COPY package.json package-lock.json ./", core_dockerfile) self.assertNotIn("COPY services/device-gateway-core", core_dockerfile) self.assertFalse(any(name.endswith(".test.mjs") for name in names)) self.assertEqual(result["services"], ["device-control-core", "device-manager"]) self.assertNotIn("device-postgres", result["services"]) self.assertIn("docker-compose.device-manager.yml", entries) self.assertNotIn("docker-compose.device-plane.yml", entries) compose = ( PLATFORM_ROOT / "device-plane/docker-compose.device-manager.yml" ).read_text(encoding="utf-8") for forbidden in ( "DEVICE_EDGE_CHANNEL_", "device-edge-channel/", "name: nodedc-device-plane-egress", ): self.assertNotIn(forbidden, compose) self.assertNotIn("PRIVATE KEY", compose) checks = RUNNER.component_healthchecks("device-plane", entries, tuple(result["services"])) self.assertEqual(checks[0]["expected_json"]["managementApi"], "enabled") self.assertEqual(checks[0]["expected_json"]["discoveryIngest"], "enabled") def test_edge_core_channel_bootstrap_is_core_only_and_secret_free(self): manifest, entries, names, result = self.assert_deterministic_artifact( "build-device-edge-core-channel-bootstrap-artifact.mjs", "device-edge-core-channel-bootstrap-unit-001", RUNNER.DEVICE_PLANE_EDGE_CORE_CHANNEL_BOOTSTRAP_ENTRIES, ) self.assertEqual(manifest["component"], "device-plane") self.assertEqual( RUNNER.component_services("device-plane", entries), ("device-control-core",), ) builds = RUNNER.component_builds("device-plane", entries) self.assertEqual(len(builds), 1) self.assertIn(RUNNER.DEVICE_PLANE_CONTROL_CORE_IMAGE, builds[0][1]) self.assertEqual(result["services"], ["device-control-core"]) self.assertFalse(any( name.startswith("payload/services/device-manager/") or name.startswith("payload/services/device-gateway/") for name in names )) self.assertIn( "payload/packages/device-edge-channel-contract/src/index.mjs", names, ) self.assertIn( "payload/docker-compose.device-edge-core-channel.yml", names, ) checks = RUNNER.component_healthchecks( "device-plane", entries, tuple(result["services"]), ) self.assertEqual(checks[0]["expected_json"]["managementApi"], "enabled") self.assertEqual(checks[0]["expected_json"]["commandTransport"], "disabled") def test_release_v2_keeps_release_v1_predecessor_contract_immutable(self): predecessor = device_manager_release_v1_descriptor( release_id="device-manager-release-20260811-010", ) candidate = device_manager_release_descriptor( release_id="device-manager-release-20260811-012", action="upgrade", predecessor_kind="release", predecessor_patch="device-manager-release-20260811-010", predecessor_sha="d" * 64, ) self.assertIs( RUNNER.validate_device_plane_manager_release_descriptor( predecessor, schema_version=( "nodedc.device-plane.device-manager-release.v1" ), boundaries=( RUNNER.expected_device_plane_manager_release_v1_boundaries() ), expected_release_id="device-manager-release-20260811-010", ), predecessor, ) self.assertIs( RUNNER.validate_device_plane_manager_release_descriptor( candidate, schema_version=( "nodedc.device-plane.device-manager-release.v2" ), boundaries=( RUNNER.expected_device_plane_manager_release_v2_boundaries() ), expected_release_id="device-manager-release-20260811-012", ), candidate, ) with self.assertRaisesRegex( RUNNER.DeployError, r"schema=.*release\.v2 missing=edgeChannel,edgeChannelEgress," r"edgeChannelIdentity extra=none", ): RUNNER.validate_device_plane_manager_release_descriptor( predecessor, schema_version=( "nodedc.device-plane.device-manager-release.v2" ), boundaries=( RUNNER.expected_device_plane_manager_release_v2_boundaries() ), ) def test_installed_manager_compose_follows_release_generation(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-manager-compose-generation-", ) as directory: root = Path(directory) deployment = root / "deployment" deployment.mkdir() compose = root / RUNNER.DEVICE_PLANE_MANAGER_COMPOSE_REL compose.write_text("services: {}\n", encoding="utf-8") v1 = root / RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_REL v2 = root / RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V2_REL v1.write_text( json.dumps(device_manager_release_v1_descriptor()), encoding="utf-8", ) with ( mock.patch.object(RUNNER, "DEVICE_PLANE_ROOT", root), mock.patch.dict( RUNNER.COMPONENTS["device-plane"], {"compose_files": ()}, ), mock.patch.object( RUNNER, "sha256_file", return_value=( RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V1_COMPOSE_SHA256 ), ), ): self.assertEqual( RUNNER.component_compose_files("device-plane"), (compose,), ) v2.write_text( json.dumps(device_manager_release_descriptor()), encoding="utf-8", ) with ( mock.patch.object(RUNNER, "DEVICE_PLANE_ROOT", root), mock.patch.dict( RUNNER.COMPONENTS["device-plane"], {"compose_files": ()}, ), mock.patch.object( RUNNER, "sha256_file", return_value=( RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V2_COMPOSE_SHA256 ), ), ): self.assertEqual( RUNNER.component_compose_files("device-plane"), (compose,), ) with ( mock.patch.object(RUNNER, "DEVICE_PLANE_ROOT", root), mock.patch.dict( RUNNER.COMPONENTS["device-plane"], {"compose_files": ()}, ), mock.patch.object( RUNNER, "sha256_file", return_value="0" * 64, ), ): with self.assertRaisesRegex( RUNNER.DeployError, "installed Device Manager Compose drift detected", ): RUNNER.component_compose_files("device-plane") def test_public_route_artifact_is_last_and_proxy_only(self): manifest, entries, _names, result = self.assert_deterministic_artifact( "build-platform-device-manager-route-artifact.mjs", "platform-device-manager-route-unit-001", RUNNER.PLATFORM_DEVICE_MANAGER_PUBLIC_ROUTE_ENTRIES, ) self.assertEqual(manifest["component"], "platform") self.assertEqual(RUNNER.component_services("platform", entries), ("reverse-proxy",)) self.assertEqual(RUNNER.component_builds("platform", entries), ()) self.assertEqual(result["services"], ["reverse-proxy"]) def test_runner_creates_only_file_backed_runtime_secrets(self): with mock.patch.object(RUNNER, "ensure_platform_runtime_secret") as ensure: RUNNER.prepare_component_runtime( "platform", RUNNER.PLATFORM_DEVICE_CORE_HUB_TRUST_ENTRIES, ) self.assertEqual( [call.args[0] for call in ensure.call_args_list], [RUNNER.PLATFORM_DEVICE_CORE_INTERNAL_TOKEN_FILE], ) with ( mock.patch.object( RUNNER, "ensure_platform_runtime_secret", ) as ensure, mock.patch.object( RUNNER, "ensure_device_edge_channel_core_identity", ) as ensure_edge_identity, ): RUNNER.prepare_component_runtime( "device-plane", RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V1_SUCCESSOR_ENTRIES, ) self.assertEqual( [call.args[0] for call in ensure.call_args_list], [ RUNNER.DEVICE_PLANE_POSTGRES_PASSWORD_FILE, RUNNER.DEVICE_PLANE_GATEWAY_CORE_TOKEN_FILE, RUNNER.DEVICE_PLANE_IDENTIFIER_PEPPER_FILE, RUNNER.DEVICE_PLANE_MANAGEMENT_CORE_TOKEN_FILE, RUNNER.PLATFORM_DEVICE_CORE_INTERNAL_TOKEN_FILE, ], ) ensure_edge_identity.assert_not_called() with ( mock.patch.object( RUNNER, "ensure_platform_runtime_secret", ) as ensure, mock.patch.object( RUNNER, "ensure_device_edge_channel_core_identity", ) as ensure_edge_identity, ): RUNNER.prepare_component_runtime( "device-plane", RUNNER.DEVICE_PLANE_EDGE_CORE_CHANNEL_BOOTSTRAP_ENTRIES, ) self.assertEqual( [call.args[0] for call in ensure.call_args_list], [ RUNNER.DEVICE_PLANE_POSTGRES_PASSWORD_FILE, RUNNER.DEVICE_PLANE_GATEWAY_CORE_TOKEN_FILE, RUNNER.DEVICE_PLANE_IDENTIFIER_PEPPER_FILE, RUNNER.DEVICE_PLANE_MANAGEMENT_CORE_TOKEN_FILE, ], ) ensure_edge_identity.assert_called_once_with( allow_invalid_unexported_recovery=True ) def test_apply_gate_checks_exact_services_core_contract_and_runtime_boundary(self): entries = RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V1_SUCCESSOR_ENTRIES services = ("device-control-core", "device-manager") with ( mock.patch.object( RUNNER, "healthcheck_compose_service_with_grace", ) as service_health, mock.patch.object(RUNNER, "healthcheck_url") as url_health, mock.patch.object( RUNNER, "validate_device_manager_control_plane_runtime", ) as runtime_acceptance, ): RUNNER.run_healthchecks("device-plane", entries, services) self.assertEqual( [call.args for call in service_health.call_args_list], [ ("device-plane", "device-control-core"), ("device-plane", "device-manager"), ], ) url_health.assert_called_once_with( RUNNER.component_healthchecks( "device-plane", entries, services, )[0] ) runtime_acceptance.assert_called_once_with() def test_activation_resolves_predecessor_from_descriptor_and_journal(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-manager-release-predecessor-", ) as directory: root = Path(directory) applied = root / "applied" backups = root / "backups" temporary = root / "tmp" device_plane = root / "device-plane" for path in (applied, backups, temporary, device_plane / "deployment"): path.mkdir(parents=True) predecessor_patch = "device-manager-reconciliation-unit-004" artifact_name = f"nodedc-device-plane-{predecessor_patch}.tgz" artifact = applied / artifact_name artifact_bytes = b"reviewed-reconciliation-artifact" artifact.write_bytes(artifact_bytes) artifact_sha = hashlib.sha256(artifact_bytes).hexdigest() descriptor = device_manager_release_descriptor( predecessor_patch=predecessor_patch, predecessor_sha=artifact_sha, ) state_file = root / "applied.jsonl" state_file.write_text( json.dumps({ "id": predecessor_patch, "artifact": artifact_name, "backup_id": "predecessor-backup", "component": "device-plane", "sha256": artifact_sha, "status": "ok", }) + "\n", encoding="utf-8", ) (backups / "predecessor-backup").mkdir() baseline_backup = root / "failed-backup" predecessor_descriptor = {"marker": "exact"} marker = ( device_plane / RUNNER.DEVICE_PLANE_MANAGER_V2_RECONCILIATION_REL ) marker.write_text( json.dumps(predecessor_descriptor), encoding="utf-8", ) with ( mock.patch.object(RUNNER, "APPLIED_DIR", applied), mock.patch.object(RUNNER, "BACKUPS_DIR", backups), mock.patch.object(RUNNER, "TMP_DIR", temporary), mock.patch.object(RUNNER, "STATE_FILE", state_file), mock.patch.object( RUNNER, "component_root", return_value=device_plane, ), mock.patch.object( RUNNER, "validate_device_plane_manager_release_payload", return_value=descriptor, ) as payload, mock.patch.object( RUNNER, "load_artifact", return_value=( { "id": predecessor_patch, "component": "device-plane", "type": "app-overlay", }, RUNNER.DEVICE_PLANE_MANAGER_V2_RECONCILIATION_ENTRIES, root / "predecessor-payload", ), ), mock.patch.object( RUNNER, "validate_device_plane_manager_v2_reconciliation_payload", return_value=predecessor_descriptor, ), mock.patch.object( RUNNER, "validate_device_plane_manager_v2_reconciliation_backup", return_value=baseline_backup, ), mock.patch.object( RUNNER, "validate_device_plane_manager_v2_reconciled_baseline", return_value={"accepted": True}, ) as baseline, ): result = ( RUNNER.validate_device_plane_manager_activation_predecessor( root / "payload" ) ) self.assertEqual( result["mode"], "reconciled-manager-forward-activation", ) payload.assert_called_once_with(root / "payload") baseline.assert_called_once_with( baseline_backup, marker_installed=True, ) def test_failed_patch_id_and_sha_are_globally_non_replayable(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-manager-failed-replay-", ) as directory: failed_state = Path(directory) / "failed.jsonl" failed_state.write_text( json.dumps({ "id": "device-manager-release-failed-001", "sha256": "b" * 64, "status": "failed", }) + "\n", encoding="utf-8", ) with mock.patch.object( RUNNER, "FAILED_STATE_FILE", failed_state, ): with self.assertRaisesRegex( RUNNER.DeployError, "SHA is terminal failed", ): RUNNER.reject_failed_artifact_replay( {"id": "different"}, "b" * 64, ) with self.assertRaisesRegex( RUNNER.DeployError, "patch id is terminal failed", ): RUNNER.reject_failed_artifact_replay( {"id": "device-manager-release-failed-001"}, "c" * 64, ) def test_upgrade_accepts_any_current_release_without_runner_patch_ids(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-manager-release-upgrade-", ) as directory: root = Path(directory) applied = root / "applied" backups = root / "backups" temporary = root / "tmp" device_plane = root / "device-plane" for path in (applied, backups, temporary, device_plane / "deployment"): path.mkdir(parents=True) predecessor_patch = "device-manager-release-arbitrary-041" artifact_name = f"nodedc-device-plane-{predecessor_patch}.tgz" artifact = applied / artifact_name artifact_bytes = b"arbitrary-reviewed-device-manager-release" artifact.write_bytes(artifact_bytes) artifact_sha = hashlib.sha256(artifact_bytes).hexdigest() predecessor_descriptor = device_manager_release_v1_descriptor( release_id=predecessor_patch, ) candidate_descriptor = device_manager_release_descriptor( release_id="device-manager-release-arbitrary-042", action="upgrade", predecessor_kind="release", predecessor_patch=predecessor_patch, predecessor_sha=artifact_sha, ) installed = ( device_plane / RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_REL ) installed.write_text( json.dumps(predecessor_descriptor), encoding="utf-8", ) state_file = root / "applied.jsonl" state_file.write_text( json.dumps({ "id": predecessor_patch, "artifact": artifact_name, "backup_id": "arbitrary-release-backup", "component": "device-plane", "sha256": artifact_sha, "status": "ok", }) + "\n", encoding="utf-8", ) (backups / "arbitrary-release-backup").mkdir() with ( mock.patch.object(RUNNER, "APPLIED_DIR", applied), mock.patch.object(RUNNER, "BACKUPS_DIR", backups), mock.patch.object(RUNNER, "TMP_DIR", temporary), mock.patch.object(RUNNER, "STATE_FILE", state_file), mock.patch.object( RUNNER, "component_root", return_value=device_plane, ), mock.patch.object( RUNNER, "validate_device_plane_manager_release_payload", side_effect=( candidate_descriptor, predecessor_descriptor, ), ), mock.patch.object( RUNNER, "load_artifact", return_value=( { "id": predecessor_patch, "component": "device-plane", "type": "app-overlay", }, RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES, root / "predecessor-payload", ), ), mock.patch.object( RUNNER, "healthcheck_compose_service", ) as health, ): result = ( RUNNER.validate_device_plane_manager_activation_predecessor( root / "candidate-payload" ) ) self.assertEqual(result["mode"], "active-manager-forward-upgrade") self.assertEqual( [call.args for call in health.call_args_list], [ ("device-plane", "device-control-core"), ("device-plane", "device-manager"), ("device-plane", "device-postgres"), ], ) def test_initial_install_rollback_removes_manager_and_restores_core_only(self): entries = RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES missing = { RUNNER.DEVICE_PLANE_MANAGER_COMPOSE_REL, "services/device-manager", RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_REL, } existing = [entry for entry in entries if entry not in missing] with tempfile.TemporaryDirectory( prefix="nodedc-device-manager-rollback-", ) as directory: backup = Path(directory) / "backup" backup.mkdir() (backup / "existing-files.txt").write_text( "\n".join(existing) + "\n", encoding="utf-8", ) (backup / "missing-files.txt").write_text( "\n".join(entry for entry in entries if entry in missing) + "\n", encoding="utf-8", ) (backup / "runtime-before.json").write_text( json.dumps(healthy_device_plane_inventory()), encoding="utf-8", ) with ( mock.patch.object( RUNNER, "stop_and_remove_compose_services", ) as stop, mock.patch.object( RUNNER, "restore_platform_overlay", return_value=len(entries), ) as restore, mock.patch.object( RUNNER, "run_component_runtime", ) as restore_runtime, mock.patch.object( RUNNER, "healthcheck_compose_service_with_grace", ) as restore_health, mock.patch.object( RUNNER, "component_healthchecks", return_value=("core-health",), ), mock.patch.object( RUNNER, "healthcheck_url", ) as restore_url_health, mock.patch.object( RUNNER, "run_healthchecks", ) as generic_health, ): result = RUNNER.rollback_device_plane_apply( Path(directory) / "live", backup, entries, "test-stamp", True, ("device-control-core", "device-manager"), ) stop.assert_called_once_with("device-plane", ("device-manager",)) restore.assert_called_once() restore_runtime.assert_called_once_with( "device-plane", existing, ("device-control-core",), ) restore_health.assert_called_once_with( "device-plane", "device-control-core", ) restore_url_health.assert_called_once_with("core-health") generic_health.assert_not_called() self.assertEqual( result, f"source+runtime-restored:{len(entries)}", ) def test_upgrade_rollback_restores_previous_core_and_manager_release(self): entries = RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES with tempfile.TemporaryDirectory( prefix="nodedc-device-manager-upgrade-rollback-", ) as directory: backup = Path(directory) / "backup" backup.mkdir() (backup / "existing-files.txt").write_text( "\n".join(entries) + "\n", encoding="utf-8", ) (backup / "missing-files.txt").write_text("", encoding="utf-8") (backup / "runtime-before.json").write_text( json.dumps(healthy_device_plane_inventory(include_manager=True)), encoding="utf-8", ) with ( mock.patch.object( RUNNER, "stop_and_remove_compose_services", ) as stop, mock.patch.object( RUNNER, "restore_platform_overlay", return_value=len(entries), ), mock.patch.object( RUNNER, "run_component_runtime", ) as restore_runtime, mock.patch.object( RUNNER, "healthcheck_compose_service_with_grace", ) as restore_health, mock.patch.object( RUNNER, "component_healthchecks", return_value=(), ), ): result = RUNNER.rollback_device_plane_apply( Path(directory) / "live", backup, entries, "test-stamp", True, ("device-control-core", "device-manager"), ) stop.assert_not_called() restore_runtime.assert_called_once_with( "device-plane", list(entries), ("device-control-core", "device-manager"), ) self.assertEqual( [call.args for call in restore_health.call_args_list], [ ("device-plane", "device-control-core"), ("device-plane", "device-manager"), ], ) self.assertEqual( result, f"source+runtime-restored:{len(entries)}", ) if __name__ == "__main__": unittest.main(verbosity=2)