#!/usr/bin/env python3 import importlib.machinery import importlib.util import json 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_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", "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", "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)