#!/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(): 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 ( ("device-control-core", "a"), ("device-gateway", "b"), ("device-postgres", "c"), ) ], } 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_CONTROL_PLANE_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.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) 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_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: RUNNER.prepare_component_runtime( "device-plane", RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_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, ], ) def test_apply_gate_checks_exact_services_core_contract_and_runtime_boundary(self): entries = RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES services = ("device-control-core", "device-manager") with ( mock.patch.object( RUNNER, "healthcheck_compose_service", ) 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_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, "run_healthchecks", ) as restore_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", existing, ("device-control-core",), ) self.assertEqual( result, f"source+runtime-restored:{len(entries)}", ) if __name__ == "__main__": unittest.main(verbosity=2)