import copy import json import os from pathlib import Path import tempfile import subprocess import sys import unittest from serviceification_d3_engine import DeployError, WorkerOwnership from standalone_l0_workers import L0_WORKER_STEMS from standalone_l0_release import atomic_role, upgrade_worker_bindings OLD = 'sha256:' + '1' * 64 NEW = 'sha256:' + '2' * 64 SOURCE = 'b' * 40 class L0ReleaseTest(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.root = Path(self.directory.name) self.roles = self.root / 'config/workers' self.roles.mkdir(parents=True) self.db = {} self.calls = [] self.original = {} for stem in L0_WORKER_STEMS: family = 'runtime.' + stem + '.v1' role = {'work_family': family, 'deployment_id': stem, 'worker_release_digest': OLD, 'worker_source_revision': 'a' * 40, 'ownership_epoch': 7, 'catalog_snapshot_store_root': '/installed/catalog', 'app_installation_id': 'existing', 'engine_generation': 1} self.original[stem] = role atomic_role(self.roles / (stem + '.json'), role) self.db[family] = WorkerOwnership(family, stem, OLD, 7, True, 'install') def activate(self, stem, request): self.calls.append(stem) current = self.db[request['workFamily']] if current.ownership_epoch != int(request['expectedOwnershipEpoch']): raise DeployError('CAS conflict') self.db[current.work_family] = WorkerOwnership(current.work_family, current.deployment_id, NEW, current.ownership_epoch + 1, True, request['operationId']) return {'schema': 'nexuscrm.runtime-work-ownership-activation-result/v1', 'operationId': request['operationId'], 'workFamily': current.work_family, 'deploymentId': current.deployment_id, 'workerReleaseDigest': NEW, 'ownershipEpoch': str(current.ownership_epoch + 1)} def upgrade(self, **kwargs): return upgrade_worker_bindings(state_root=self.root, target_digest=NEW, target_source=kwargs.pop('target_source', SOURCE), envv={}, read_ownership=lambda _: copy.deepcopy(self.db), activate=kwargs.pop('activate', self.activate), **kwargs) def test_all_roles_preserve_installation_and_completed_retry_is_noop(self): self.assertEqual(self.upgrade(), len(L0_WORKER_STEMS)) for stem, previous in self.original.items(): actual = json.loads((self.roles / (stem + '.json')).read_text()) self.assertEqual(actual, {**previous, 'worker_release_digest': NEW, 'worker_source_revision': SOURCE, 'ownership_epoch': 8}) self.assertEqual(self.upgrade(), 0) self.assertEqual(len(self.calls), len(L0_WORKER_STEMS)) def test_remote_entrypoint_keeps_exact_source_policy_before_environment_access(self): result = subprocess.run([sys.executable, str(Path(__file__).with_name('standalone_l0_release.py')), '--state-root', str(self.root), '--worker-image', 'registry.invalid/worker@' + NEW, '--source', 'local', '--runtime-container', 'test-api', '--worker-container-prefix', 'test-worker-'], capture_output=True, text=True) self.assertNotEqual(result.returncode, 0) self.assertIn('remote L0 source must be an exact commit SHA', result.stderr) self.assertEqual(self.calls, []) def remote_cli(self, worker_image, catalog_image, *, inspector_source=SOURCE, inspector_available=True): (self.root / 'config/catalog-controlplane-image').write_text(catalog_image + '\n') binary = self.root / 'bin' binary.mkdir(exist_ok=True) trace = self.root / 'docker-calls' trace.unlink(missing_ok=True) docker = binary / 'docker' docker.write_text('#!' + sys.executable + '\n' + 'import os, pathlib, sys\n' 'args = sys.argv[1:]\n' 'with pathlib.Path(os.environ["DOCKER_TRACE"]).open("a") as out: out.write(" ".join(args) + "\\n")\n' 'if args[:2] == ["image", "inspect"]:\n' ' if os.environ["IMAGE_AVAILABLE"] != "true": sys.exit(1)\n' ' print(os.environ["IMAGE_SOURCE"])\n' 'elif args[0] == "ps": print("test-api")\n' 'else: sys.exit("unexpected Docker command")\n') docker.chmod(0o755) env = {**os.environ, 'TEST_EXEC_PATH': str(binary) + os.pathsep + os.environ['PATH'], 'DOCKER_TRACE': str(trace), 'IMAGE_SOURCE': inspector_source, 'IMAGE_AVAILABLE': str(inspector_available).lower()} # common initializes the tool PATH during import; select the test's # executable fixture after that initialization, then run the real CLI. entry = ('import os,sys; sys.path.insert(0,sys.argv.pop(1)); ' 'import standalone_l0_release; os.environ["PATH"]=os.environ["TEST_EXEC_PATH"]; ' 'standalone_l0_release.main()') result = subprocess.run([sys.executable, '-c', entry, str(Path(__file__).resolve().parent), '--state-root', str(self.root), '--worker-image', worker_image, '--source', SOURCE, '--runtime-container', 'test-api', '--worker-container-prefix', 'test-worker-'], env=env, capture_output=True, text=True, timeout=10) self.assertNotEqual(result.returncode, 0) self.assertFalse((self.root / 'config/runtime.env').exists()) return result.stderr, trace.read_text() if trace.exists() else '' def test_remote_owner_images_reject_before_any_docker_or_database_access(self): worker = 'ghcr.io/pingcap-inc/nexuscrm-runtime-worker@' + NEW catalog = 'ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@' + NEW for label, valid in (('Worker', worker), ('Catalog', catalog)): for invalid in ('', 'registry.invalid/foreign@' + NEW, valid.replace('@sha256:', '-foreign@sha256:'), valid.split('@')[0] + ':main', valid[:-1]): with self.subTest(role=label, image=invalid): error, calls = self.remote_cli(invalid if label == 'Worker' else worker, invalid if label == 'Catalog' else catalog) self.assertIn('remote ' + label + ' image must use the exact official repository and digest', error) self.assertEqual(calls, '') def test_remote_worker_must_exist_and_match_source_before_database_access(self): worker = 'ghcr.io/pingcap-inc/nexuscrm-runtime-worker@' + NEW catalog = 'ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@' + NEW error, calls = self.remote_cli(worker, catalog, inspector_source='f' * 40) self.assertIn('remote Worker image source differs', error) self.assertEqual(len(calls.splitlines()), 1) error, calls = self.remote_cli(worker, catalog, inspector_available=False) self.assertIn('CalledProcessError', error) self.assertEqual(len(calls.splitlines()), 1) def test_valid_remote_images_reach_the_existing_quiescence_guard(self): error, calls = self.remote_cli('ghcr.io/pingcap-inc/nexuscrm-runtime-worker@' + NEW, 'ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@' + NEW) self.assertIn('stop L0 API and all Workers', error) self.assertEqual(len(calls.splitlines()), 2) self.assertNotIn('run ', calls) def test_local_opaque_source_is_preserved_without_provenance_policy(self): self.assertEqual(self.upgrade(target_source='local'), len(L0_WORKER_STEMS)) self.assertTrue(all(json.loads(path.read_text())['worker_source_revision'] == 'local' for path in self.roles.glob('*.json'))) self.assertEqual(self.upgrade(target_source='local'), 0) def test_supported_eight_role_installation_does_not_invent_new_role(self): stem = 'permission-authoring-command-lifecycle' (self.roles / (stem + '.json')).unlink() self.db.pop('runtime.' + stem + '.v1') self.assertEqual(self.upgrade(), 8) self.assertEqual(len(list(self.roles.glob('*.json'))), 8) self.assertNotIn(stem, self.calls) def test_resume_after_cas_before_role_write_never_repeats_epoch(self): def crash(path, role): raise OSError('simulated process failure before rename') with self.assertRaises(OSError): self.upgrade(write_role=crash) self.assertEqual(len(self.calls), 1) self.upgrade() self.assertEqual(len(self.calls), len(L0_WORKER_STEMS)) self.assertTrue(all(row.ownership_epoch == 8 for row in self.db.values())) def test_partial_completed_roles_resume_without_rewriting_them(self): def write(path, role): if path.stem == L0_WORKER_STEMS[1]: raise OSError('crash') atomic_role(path, role) with self.assertRaises(OSError): self.upgrade(write_role=write) self.upgrade() self.assertEqual(len(self.calls), len(L0_WORKER_STEMS)) def test_last_role_conflict_preflights_before_any_mutation(self): family = 'runtime.' + L0_WORKER_STEMS[-1] + '.v1' self.db[family] = WorkerOwnership(family, 'other-deployment', OLD, 7, True, 'other') with self.assertRaises(DeployError): self.upgrade() self.assertEqual(self.calls, []) def test_competing_cas_stops_before_role_write(self): before = {p: p.read_bytes() for p in self.roles.glob('*.json')} def conflict(stem, request): raise DeployError('CAS conflict') with self.assertRaises(DeployError): self.upgrade(activate=conflict) self.assertEqual(before, {p: p.read_bytes() for p in before}) def test_result_and_authority_are_both_required(self): def corrupt(stem, request): result = self.activate(stem, request) result['ownershipEpoch'] = '99' return result with self.assertRaises(DeployError): self.upgrade(activate=corrupt) self.assertEqual(json.loads((self.roles / (L0_WORKER_STEMS[0] + '.json')).read_text())['ownership_epoch'], 7) # Exact committed operation is recoverable despite a lost/corrupt result. self.upgrade() self.assertTrue(all(row.ownership_epoch == 8 for row in self.db.values())) def test_unrelated_successor_is_not_accepted_as_retry(self): family = 'runtime.' + L0_WORKER_STEMS[-1] + '.v1' self.db[family] = WorkerOwnership(family, L0_WORKER_STEMS[-1], NEW, 8, True, 'other') with self.assertRaises(DeployError): self.upgrade() self.assertEqual(self.calls, []) def test_incomplete_roles_do_not_change_authority(self): (self.roles / (L0_WORKER_STEMS[-1] + '.json')).unlink() with self.assertRaises(RuntimeError): self.upgrade() self.assertEqual(self.calls, []) if __name__ == '__main__': unittest.main()