#!/usr/bin/env python3 import importlib.machinery import importlib.util import inspect import json import stat import tempfile import unittest 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_device_plane_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() class DevicePlaneRegistryTest(unittest.TestCase): def test_manager_v4_pins_032_and_persistent_white_boundary(self): self.assertEqual( RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V4_PREDECESSOR_PATCH_ID, "device-manager-release-v3-20260822-032", ) self.assertEqual( RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V4_PREDECESSOR_ARTIFACT_SHA256, "6e0eb3a0a6f19ceab92d46832b93bffbcea21247dbdc2ea50625a51ff460e4ca", ) self.assertEqual( RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V4_COMPOSE_SHA256, "e7dff0f5873ad4586bd55946d3db2bb86092a5e149e886d120adc041e056c256", ) boundaries = RUNNER.expected_device_plane_manager_release_v4_boundaries() self.assertEqual(boundaries["defaultAccentHex"], "#f5f5f5") self.assertEqual( boundaries["presentationDataHostPath"], "/volume1/docker/nodedc-device-plane/data/device-manager", ) self.assertEqual( boundaries["presentationDataContainerPath"], "/var/lib/nodedc-device-manager", ) def test_manager_v3_targets_applied_control_core_recovery(self): self.assertEqual( RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V3_CONTROL_CORE_PREDECESSOR_PATCH_ID, "device-control-core-release-v2-20260821-030", ) self.assertEqual( RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V3_CONTROL_CORE_PREDECESSOR_ARTIFACT_SHA256, "8459521a662541a5a87cb0188991cdcfb51727427db8ec2a232ce4846bfc3454", ) self.assertEqual( RUNNER.expected_device_plane_manager_release_v3_boundaries()[ "controlCorePredecessor" ], { "patchId": ( "device-control-core-release-v2-20260821-030" ), "artifactSha256": ( "8459521a662541a5a87cb0188991cdcfb51727427db8ec2a232ce4846bfc3454" ), }, ) def test_control_core_predecessor_health_is_phase_scoped(self): with self.assertRaisesRegex( RUNNER.DeployError, "preflight phase is invalid", ): RUNNER.validate_device_plane_control_core_release_predecessor( Path("/not-used"), preflight_phase="unknown", ) plan_source = inspect.getsource(RUNNER.plan_artifact) self.assertIn( 'validate_device_plane_control_core_release_predecessor(\n' ' payload_dir,\n' ' preflight_phase="plan",', plan_source, ) apply_source = inspect.getsource(RUNNER.apply_artifact) self.assertIn( 'validate_device_plane_control_core_release_predecessor(\n' ' payload_dir,\n' ' preflight_phase="apply",', apply_source, ) def test_control_core_preflight_health_excludes_selected_target(self): descriptor = { "preservedServices": [ "device-manager", "device-gateway", "device-postgres", "device-backhaul-target", ], } with mock.patch.object( RUNNER, "healthcheck_compose_service", ) as healthcheck: services = ( RUNNER.validate_device_plane_control_core_preserved_runtime_health( descriptor ) ) self.assertEqual( services, ( "device-manager", "device-gateway", "device-postgres", "device-backhaul-target", ), ) self.assertEqual( healthcheck.call_args_list, [ mock.call("device-plane", "device-manager"), mock.call("device-plane", "device-gateway"), mock.call("device-plane", "device-postgres"), mock.call("device-plane", "device-backhaul-target"), ], ) self.assertNotIn( mock.call("device-plane", "device-control-core"), healthcheck.call_args_list, ) def test_control_core_selected_predecessor_may_be_unhealthy(self): inventory = { "schemaVersion": "nodedc.device-plane.runtime-inventory.v1", "composeProject": "nodedc-device-plane", "services": [{ "service": "device-control-core", "containerId": "a" * 64, "imageId": "sha256:" + "b" * 64, "status": "running", "running": True, "health": "unhealthy", "restartCount": 4, }], } with mock.patch.object( RUNNER, "device_plane_runtime_inventory", return_value=inventory, ): selected = ( RUNNER.validate_device_plane_control_core_selected_predecessor_runtime() ) self.assertEqual(selected["health"], "unhealthy") def test_control_core_selected_predecessor_may_be_restarting(self): inventory = { "schemaVersion": "nodedc.device-plane.runtime-inventory.v1", "composeProject": "nodedc-device-plane", "services": [{ "service": "device-control-core", "containerId": "a" * 64, "imageId": "sha256:" + "b" * 64, "status": "restarting", "running": True, "health": "starting", "restartCount": 5, }], } with mock.patch.object( RUNNER, "device_plane_runtime_inventory", return_value=inventory, ): selected = ( RUNNER.validate_device_plane_control_core_selected_predecessor_runtime() ) self.assertEqual(selected["status"], "restarting") self.assertEqual(selected["health"], "starting") def test_control_core_rollback_accepts_restored_unhealthy_boundary(self): service_names = ( "device-control-core", "device-manager", "device-gateway", "device-postgres", "device-backhaul-target", ) inventory = { "schemaVersion": "nodedc.device-plane.runtime-inventory.v1", "composeProject": "nodedc-device-plane", "services": [ { "service": service, "containerId": chr(97 + index) * 64, "imageId": "sha256:" + str(index + 1) * 64, "status": "running", "running": True, "health": "unhealthy" if index == 0 else "healthy", "restartCount": index, } for index, service in enumerate(service_names) ], } restored_id = "f" * 64 restored = { "Id": restored_id, "State": { "Status": "running", "Running": True, "Health": {"Status": "unhealthy"}, }, } with ( mock.patch.object( RUNNER, "healthcheck_compose_service_with_grace", ) as preserved_health, mock.patch.object( RUNNER, "compose_service_container_id", return_value=restored_id, ), mock.patch.object( RUNNER, "inspect_device_plane_container", return_value=restored, ), ): accepted = ( RUNNER.accept_device_plane_control_core_rollback_runtime( inventory ) ) self.assertEqual(accepted["health"], "unhealthy") self.assertEqual(accepted["status"], "running") self.assertEqual(accepted["predecessorHealth"], "unhealthy") self.assertEqual( preserved_health.call_args_list, [ mock.call("device-plane", "device-manager"), mock.call("device-plane", "device-gateway"), mock.call("device-plane", "device-postgres"), mock.call("device-plane", "device-backhaul-target"), ], ) def test_control_core_post_apply_health_includes_backhaul(self): with ( mock.patch.object( RUNNER, "healthcheck_compose_service_with_grace", ) as healthcheck, mock.patch.object( RUNNER, "component_healthchecks", return_value=(), ), mock.patch.object( RUNNER, "validate_device_manager_control_plane_runtime", ), ): RUNNER.run_healthchecks( "device-plane", RUNNER.DEVICE_PLANE_CONTROL_CORE_RELEASE_V2_ENTRIES, ("device-control-core",), ) self.assertEqual( healthcheck.call_args_list, [ mock.call("device-plane", "device-control-core"), mock.call("device-plane", "device-manager"), mock.call("device-plane", "device-gateway"), mock.call("device-plane", "device-postgres"), mock.call("device-plane", "device-backhaul-target"), ], ) def test_manager_v3_post_apply_uses_private_egress_core_boundary(self): with ( mock.patch.object( RUNNER, "healthcheck_compose_service_with_grace", ), mock.patch.object( RUNNER, "component_healthchecks", return_value=(), ), mock.patch.object( RUNNER, "validate_device_manager_control_plane_runtime", ) as runtime_acceptance, ): RUNNER.run_healthchecks( "device-plane", RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V3_ENTRIES, ("device-manager",), ) runtime_acceptance.assert_called_once_with( require_edge_channel=True, core_network_mode="private-egress", ) def test_manager_v4_post_apply_requires_persistent_data(self): with ( mock.patch.object( RUNNER, "healthcheck_compose_service_with_grace", ), mock.patch.object( RUNNER, "component_healthchecks", return_value=(), ), mock.patch.object( RUNNER, "validate_device_manager_control_plane_runtime", ) as runtime_acceptance, ): RUNNER.run_healthchecks( "device-plane", RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V4_ENTRIES, ("device-manager",), ) runtime_acceptance.assert_called_once_with( require_edge_channel=True, core_network_mode="private-egress", require_persistent_data=True, ) def test_manager_v4_prepare_owns_new_parent_and_managed_data_directory(self): with tempfile.TemporaryDirectory( prefix="nodedc-manager-v4-data-", ) as directory: data_dir = Path(directory) / "data" / "device-manager" with ( mock.patch.object( RUNNER, "DEVICE_PLANE_MANAGER_DATA_DIR", data_dir, ), mock.patch.object(RUNNER.os, "chown") as chown, mock.patch.object( RUNNER, "validate_device_plane_manager_persistent_data_metadata", return_value="uid-1000-gid-1000-mode-0750", ) as validate, ): result = RUNNER.ensure_device_plane_manager_persistent_data() self.assertEqual(result, "uid-1000-gid-1000-mode-0750") self.assertEqual( chown.call_args_list, [ mock.call(data_dir.parent, 0, 0), mock.call(data_dir, 1000, 1000), ], ) validate.assert_called_once_with() def test_manager_v4_prepare_preserves_existing_safe_parent_metadata(self): with tempfile.TemporaryDirectory( prefix="nodedc-manager-v4-parent-", ) as directory: data_parent = Path(directory) / "data" data_parent.mkdir(mode=0o700) data_dir = data_parent / "device-manager" with ( mock.patch.object( RUNNER, "DEVICE_PLANE_MANAGER_DATA_DIR", data_dir, ), mock.patch.object(RUNNER.os, "chown") as chown, mock.patch.object( RUNNER, "validate_device_plane_manager_persistent_data_metadata", return_value="uid-1000-gid-1000-mode-0750", ), ): RUNNER.ensure_device_plane_manager_persistent_data() self.assertEqual(stat.S_IMODE(data_parent.stat().st_mode), 0o700) chown.assert_called_once_with(data_dir, 1000, 1000) def test_manager_v4_prepare_rejects_symlink_parent_before_child_creation(self): with tempfile.TemporaryDirectory( prefix="nodedc-manager-v4-symlink-", ) as directory: root = Path(directory) outside = root / "outside" outside.mkdir() data_parent = root / "data" data_parent.symlink_to(outside, target_is_directory=True) with mock.patch.object( RUNNER, "DEVICE_PLANE_MANAGER_DATA_DIR", data_parent / "device-manager", ): with self.assertRaisesRegex( RUNNER.DeployError, "persistent data parent is unsafe", ): RUNNER.ensure_device_plane_manager_persistent_data() self.assertFalse((outside / "device-manager").exists()) def test_manager_predecessor_health_is_phase_scoped(self): with self.assertRaisesRegex( RUNNER.DeployError, "preflight phase is invalid", ): RUNNER.validate_device_plane_manager_activation_predecessor( Path("/not-used"), preflight_phase="unknown", ) plan_source = inspect.getsource(RUNNER.plan_artifact) self.assertIn('preflight_phase="plan"', plan_source) apply_source = inspect.getsource(RUNNER.apply_artifact) self.assertIn('preflight_phase="apply"', apply_source) def test_manager_v3_health_gate_checks_only_preserved_services(self): descriptor = { "schemaVersion": ( "nodedc.device-plane.device-manager-release.v3" ), } with mock.patch.object( RUNNER, "healthcheck_compose_service", ) as healthcheck: services = ( RUNNER.validate_device_plane_manager_preserved_runtime_health( descriptor ) ) self.assertEqual( services, ("device-control-core", "device-postgres"), ) self.assertEqual( healthcheck.call_args_list, [ mock.call("device-plane", "device-control-core"), mock.call("device-plane", "device-postgres"), ], ) self.assertNotIn( mock.call("device-plane", "device-manager"), healthcheck.call_args_list, ) def test_legacy_manager_health_gate_excludes_selected_services(self): descriptor = { "schemaVersion": ( "nodedc.device-plane.device-manager-release.v2" ), } with mock.patch.object( RUNNER, "healthcheck_compose_service", ) as healthcheck: services = ( RUNNER.validate_device_plane_manager_preserved_runtime_health( descriptor ) ) self.assertEqual(services, ("device-postgres",)) healthcheck.assert_called_once_with( "device-plane", "device-postgres", ) def test_registry_has_exact_roots_project_and_stateless_services(self): component = RUNNER.COMPONENTS["device-plane"] root = Path("/volume1/docker/nodedc-device-plane") self.assertEqual(component["payload_root"], root) self.assertEqual(component["compose_root"], root) self.assertEqual(component["compose_project"], "nodedc-device-plane") self.assertEqual( component["compose_files"], (root / "docker-compose.device-plane.yml",), ) self.assertTrue(component["bootstrap_root"]) self.assertTrue(component["compose_no_deps"]) self.assertEqual( component["services"], ("device-control-core", "device-gateway"), ) self.assertNotIn("device-postgres", component["services"]) self.assertIsNone(component.get("compose_env_file")) def test_files_select_only_the_affected_stateless_services(self): self.assertEqual( RUNNER.component_services( "device-plane", ("services/device-control-core/src/server.mjs",), ), ("device-control-core",), ) self.assertEqual( RUNNER.component_services( "device-plane", ("services/device-gateway/src/server.mjs",), ), ("device-gateway",), ) self.assertEqual( RUNNER.component_services( "device-plane", ("packages/arusnavi-b2-adapter",), ), ("device-control-core", "device-gateway"), ) services = RUNNER.component_services( "device-plane", ("docker-compose.device-plane.yml",), ) self.assertEqual( services, ("device-control-core", "device-gateway"), ) self.assertNotIn("device-postgres", services) def test_build_selection_is_exact_and_database_free(self): builds = RUNNER.component_builds( "device-plane", ("services/device-gateway",), ) self.assertEqual(len(builds), 1) self.assertEqual(builds[0][0], RUNNER.DEVICE_PLANE_ROOT) self.assertIn( "services/device-gateway/Dockerfile", builds[0][1], ) self.assertIn(RUNNER.DEVICE_PLANE_GATEWAY_IMAGE, builds[0][1]) all_builds = RUNNER.component_builds( "device-plane", ("package-lock.json",), ) self.assertEqual(len(all_builds), 2) self.assertNotIn("device-postgres", " ".join( argument for _root, arguments in all_builds for argument in arguments )) def test_payload_allowlist_rejects_runtime_secrets_and_broad_paths(self): for allowed in ( ".dockerignore", "docker-compose.device-plane.yml", "packages/device-protocol-contract/src/index.mjs", "packages/device-edge-channel-contract/src/index.mjs", "services/device-control-core/Dockerfile", "services/device-gateway/src/runtime.mjs", ): self.assertTrue( RUNNER.allowed_payload_path("device-plane", allowed), ) for rejected in ( ".env", "secrets/postgres-password", "runtime/postgres/data", "services/unknown/server.mjs", "packages/device-adapter-runtime/package.json", "packages/device-adapter-catalog/package.json", "services/device-gateway-core/src/runtime.mjs", "docker-compose.yml", "services/device-control-core/start.sh", "services/device-control-core/test/app.test.mjs", ): with self.assertRaises(RUNNER.DeployError): RUNNER.allowed_payload_path("device-plane", rejected) def test_healthchecks_are_service_scoped_and_fail_closed(self): core_checks = RUNNER.component_healthchecks( "device-plane", ("services/device-control-core",), ("device-control-core",), ) self.assertEqual(len(core_checks), 1) self.assertEqual( core_checks[0]["expected_json"]["commandTransport"], "disabled", ) self.assertEqual( core_checks[0]["expected_json"]["discoveryIngest"], "disabled", ) gateway_checks = RUNNER.component_healthchecks( "device-plane", ("services/device-gateway",), ("device-gateway",), ) self.assertEqual(len(gateway_checks), 1) self.assertEqual( gateway_checks[0]["expected_json"]["publicIngress"], "disabled", ) self.assertEqual( gateway_checks[0]["expected_json"]["tcpListener"], "disabled", ) def test_runtime_secrets_are_runner_owned_and_not_manifest_selected(self): with mock.patch.object( RUNNER, "ensure_platform_runtime_secret", ) as ensure: RUNNER.prepare_component_runtime( "device-plane", ("services/device-control-core",), ) 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, ], ) def test_compose_runtime_uses_no_deps_and_never_selects_database(self): with mock.patch.object(RUNNER.subprocess, "run") as run: RUNNER.run_compose( "device-plane", ("device-control-core", "device-gateway"), ("docker-compose.device-plane.yml",), ) command = run.call_args_list[0].args[0] self.assertIn("--no-deps", command) self.assertNotIn("device-postgres", command) self.assertEqual( command[-2:], ["device-control-core", "device-gateway"], ) def test_initial_candidate_rollback_removes_only_stateless_services(self): entries = ( "docker-compose.device-plane.yml", "services/device-control-core", "services/device-gateway", ) with tempfile.TemporaryDirectory( prefix="nodedc-device-plane-rollback-", ) as directory: root = Path(directory) / "live" backup = Path(directory) / "backup" root.mkdir() backup.mkdir() (backup / "existing-files.txt").write_text("", encoding="utf-8") (backup / "missing-files.txt").write_text( "\n".join(entries) + "\n", encoding="utf-8", ) (backup / "runtime-before.json").write_text( json.dumps({ "schemaVersion": "nodedc.device-plane.runtime-inventory.v1", "composeProject": "nodedc-device-plane", "services": [], }), encoding="utf-8", ) with ( mock.patch.object( RUNNER, "stop_and_remove_compose_services", ) as stop, mock.patch.object( RUNNER, "restore_platform_overlay", return_value=3, ) as restore, ): result = RUNNER.rollback_device_plane_apply( root, backup, entries, "test-stamp", True, ("device-control-core", "device-gateway"), ) stop.assert_called_once_with( "device-plane", ("device-control-core", "device-gateway"), ) restore.assert_called_once() self.assertEqual(result, "source-restored-runtime-unchanged:3") def test_db_bootstrap_compose_does_not_invent_stateless_baseline(self): entries = ( "docker-compose.device-plane.yml", "services/device-control-core", "services/device-gateway", ) with tempfile.TemporaryDirectory( prefix="nodedc-device-plane-compose-predecessor-", ) as directory: root = Path(directory) / "live" backup = Path(directory) / "backup" root.mkdir() backup.mkdir() (backup / "existing-files.txt").write_text( "docker-compose.device-plane.yml\n", encoding="utf-8", ) (backup / "missing-files.txt").write_text( "services/device-control-core\n" "services/device-gateway\n", encoding="utf-8", ) (backup / "runtime-before.json").write_text( json.dumps({ "schemaVersion": "nodedc.device-plane.runtime-inventory.v1", "composeProject": "nodedc-device-plane", "services": [{ "service": "device-postgres", "containerId": "a" * 64, "imageId": "sha256:" + "b" * 64, "status": "running", "running": True, "health": "healthy", "restartCount": 0, }], }), encoding="utf-8", ) with ( mock.patch.object( RUNNER, "stop_and_remove_compose_services", ) as stop, mock.patch.object( RUNNER, "restore_platform_overlay", return_value=3, ), mock.patch.object( RUNNER, "run_component_runtime", ) as run_runtime, ): result = RUNNER.rollback_device_plane_apply( root, backup, entries, "test-stamp", True, ("device-control-core", "device-gateway"), ) stop.assert_called_once_with( "device-plane", ("device-control-core", "device-gateway"), ) run_runtime.assert_not_called() self.assertEqual(result, "source-restored-runtime-unchanged:3") def test_installed_stateless_baseline_is_restored_from_runtime_inventory( self, ): entries = ( "docker-compose.device-plane.yml", "services/device-control-core", "services/device-gateway", ) with tempfile.TemporaryDirectory( prefix="nodedc-device-plane-installed-predecessor-", ) as directory: root = Path(directory) / "live" backup = Path(directory) / "backup" root.mkdir() 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({ "schemaVersion": "nodedc.device-plane.runtime-inventory.v1", "composeProject": "nodedc-device-plane", "services": [ { "service": "device-control-core", "containerId": "a" * 64, "imageId": "sha256:" + "b" * 64, "status": "running", "running": True, "health": "healthy", "restartCount": 0, }, { "service": "device-gateway", "containerId": "c" * 64, "imageId": "sha256:" + "d" * 64, "status": "running", "running": True, "health": "healthy", "restartCount": 0, }, ], }), encoding="utf-8", ) with ( mock.patch.object( RUNNER, "stop_and_remove_compose_services", ) as stop, mock.patch.object( RUNNER, "restore_platform_overlay", return_value=3, ), mock.patch.object( RUNNER, "run_component_runtime", ) as run_runtime, mock.patch.object(RUNNER, "run_healthchecks") as healthchecks, ): result = RUNNER.rollback_device_plane_apply( root, backup, entries, "test-stamp", True, ("device-control-core", "device-gateway"), ) stop.assert_not_called() run_runtime.assert_called_once_with( "device-plane", list(entries), ("device-control-core", "device-gateway"), ) healthchecks.assert_called_once_with( "device-plane", list(entries), ("device-control-core", "device-gateway"), ) self.assertEqual(result, "source+runtime-restored:3") if __name__ == "__main__": unittest.main(verbosity=2)