#!/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 unittest import mock from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent PLATFORM_ROOT = SCRIPT_DIR.parent.parent RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy" BUILDER_PATH = SCRIPT_DIR / "build-engine-mcp-control-plane-artifact.mjs" TRANSITION_ID = "20260718-003" STAGE_ARTIFACT = ( PLATFORM_ROOT / "infra/deploy-artifacts" / "nodedc-engine-mcp-control-plane-20260718-003.tgz" ) STAGE_ARTIFACT_SHA256 = "249aef9527666c562e9648b15737e66cc5c1dc7c0788b58ea714da270b5eb4ba" def load_runner(): loader = importlib.machinery.SourceFileLoader("nodedc_engine_mcp_runner", 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 EngineMcpControlPlaneTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.temporary = tempfile.TemporaryDirectory(prefix="nodedc-engine-mcp-control-plane-") cls.root = Path(cls.temporary.name) stage_bytes = STAGE_ARTIFACT.read_bytes() if hashlib.sha256(stage_bytes).hexdigest() != STAGE_ARTIFACT_SHA256: raise RuntimeError("engine_mcp_control_plane_stage_artifact_sha256_mismatch") source_stage = cls.root / "immutable-source" source_stage.mkdir() RUNNER.safe_extract(STAGE_ARTIFACT, source_stage) cls.engine_source_root = source_stage / "payload" cls.results = [] for index in range(2): output = cls.root / f"build-{index}" output.mkdir() env = os.environ.copy() env["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(output) env["NODEDC_ENGINE_SOURCE_ROOT"] = str(cls.engine_source_root) result = subprocess.run( ["node", str(BUILDER_PATH), TRANSITION_ID], cwd=PLATFORM_ROOT, env=env, check=True, capture_output=True, text=True, ) cls.results.append(json.loads(result.stdout)) @classmethod def tearDownClass(cls): cls.temporary.cleanup() def artifact(self, index=0): return Path(self.results[index]["artifact"]) def extract(self, artifact, directory): RUNNER.safe_extract(artifact, directory) return directory / "payload", RUNNER.parse_files_list(directory / "files.txt") def test_artifact_is_byte_reproducible_and_canonical(self): first = self.artifact(0).read_bytes() second = self.artifact(1).read_bytes() self.assertEqual(first, second) self.assertEqual(first, STAGE_ARTIFACT.read_bytes()) self.assertEqual(first[4:8], b"\0\0\0\0") self.assertEqual(self.results[0]["artifactSha256"], hashlib.sha256(first).hexdigest()) with tarfile.open(self.artifact(0), "r:gz") as archive: for member in archive: self.assertTrue(member.isfile() or member.isdir()) self.assertFalse(Path(member.name).name.startswith("._")) def test_runner_selects_only_existing_backend(self): with tempfile.TemporaryDirectory() as directory: manifest, entries, payload = RUNNER.load_artifact( self.artifact(0), Path(directory), ) descriptor = RUNNER.validate_engine_mcp_control_plane_payload(payload, entries) override_text = ( payload / RUNNER.ENGINE_DATA_PRODUCT_READ_GRANT_OVERRIDE_REL ).read_text(encoding="utf-8") self.assertEqual(manifest["component"], "engine") self.assertEqual(tuple(entries), RUNNER.ENGINE_MCP_CONTROL_PLANE_ARTIFACT_ENTRIES) self.assertEqual(RUNNER.component_services("engine", entries), ("nodedc-backend",)) self.assertEqual(descriptor["action"], "activate") self.assertEqual( descriptor["source"]["gatewaySha256"], RUNNER.ENGINE_MCP_CONTROL_PLANE_TARGET_SHA256[ RUNNER.ENGINE_NODE_INTELLIGENCE_GATEWAY_REL ], ) self.assertEqual(override_text, RUNNER.expected_engine_data_product_read_grant_override()) def test_runtime_preparation_adds_only_private_reader_state(self): with ( mock.patch.object( RUNNER, "ensure_engine_edp_managed_provisioner_keypair", ) as keypair, mock.patch.object( RUNNER, "ensure_engine_data_product_grant_private_state", ) as private_state, ): RUNNER.prepare_component_runtime( "engine", RUNNER.ENGINE_MCP_CONTROL_PLANE_ARTIFACT_ENTRIES, ) keypair.assert_called_once_with() private_state.assert_called_once_with(include_reader=True) def test_candidate_preserves_node_intelligence_identity_except_exact_control_plane_digests(self): with tempfile.TemporaryDirectory() as directory: work = Path(directory) payload, entries = self.extract(self.artifact(0), work) candidate = RUNNER.validate_engine_mcp_control_plane_payload(payload, entries) self.assertEqual(candidate["releaseId"], RUNNER.ENGINE_NODE_INTELLIGENCE_RELEASE_ID) self.assertEqual(candidate["upstream"]["commit"], RUNNER.ENGINE_NODE_INTELLIGENCE_UPSTREAM_COMMIT) self.assertEqual(candidate["image"]["tag"], RUNNER.ENGINE_NODE_INTELLIGENCE_IMAGE) self.assertEqual(candidate["expectedCurrent"], "inactive") self.assertEqual( candidate["source"]["upstreamProjectionSha256"], RUNNER.ENGINE_MCP_CONTROL_PLANE_TARGET_SHA256[ "nodedc-source/server/nodeIntelligence/upstreamProjection.js" ], ) def test_descriptor_boundary_and_source_digest_are_enforced(self): with tempfile.TemporaryDirectory() as directory: work = Path(directory) payload, entries = self.extract(self.artifact(0), work) descriptor_path = payload / RUNNER.ENGINE_MCP_CONTROL_PLANE_DESCRIPTOR_REL descriptor = json.loads(descriptor_path.read_text(encoding="utf-8")) descriptor["source"]["catalogSha256"] = "0" * 64 descriptor_path.write_text(json.dumps(descriptor), encoding="utf-8") # Payload validation permits only a structurally valid descriptor; # predecessor equality is the state-aware barrier for untouched # node-intelligence source identities. with self.assertRaisesRegex( RUNNER.DeployError, "descriptor crosses node-intelligence source boundary", ): installed = json.loads(json.dumps(descriptor)) installed["source"]["catalogSha256"] = "1" * 64 installed["source"]["gatewaySha256"] = ( RUNNER.ENGINE_MCP_CONTROL_PLANE_PREDECESSOR_SHA256[ RUNNER.ENGINE_NODE_INTELLIGENCE_GATEWAY_REL ] ) from unittest import mock with ( mock.patch.object( RUNNER, "current_engine_node_intelligence_descriptor", return_value=installed, ), mock.patch.object(RUNNER, "validate_installed_engine_node_intelligence_source"), ): RUNNER.preflight_engine_mcp_control_plane_predecessor(payload) def test_existing_artifact_is_not_overwritten(self): env = os.environ.copy() env["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(self.artifact(0).parent) env["NODEDC_ENGINE_SOURCE_ROOT"] = str(self.engine_source_root) result = subprocess.run( ["node", str(BUILDER_PATH), TRANSITION_ID], cwd=PLATFORM_ROOT, env=env, check=False, capture_output=True, text=True, ) self.assertNotEqual(result.returncode, 0) self.assertIn("artifact_already_exists", result.stderr) if __name__ == "__main__": unittest.main(verbosity=2)