import hashlib import importlib.machinery import importlib.util import json import os from pathlib import Path import subprocess import tarfile import tempfile import unittest from unittest import mock SCRIPT_DIR = Path(__file__).resolve().parent RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy" BUILDER_PATH = SCRIPT_DIR / "build-engine-provider-rotating-slot-artifact.mjs" ENGINE_ROOT = SCRIPT_DIR.parent.parent.parent / "NODEDC_ENGINE_INFRA" PATCH_ID = "engine-provider-rotating-slot-20991231-999" AUTHORITY_SUCCESSOR_SHA256 = "1a14299167ebe17efd677fa3c59b84c80e12d6846bac6f40f9bcae0729ab22c6" def load_runner(): loader = importlib.machinery.SourceFileLoader( "nodedc_engine_provider_rotating_slot", 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 EngineProviderRotatingSlotTest(unittest.TestCase): def source_has_advanced(self): current_sha256 = { relative_path: hashlib.sha256((ENGINE_ROOT / relative_path).read_bytes()).hexdigest() for relative_path in RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_ARTIFACT_ENTRIES } return current_sha256 != RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_TARGET_SHA256 def build(self, artifact_dir): environment = os.environ.copy() environment["NODEDC_ENGINE_SOURCE_ROOT"] = str(ENGINE_ROOT) environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir) completed = subprocess.run( ["node", str(BUILDER_PATH), PATCH_ID], check=True, capture_output=True, text=True, env=environment, ) return json.loads(completed.stdout) def test_builder_emits_exact_deterministic_two_file_slice(self): if self.source_has_advanced(): self.skipTest("historical rotating-slot builder source has advanced to its exact successor") with tempfile.TemporaryDirectory(prefix="nodedc-provider-rotating-slot-") as directory: root = Path(directory) first = self.build(root / "first") second = self.build(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(first["services"], ["nodedc-backend"]) self.assertEqual(first["credentialSlot"], "ndcProviderRotatingAccessApi") self.assertEqual( tuple(first["entries"]), RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_ARTIFACT_ENTRIES, ) extract = root / "extract" with tarfile.open(first_artifact, "r:gz") as archive: archive.extractall(extract, filter="data") entries = tuple( (extract / "files.txt").read_text(encoding="utf-8").splitlines() ) payload = extract / "payload" self.assertTrue(RUNNER.is_engine_provider_rotating_slot_slice("engine", entries)) self.assertEqual(RUNNER.component_services("engine", entries), ("nodedc-backend",)) self.assertEqual( RUNNER.component_healthchecks("engine", entries, ("nodedc-backend",)), ("http://127.0.0.1:3001/health",), ) RUNNER.validate_engine_provider_rotating_slot_slice(payload, entries) def test_preflight_is_exact_and_rejects_any_drift(self): with tempfile.TemporaryDirectory(prefix="nodedc-provider-rotating-preflight-") as directory: root = Path(directory) for relative_path in RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_ARTIFACT_ENTRIES: path = root / relative_path path.parent.mkdir(parents=True, exist_ok=True) path.write_text("predecessor\n", encoding="utf-8") def exact_sha(path): relative_path = Path(path).relative_to(root).as_posix() return RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_PREDECESSOR_SHA256[relative_path] with ( mock.patch.object(RUNNER, "component_root", return_value=root), mock.patch.object(RUNNER, "sha256_file", side_effect=exact_sha), mock.patch.object( RUNNER, "preflight_engine_credential_backend_runtime", return_value={"mode": "verified-derived-retry"}, ), ): result = RUNNER.preflight_engine_provider_rotating_slot_predecessor() self.assertEqual(result["mode"], "exact-gelios-v4-credential-slot-alignment") self.assertEqual( result["predecessor_sha256"], RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_PREDECESSOR_SHA256, ) drift_path = root / RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_ARTIFACT_ENTRIES[0] def drift_sha(path): if Path(path) == drift_path: return "0" * 64 return exact_sha(path) with ( mock.patch.object(RUNNER, "component_root", return_value=root), mock.patch.object(RUNNER, "sha256_file", side_effect=drift_sha), ): with self.assertRaises(RUNNER.DeployError): RUNNER.preflight_engine_provider_rotating_slot_predecessor() def test_healthchecks_dispatch_exact_rotating_slot_acceptance(self): entries = RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_ARTIFACT_ENTRIES root = Path("/engine") def installed_sha(path): relative_path = Path(path).relative_to(root).as_posix() return RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_TARGET_SHA256[relative_path] with ( mock.patch.object(RUNNER, "component_root", return_value=root), mock.patch.object(RUNNER, "sha256_file", side_effect=installed_sha), mock.patch.object(RUNNER, "healthcheck_compose_service"), mock.patch.object(RUNNER, "component_healthchecks", return_value=()), mock.patch.object( RUNNER, "preflight_engine_credential_backend_runtime", return_value={"mode": "verified-derived-retry"}, ), mock.patch.object( RUNNER, "accept_engine_provider_rotating_slot_runtime", return_value={"live": "accepted"}, ) as rotating_acceptance, mock.patch.object(RUNNER, "accept_engine_composite_provider_runtime") as composite_acceptance, mock.patch.object(RUNNER, "healthcheck_container"), ): RUNNER.run_healthchecks("engine", entries, ("nodedc-backend",)) rotating_acceptance.assert_called_once_with() composite_acceptance.assert_not_called() def test_live_acceptance_uses_the_rotating_slot_target_contract(self): if self.source_has_advanced(): self.skipTest("historical rotating-slot live fixture source has advanced to its exact successor") expected_live = ( "engine-provider-rotating-slot:gelios.provider.v4:" "ndcProviderRotatingAccessApi:fleet.positions.current.v3:monitoring,units" ) with ( mock.patch.object(RUNNER, "component_root", return_value=ENGINE_ROOT), mock.patch.object(RUNNER, "engine_backend_container_id", return_value="backend"), mock.patch.object( RUNNER, "run_engine_backend_probe", return_value=expected_live, ) as backend_probe, ): result = RUNNER.accept_engine_provider_rotating_slot_runtime() self.assertEqual(result["live"], expected_live) self.assertEqual( result["target_sha256"], RUNNER.ENGINE_PROVIDER_ROTATING_SLOT_TARGET_SHA256, ) self.assertEqual( backend_probe.call_args.args[1], "Engine provider rotating slot authority", ) if __name__ == "__main__": unittest.main(verbosity=2)