from __future__ import annotations import json import subprocess import sys import tempfile import unittest from pathlib import Path from types import SimpleNamespace from scripts import serviceification_d3_engine as deploy from scripts import serviceification_deployment_tools as deployment_tools class D3EngineOperationsTests(unittest.TestCase): def test_unsigned_images_only_admit_host_test3(self): # Exercise the real admission function in isolated interpreters. The # child owns its environment and engine identity configuration, just # as the deployment entrypoint does; no function is replaced and no # parent-process environment/global configuration is modified. probe = """ import sys from types import SimpleNamespace from scripts import serviceification_d3_engine as engine engine.CONTAINER_LABEL_VALUE = sys.argv[2] try: engine.require_local_image_context(SimpleNamespace(envv={"ENV_NAME": sys.argv[1]})) except engine.DeployError as exc: print("rejected: " + str(exc)) else: print("accepted") """ for environment, label, actions, accepted in ( ("test-test3", "test3", "false", True), ("local-d3", "local", "false", False), ("test-test1", "test1", "false", False), ("test-test2", "test2", "false", False), ("staging", "staging", "false", False), ("production", "production", "false", False), ("test-test3", "local", "false", False), ("test-test3", "test3", "true", False), ): with self.subTest(environment=environment, label=label, actions=actions): result = subprocess.run( [sys.executable, "-c", probe, environment, label], cwd=Path(__file__).resolve().parents[1], env={"GITHUB_ACTIONS": actions, "PYTHONUTF8": "1"}, capture_output=True, text=True, timeout=30, check=True, ) if accepted: self.assertEqual(result.stdout.strip(), "accepted") else: self.assertTrue(result.stdout.startswith("rejected: "), result.stdout) if actions == "true": self.assertIn("host-only", result.stdout) @classmethod def seal_baseline_worker_bom( self, state: dict[str, object], root: Path, *, image: str, worker_source: str, runtime_source: str, ) -> None: repository, digest = image.split("@", 1) bom = { "schema_version": deploy.RESOLVED_SCHEMA_VERSION_V10, "l0_engine": { "provenance_digest": "sha256:" + "1" * 64, "sbom_digest": "sha256:" + "2" * 64, "release": { "source": { "repository": "pingcap-inc/NexusCRM", "commit": worker_source, }, "subjects": { "runtime-api": { "image_repository": "ghcr.io/pingcap-inc/nexuscrm-runtime-api", "image_digest": "sha256:" + "3" * 64, }, "runtime-worker": { "image_repository": repository, "image_digest": digest, }, "integration-service": { "image_repository": "ghcr.io/pingcap-inc/nexuscrm-integration-service", "image_digest": "sha256:" + "4" * 64, }, }, "catalog_reconcile": { "image_repository": "ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile", "image_digest": "sha256:" + "5" * 64, "source_sha": "6" * 40, }, }, }, "services": {"sfcrm_app_service": {}}, } path = root / "resolved-bom.json" path.write_text(json.dumps(bom), encoding="utf-8") state.update({ "runRoot": str(root), "resolvedBomFileDigest": deploy.sha256_bytes(path.read_bytes()), "prepared": {"resolvedBomCopy": str(path)}, }) def test_detached_runtime_container_overrides_image_healthcheck(self) -> None: with tempfile.TemporaryDirectory() as directory: env_file = Path(directory) / "runtime.env" env_file.write_text("NEXUSCRM_RUNTIME_API_ADMIN_ADDR=127.0.0.1:3502\n") env_file.chmod(0o600) commands: list[tuple[str, ...]] = [] deploy.run_detached_container( SimpleNamespace(run_id="d3-staging-blue-1-1"), name="runtime-api", image="runtime-api@sha256:test", env_file=env_file, mounts=(), healthcheck_url="http://127.0.0.1:3502/readyz", execute=lambda command: commands.append(tuple(command)), ) self.assertEqual(1, len(commands)) command = commands[0] health = command.index("--health-cmd") self.assertEqual( ( "--health-cmd", "wget -qO- http://127.0.0.1:3502/readyz >/dev/null || exit 1", "--health-interval", "5s", "--health-timeout", "3s", "--health-retries", "12", "--health-start-period", "5s", ), command[health : health + 10], ) self.assertLess(health, command.index("runtime-api@sha256:test")) def test_detached_runtime_container_rejects_non_loopback_healthcheck(self) -> None: with tempfile.TemporaryDirectory() as directory: env_file = Path(directory) / "runtime.env" env_file.write_text("NEXUSCRM_RUNTIME_API_ADMIN_ADDR=127.0.0.1:3502\n") env_file.chmod(0o600) with self.assertRaisesRegex( deploy.DeployError, "healthcheck URL must use one canonical loopback ready endpoint", ): deploy.run_detached_container( SimpleNamespace(run_id="d3-staging-blue-1-1"), name="runtime-api", image="runtime-api@sha256:test", env_file=env_file, mounts=(), healthcheck_url="http://127.0.0.1:3502/readyz;exit 0", execute=lambda _command: None, ) def test_detached_runtime_waits_for_docker_health_after_readiness(self) -> None: with tempfile.TemporaryDirectory() as directory: env_file = Path(directory) / "runtime.env" env_file.write_text("NEXUSCRM_RUNTIME_API_ADMIN_ADDR=127.0.0.1:3502\n") env_file.chmod(0o600) events: list[str] = [] deploy.ensure_detached_container( SimpleNamespace(run_id="d3-staging-blue-1-1"), name="runtime-api", image="runtime-api@sha256:test", env_file=env_file, mounts=(), ready_url="http://127.0.0.1:3502/readyz", healthcheck_url="http://127.0.0.1:3502/readyz", present=set(), execute=lambda _command: events.append("created"), capture_output=lambda _command: "", wait_ready=lambda _url, _name: events.append("ready"), wait_healthy=lambda _url, _name: events.append("healthy"), ) self.assertEqual(["created", "ready", "healthy"], events) def test_docker_health_probe_rejects_inherited_default_port(self) -> None: inherited = json.dumps([ "CMD-SHELL", "wget -qO- http://127.0.0.1:9091/livez >/dev/null || exit 1", ]) with self.assertRaisesRegex(deploy.DeployError, "healthcheck drifted"): deploy.wait_docker_healthy( "http://127.0.0.1:3502/readyz", "runtime-api", capture_output=lambda _command: inherited, timeout_seconds=1, ) def test_runtime_docker_health_targets_follow_all_dynamic_ports(self) -> None: workers = { stem: f"worker-{stem}" for stem in deploy.WORKER_STEMS } state = { "containers": { "runtime_api": "runtime-api", "workers": workers, "integration_service": "integration", "sfcrm_app_service": "sfcrm", }, } ports = { stem: 3510 + index for index, stem in enumerate(deploy.WORKER_STEMS) } context = SimpleNamespace( ports=SimpleNamespace(runtime_admin=3502, worker_admin=ports), ) self.assertEqual( [ ("runtime-api", "http://127.0.0.1:3502/readyz"), *( (workers[stem], f"http://127.0.0.1:{ports[stem]}/readyz") for stem in deploy.WORKER_STEMS ), ], deploy.runtime_docker_health_targets(context, state), ) def test_serving_health_fails_closed_on_runtime_docker_health(self) -> None: workers = { stem: f"worker-{stem}" for stem in deploy.WORKER_STEMS } ports = { stem: 3510 + index for index, stem in enumerate(deploy.WORKER_STEMS) } state = { "containers": { "runtime_api": "runtime-api", "workers": workers, "integration_service": "integration", "sfcrm_app_service": "sfcrm", }, } context = SimpleNamespace(images={}, ports=SimpleNamespace( runtime_admin=3502, worker_admin=ports, integration_http=3203, sfcrm_http=3101, )) events: list[tuple[str, str]] = [] count = deploy.wait_for_serving_health( context, state, wait_ready=lambda _url, name: events.append(("ready", name)), wait_healthy=lambda _url, name: events.append(("healthy", name)), ) ready = [name for event, name in events if event == "ready"] healthy = [name for event, name in events if event == "healthy"] self.assertEqual(["runtime-api", *workers.values(), "integration", "sfcrm"], ready) self.assertEqual(["runtime-api", *workers.values()], healthy) self.assertEqual(1 + len(deploy.WORKER_STEMS), count) self.assertLess(events.index(("ready", "sfcrm")), events.index(("healthy", "runtime-api"))) def test_catalog_abandonment_retry_receipt_requires_no_publication(self) -> None: operation_id = "018f0000-0000-7000-9000-000000000064" evidence = "sha256:" + "a" * 64 row = "\t".join(( operation_id.encode().hex().upper(), "FAILED", "ABANDONED", evidence, "1", "1", "0", )) def query(_env: dict[str, str], sql: str, *, database: str) -> str: self.assertEqual("nexuscrm", database) self.assertIn("o.Step", sql) self.assertIn("o.LastError", sql) self.assertIn("METADATA_SELECTION_COMMITTED", sql) self.assertIn("SERVING_ARTIFACTS_COMMITTED", sql) self.assertIn(operation_id.encode().hex(), sql) return row self.assertTrue(deploy.read_optional_catalog_abandonment_receipt( {"TIDB_DATABASE": "nexuscrm"}, operation_id, evidence, tables_present=lambda _env: True, query=query, )) def test_unpublished_upgrade_authority_requires_exact_operation_and_no_publication(self) -> None: operation_id = "018f0000-0000-7000-9000-000000000071" release = "sha256:" + "a" * 64 publication = "sha256:" + "b" * 64 row = "\t".join(( "installation-test".encode().hex().upper(), "org-test".encode().hex().upper(), release, "7", "nexuscrm-runtime-api".encode().hex().upper(), "7", "11", publication, "1", "1", )) def query(_env: dict[str, str], sql: str, *, database: str) -> str: self.assertEqual("nexuscrm", database) self.assertIn("i.State = 'UPGRADING'", sql) self.assertIn("s.ActiveOperationId = i.ActiveOperationId", sql) self.assertIn("o.Step = 'CANDIDATE_COMMITTED'", sql) self.assertIn("NOT EXISTS", sql) self.assertIn("METADATA_SELECTION_COMMITTED", sql) self.assertIn(operation_id.encode().hex(), sql) return row authority = deploy.read_optional_unpublished_upgrade_authority( {"TIDB_DATABASE": "nexuscrm"}, operation_id, tables_present=lambda _env: True, query=query, ) self.assertIsNotNone(authority) assert authority is not None self.assertEqual("installation-test", authority.app_installation_id) self.assertEqual(release, authority.active_release_digest) self.assertEqual(publication, authority.active_publication_digest) def test_abandoned_operation_uses_a_new_evidence_bound_idempotency_key(self) -> None: bom = {"app_installation": {"app_id": "io.nexuscrm.sfcrm"}} start = {"mode": "UPGRADE", "expectedCatalogRevision": 7} arguments = { "l1_version": "0.46.0", "l2_version": "1.2.3", "integration_bindings_digest": "sha256:" + "a" * 64, "outbound_mail_actions_digest": "sha256:" + "b" * 64, "approval_publication_digest": "sha256:" + "c" * 64, } original = deploy.catalog_input_idempotency_key(bom, start, **arguments) evidence = "sha256:" + "d" * 64 successor = deploy.catalog_input_idempotency_key( bom, start, **arguments, operation_attempt_digest=evidence, ) self.assertNotEqual(original, successor) self.assertTrue(successor.endswith("-a" + "d" * 64)) self.assertEqual( successor, deploy.catalog_input_idempotency_key( bom, start, **arguments, operation_attempt_digest=evidence, ), ) first_attempt = deploy.catalog_successor_attempt_digest( evidence, "sha256:" + "e" * 64, ) later_l0_attempt = deploy.catalog_successor_attempt_digest( evidence, "sha256:" + "f" * 64, ) self.assertNotEqual(first_attempt, later_l0_attempt) self.assertNotEqual( deploy.catalog_input_idempotency_key( bom, start, **arguments, operation_attempt_digest=first_attempt, ), deploy.catalog_input_idempotency_key( bom, start, **arguments, operation_attempt_digest=later_l0_attempt, ), ) def test_catalog_operation_identity_is_owned_by_the_signed_bundle(self) -> None: def context(*, l0: str = "a", app: str = "b", l2: str = "d") -> SimpleNamespace: return SimpleNamespace(bom={ "l0_engine": {"release": {"digest": "sha256:" + l0 * 64}}, "app_installation": { "app_id": "io.nexuscrm.sfcrm", "app_release_digest": "sha256:" + app * 64, "metadata_bundle_digest": "sha256:" + "c" * 64, }, "l2_metadata": {"tree_sha256": "sha256:" + l2 * 64}, }) first = context() l0_only = context(l0="e") changed_app = context(app="f") changed_l2 = context(l2="0") self.assertRegex( deploy.catalog_bundle_operation_id(first), # type: ignore[arg-type] r"^d3-app-bundle-[0-9a-f]{64}$", ) self.assertEqual( deploy.catalog_bundle_operation_id(first), # type: ignore[arg-type] deploy.catalog_bundle_operation_id(l0_only), # type: ignore[arg-type] ) self.assertNotEqual( deploy.catalog_bundle_operation_id(first), # type: ignore[arg-type] deploy.catalog_bundle_operation_id(changed_app), # type: ignore[arg-type] ) self.assertNotEqual( deploy.catalog_bundle_operation_id(first), # type: ignore[arg-type] deploy.catalog_bundle_operation_id(changed_l2), # type: ignore[arg-type] ) with self.assertRaisesRegex(deploy.DeployError, "exact appReleaseDigest"): deploy.catalog_bundle_operation_id( # type: ignore[arg-type] SimpleNamespace(bom={ "app_installation": { "app_id": "io.nexuscrm.sfcrm", "app_release_digest": "latest", "metadata_bundle_digest": "sha256:" + "c" * 64, }, "l2_metadata": {"tree_sha256": "sha256:" + "d" * 64}, }), ) def test_worker_schema_convergence_uses_exact_same_source_l0_owner(self) -> None: worker = "ghcr.io/pingcap-inc/nexuscrm-runtime-worker@sha256:" + "a" * 64 carrier = "ghcr.io/pingcap-inc/nexuscrm-runtime-api@sha256:" + "b" * 64 source = "c" * 40 commands: list[tuple[str, ...]] = [] with tempfile.TemporaryDirectory() as directory: database_env = Path(directory) / "database.env" database_env.write_text("TIDB_DATABASE=test\n") database_env.chmod(0o600) context = SimpleNamespace( images={ "runtime_worker": worker, "runtime_api": carrier, }, run_id="d3-test-run", ) state = {"catalogArtifacts": {"databaseEnv": str(database_env)}} actual = deploy.converge_worker_image_schema( context, state, worker, source, carrier, execute=lambda command: commands.append(tuple(command)), capture_output=lambda _command: source + "\n", ) self.assertEqual(source, actual) self.assertEqual(("docker", "pull", "--quiet", carrier), commands[0]) converge = commands[1] self.assertEqual((carrier, "ensure-l0-schema-ddl"), converge[-2:]) self.assertIn("--entrypoint", converge) self.assertIn("/usr/local/bin/l0-controlplane", converge) self.assertIn("--env-file", converge) def test_schema_receipt_survives_a_later_setup_implementation_repair(self) -> None: receipt = { "command": "ensure-l0-schema-ddl", "coreCommit": "a" * 40, "setupCoreRevision": "b" * 40, } self.assertTrue( deploy.runtime_l0_schema_convergence_receipt_matches( receipt, command="ensure-l0-schema-ddl", core_commit="a" * 40, ) ) self.assertFalse( deploy.runtime_l0_schema_convergence_receipt_matches( receipt, command="ensure-l0-schema-ddl", core_commit="c" * 40, ) ) self.assertFalse( deploy.runtime_l0_schema_convergence_receipt_matches( {**receipt, "setupCoreRevision": "not-an-exact-commit"}, command="ensure-l0-schema-ddl", core_commit="a" * 40, ) ) def test_deployment_tool_provenance_is_independent_of_catalog_image_commit(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) resolved_bom = root / "resolved-bom.json" resolved_bom.write_text(json.dumps({ "schema_version": "nexuscrm.serviceification-resolved-bom/v10", "l0_engine": {"release": {"catalog_reconcile": {"source_sha": "a" * 40}}}, }), encoding="utf-8") tool_root = root / "tools" tool_root.mkdir() receipts: dict[str, object] = {} for key, name in deployment_tools.FILES.items(): path = tool_root / name path.write_bytes(key.encode("utf-8")) path.chmod(0o555) receipts[key] = { "path": name, "sha256": deployment_tools.sha256_file(path), "size": path.stat().st_size, } manifest = { "schema": deployment_tools.SCHEMA, "platform": "linux/amd64", "builderImage": deployment_tools.BUILDER_IMAGE, "coreRevision": "b" * 40, "controlplaneRevision": "c" * 40, "appSourceRevision": "d" * 40, "appReleaseRevision": "e" * 40, "setupCommands": list(deployment_tools.REQUIRED_SETUP_COMMANDS), "files": receipts, } (tool_root / "manifest.json").write_text( deployment_tools.canonical_json(manifest), encoding="utf-8", ) actual = deployment_tools.verify_tools( root=tool_root, resolved_bom=resolved_bom, ) self.assertEqual("c" * 40, actual["controlplaneRevision"]) self.assertNotEqual( "a" * 40, actual["controlplaneRevision"], ) def test_worker_schema_convergence_rejects_source_mismatch_before_ddl(self) -> None: worker = "ghcr.io/pingcap-inc/nexuscrm-runtime-worker@sha256:" + "a" * 64 carrier = "ghcr.io/pingcap-inc/nexuscrm-runtime-api@sha256:" + "b" * 64 commands: list[tuple[str, ...]] = [] with tempfile.TemporaryDirectory() as directory: database_env = Path(directory) / "database.env" database_env.write_text("TIDB_DATABASE=test\n") database_env.chmod(0o600) context = SimpleNamespace( images={ "runtime_worker": worker, "runtime_api": carrier, }, run_id="d3-test-run", ) state = {"catalogArtifacts": {"databaseEnv": str(database_env)}} with self.assertRaisesRegex( deploy.DeployError, "must have the same source revision", ): deploy.converge_worker_image_schema( context, state, worker, "c" * 40, carrier, execute=lambda command: commands.append(tuple(command)), capture_output=lambda _command: "d" * 40, ) self.assertEqual([("docker", "pull", "--quiet", carrier)], commands) def test_record_command_cadence_changes_only_the_explicit_poll_field(self) -> None: current = { "schema": "nexuscrm.runtime-worker-role/v3", "work_family": "runtime.record-command-lifecycle.v1", "poll_interval": "1s", "batch_size": 10, "db_max_open_conns": 4, "db_max_idle_conns": 2, } candidate, patch = deploy.worker_runtime_config_candidate( current, "record-command-lifecycle", poll_interval="30s", ) self.assertEqual({"poll_interval": "30s"}, patch) self.assertEqual("30s", candidate["poll_interval"]) self.assertEqual( {key: value for key, value in current.items() if key != "poll_interval"}, {key: value for key, value in candidate.items() if key != "poll_interval"}, ) self.assertEqual("1s", current["poll_interval"]) def test_record_command_cadence_rejects_wrong_role_and_unbounded_value(self) -> None: current = { "schema": "nexuscrm.runtime-worker-role/v3", "work_family": "runtime.record-command-lifecycle.v1", "poll_interval": "1s", } with self.assertRaises(deploy.DeployError): deploy.worker_runtime_config_candidate( current, "record-command-lifecycle", poll_interval="61s", ) with self.assertRaises(deploy.DeployError): deploy.validate_record_command_poll_interval("61s") with self.assertRaises(deploy.DeployError): deploy.worker_runtime_config_candidate( { **current, "work_family": "runtime.record-event-dispatch.v1", }, "record-event-dispatch", poll_interval="30s", ) def test_runtime_tuning_follows_the_effective_worker_image(self) -> None: base = "ghcr.io/pingcap-inc/nexuscrm-runtime-worker@sha256:" + "a" * 64 effective = "ghcr.io/pingcap-inc/nexuscrm-runtime-worker@sha256:" + "b" * 64 context = SimpleNamespace( bom={"schema_version": deploy.RESOLVED_SCHEMA_VERSION_V10}, images={"runtime_worker": base}, ) state = {"workerImageOverride": effective} image, rebound_from = deploy.resolve_worker_runtime_config_image( context, state, {"runtimeWorkerImage": effective}, ) self.assertEqual(effective, image) self.assertIsNone(rebound_from) image, rebound_from = deploy.resolve_worker_runtime_config_image( context, state, {"runtimeWorkerImage": base}, ) self.assertEqual(effective, image) self.assertEqual(base, rebound_from) third = "ghcr.io/pingcap-inc/nexuscrm-runtime-worker@sha256:" + "c" * 64 with self.assertRaises(deploy.DeployError): deploy.resolve_worker_runtime_config_image( context, state, {"runtimeWorkerImage": third}, ) def test_baseline_worker_provenance_does_not_require_a_replacement(self) -> None: source = "d" * 40 image = "ghcr.io/pingcap-inc/nexuscrm-runtime-worker@sha256:" + "a" * 64 state = { "runId": "d3-test-run", "images": {"runtime_worker": image}, "exactSources": {"coreCommit": "c" * 40}, "workerRoleConfigDigests": { stem: "sha256:" + format(index + 1, "064x") for index, stem in enumerate(deploy.WORKER_STEMS) }, "workerOwnershipAfter": { family: { "workFamily": family, "active": True, "workerReleaseDigest": "sha256:" + "a" * 64, } for family in deploy.WORKER_FAMILIES }, } with tempfile.TemporaryDirectory() as directory: self.seal_baseline_worker_bom( state, Path(directory), image=image, worker_source=source, runtime_source="c" * 40, ) actual = deploy.sanitized_worker_serving_provenance(state) self.assertEqual("resolved-bom-baseline", actual["lineageKind"]) self.assertEqual(image, actual["servingImage"]) self.assertEqual(source, actual["servingSourceRevision"]) self.assertNotIn("imageReplacement", actual) def test_baseline_worker_provenance_rejects_ownership_drift(self) -> None: source = "c" * 40 image = "ghcr.io/pingcap-inc/nexuscrm-runtime-worker@sha256:" + "a" * 64 ownership = { family: { "workFamily": family, "active": True, "workerReleaseDigest": "sha256:" + "a" * 64, } for family in deploy.WORKER_FAMILIES } drifted_family = next(iter(deploy.WORKER_FAMILIES)) ownership[drifted_family]["workerReleaseDigest"] = "sha256:" + "b" * 64 state = { "runId": "d3-test-run", "images": {"runtime_worker": image}, "exactSources": {"coreCommit": source}, "workerRoleConfigDigests": { stem: "sha256:" + format(index + 1, "064x") for index, stem in enumerate(deploy.WORKER_STEMS) }, "workerOwnershipAfter": ownership, } with tempfile.TemporaryDirectory() as directory: self.seal_baseline_worker_bom( state, Path(directory), image=image, worker_source=source, runtime_source="b" * 40, ) with self.assertRaisesRegex(deploy.DeployError, "serving receipt disagrees"): deploy.sanitized_worker_serving_provenance(state) def test_future_d3_runs_use_the_same_record_command_cadence(self) -> None: bom = Path("release/serviceification-d3-bom.json").read_text() self.assertIn( '"work_family": "runtime.record-command-lifecycle.v1", "config_schema": "nexuscrm.runtime-worker-role/v3", "batch_size": 10, "poll_interval": "30s"', bom, ) def test_pack_builder_resolves_relative_git_dir_from_its_checkout(self) -> None: repository = Path.cwd().resolve() with tempfile.TemporaryDirectory() as temporary: clone = Path(temporary) / "nested" / "publisher" clone.parent.mkdir() subprocess.run( ["git", "clone", "--quiet", "--no-local", str(repository), str(clone)], check=True, ) clone_script = clone / "scripts/build_pingcap_salesforce_sample_pack.sh" clone_script.write_bytes( (repository / "scripts/build_pingcap_salesforce_sample_pack.sh").read_bytes() ) result = subprocess.run( ["bash", str(clone_script), "--help"], cwd=clone.parent, text=True, capture_output=True, check=False, ) self.assertEqual(0, result.returncode, result.stderr) self.assertIn("--official-app-packs-ref", result.stdout) if __name__ == "__main__": unittest.main()