#!/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.mock import patch SCRIPT_DIR = Path(__file__).resolve().parent BUILDER = SCRIPT_DIR / "build-device-edge-ingress-artifact.mjs" RUNNER_PATH = SCRIPT_DIR / "nodedc-edge-deploy" def load_runner(): loader = importlib.machinery.SourceFileLoader( "nodedc_edge_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 DeviceEdgeIngressArtifactTest(unittest.TestCase): def build(self, artifact_dir, patch_id): environment = os.environ.copy() environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir) return subprocess.run( ["node", str(BUILDER), patch_id], check=False, capture_output=True, text=True, env=environment, ) def test_builder_is_deterministic_narrow_and_secret_free(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-edge-artifact-", ) as directory: artifact_dir = Path(directory) patch_id = "device-edge-ingress-ipvlan-unit-001" first_result = self.build(artifact_dir, patch_id) self.assertEqual(first_result.returncode, 0, first_result.stderr) first = json.loads(first_result.stdout) first_bytes = Path(first["artifact"]).read_bytes() second_result = self.build(artifact_dir, patch_id) self.assertEqual(second_result.returncode, 0, second_result.stderr) second = json.loads(second_result.stdout) second_bytes = Path(second["artifact"]).read_bytes() self.assertEqual(first_bytes, second_bytes) self.assertEqual(first["sha256"], second["sha256"]) self.assertEqual( first["sha256"], hashlib.sha256(first_bytes).hexdigest(), ) self.assertEqual(first["component"], "device-edge") self.assertEqual(first["entries"], list(RUNNER.ENTRIES)) self.assertEqual(first["services"], ["device-edge-relay"]) self.assertEqual( first["ingress"]["ipv4Approval"], "approved-outside-dhcp-pool", ) with tarfile.open(first["artifact"], "r:gz") as archive: members = archive.getmembers() names = {member.name for member in members} manifest = archive.extractfile("manifest.env").read().decode() files = archive.extractfile("files.txt").read().decode().splitlines() payload_bytes = b"\n".join( archive.extractfile(member).read() for member in members if member.isfile() ) self.assertEqual( manifest, f"id={patch_id}\ncomponent=device-edge\ntype=app-overlay\n", ) self.assertEqual(files, list(RUNNER.ENTRIES)) self.assertIn( "payload/docker-compose.device-edge.ingress.yml", names, ) self.assertNotIn(b"PRIVATE KEY", payload_bytes) self.assertFalse(any( "/test/" in name or "/secrets/" in name or "/keys/" in name or "/trust/" in name or "/node_modules/" in name or Path(name).name.startswith(".env") for name in names )) def test_production_builder_accepts_the_explicitly_approved_address(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-edge-address-gate-", ) as directory: result = self.build( Path(directory), "device-edge-admission-gate-20260804-002", ) self.assertEqual(result.returncode, 0, result.stderr) built = json.loads(result.stdout) self.assertEqual( built["ingress"]["ipv4Approval"], "approved-outside-dhcp-pool", ) self.assertTrue(Path(built["artifact"]).is_file()) def test_runner_loads_exact_artifact_and_enters_runtime_preflight(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-edge-runner-load-", ) as directory: workspace = Path(directory) inbox = workspace / "inbox" inbox.mkdir() result = self.build( inbox, "device-edge-admission-gate-20260804-003", ) self.assertEqual(result.returncode, 0, result.stderr) artifact = Path(json.loads(result.stdout)["artifact"]) extracted = workspace / "extracted" extracted.mkdir() old_inbox = RUNNER.INBOX_ROOT RUNNER.INBOX_ROOT = inbox try: manifest, entries, payload, digest, resolved = ( RUNNER.load_artifact(artifact, extracted) ) finally: RUNNER.INBOX_ROOT = old_inbox self.assertEqual(manifest["component"], "device-edge") self.assertEqual(entries, RUNNER.ENTRIES) self.assertEqual(resolved, artifact.resolve()) self.assertEqual(digest, hashlib.sha256(artifact.read_bytes()).hexdigest()) self.assertEqual( json.loads( (payload / "deployment/device-edge-admission-gate-v1.json") .read_text(encoding="utf-8") ), RUNNER.expected_descriptor(), ) preserved = { RUNNER.BACKHAUL_CONTAINER: {"Id": "backhaul"}, RUNNER.TAILNET_CONTAINER: {"Id": "tailnet"}, } with patch.object(RUNNER, "assert_new_identity"), patch.object( RUNNER, "current_source_state", ), patch.object(RUNNER, "validate_predecessor_runtime"), patch.object( RUNNER, "preserved_runtime_snapshot", return_value=preserved, ), patch.object(RUNNER, "validate_host_network_boundary"), patch.object( RUNNER, "arp_duplicate_detected", return_value=False, ), patch.object( RUNNER, "run", return_value=subprocess.CompletedProcess([], 1, "", ""), ): self.assertEqual(RUNNER.preflight(manifest, digest), preserved) def test_backup_restore_preserves_exact_predecessor_partition(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-edge-backup-", ) as directory: workspace = Path(directory) live = workspace / "live" backups = workspace / "backups" live.mkdir() backups.mkdir() for relative in RUNNER.ENTRIES: if relative in RUNNER.PREDECESSOR_ABSENT: continue target = live / relative if relative.endswith("/src"): target.mkdir(parents=True) (target / "server.mjs").write_text("old\n", encoding="utf-8") else: target.parent.mkdir(parents=True, exist_ok=True) target.write_text(f"old:{relative}\n", encoding="utf-8") old_live = RUNNER.LIVE_ROOT old_backups = RUNNER.BACKUP_ROOT RUNNER.LIVE_ROOT = live RUNNER.BACKUP_ROOT = backups try: _backup_id, backup = RUNNER.create_backup("unit-backup") for relative in RUNNER.ENTRIES: target = live / relative if target.exists(): if target.is_dir(): import shutil shutil.rmtree(target) else: target.unlink() target.parent.mkdir(parents=True, exist_ok=True) target.write_text("candidate\n", encoding="utf-8") RUNNER.restore_backup(backup) finally: RUNNER.LIVE_ROOT = old_live RUNNER.BACKUP_ROOT = old_backups for relative in RUNNER.PREDECESSOR_ABSENT: self.assertFalse((live / relative).exists()) self.assertEqual( (live / "docker-compose.device-edge.yml").read_text(), "old:docker-compose.device-edge.yml\n", ) self.assertEqual( (live / "services/device-edge-relay/src/server.mjs").read_text(), "old\n", ) def test_runner_selection_and_compose_commands_are_exact(self): self.assertTrue(RUNNER.INGRESS_IPV4_APPROVED) self.assertEqual( RUNNER.INGRESS_IPV4_APPROVAL, "approved-outside-dhcp-pool", ) self.assertEqual(RUNNER.RELAY_SERVICE, "device-edge-relay") self.assertEqual( RUNNER.expected_descriptor()["preservedServices"], ["device-edge-backhaul", "tailnet"], ) self.assertEqual( RUNNER.compose_command( "up", "--detach", "--no-deps", "--force-recreate", "--pull", "never", RUNNER.RELAY_SERVICE, ), [ RUNNER.DOCKER, "compose", "--project-name", RUNNER.COMPOSE_PROJECT, "--file", str(RUNNER.BASE_COMPOSE), "--file", str(RUNNER.INGRESS_COMPOSE), "up", "--detach", "--no-deps", "--force-recreate", "--pull", "never", RUNNER.RELAY_SERVICE, ], ) self.assertNotIn("down", RUNNER_PATH.read_text(encoding="utf-8")) def test_preserved_runtime_is_compared_from_one_atomic_snapshot(self): expected = { RUNNER.BACKHAUL_CONTAINER: {"Id": "backhaul"}, RUNNER.TAILNET_CONTAINER: {"Id": "tailnet"}, } with patch.object( RUNNER, "preserved_runtime_snapshot", return_value=expected, ) as snapshot: RUNNER.assert_preserved_runtime(expected) snapshot.assert_called_once_with() def test_network_acceptance_rejects_any_non_ipvlan_substitution(self): relay = { "NetworkSettings": { "Networks": { "nodedc-device-edge-private": {"IPAddress": "172.18.0.4"}, RUNNER.INGRESS_NETWORK: {"IPAddress": RUNNER.INGRESS_IPV4}, }, }, } accepted_network = [{ "Driver": "ipvlan", "Internal": False, "Options": { "parent": RUNNER.INGRESS_PARENT, "ipvlan_mode": "l2", }, "IPAM": { "Config": [{ "Subnet": RUNNER.INGRESS_SUBNET, "Gateway": RUNNER.INGRESS_GATEWAY, }], }, }] with patch.object(RUNNER, "docker_json", return_value=accepted_network): RUNNER.validate_network_runtime(relay) rejected_network = json.loads(json.dumps(accepted_network)) rejected_network[0]["Driver"] = "bridge" with patch.object(RUNNER, "docker_json", return_value=rejected_network): with self.assertRaisesRegex( RUNNER.DeployError, "ingress network driver mismatch", ): RUNNER.validate_network_runtime(relay) if __name__ == "__main__": unittest.main(verbosity=2)