from __future__ import annotations import json import os import subprocess import sys import tempfile import unittest from pathlib import Path import yaml ROOT = Path(__file__).resolve().parents[1] WORKFLOW = (ROOT / ".github/workflows/manage-test2-apprelease.yml").read_text() IMPLEMENTATION = (ROOT / "scripts/manage_test2_apprelease.py").read_text() sys.path.insert(0, str(ROOT / "scripts")) sys.path.insert(0, str(ROOT / "deployments/offline/runtime")) from manage_test2_apprelease import ( # noqa: E402 AppReleaseError, deterministic_uuid7, exact_checkout, invalidate_worker_state, load_execution_bom, require_no_active_d3, require_published_authority, select_catalog_start, ) from render_serviceification_test_stack import ( # noqa: E402 ValidationFailure, runtime_worker_release_identity, ) from serviceification_d3_engine import CatalogAuthority # noqa: E402 def authority( release: str = "sha256:" + "1" * 64, *, revision: int = 7, security_epoch: int = 8, ) -> CatalogAuthority: return CatalogAuthority( app_installation_id="installation-1", org_id="org-1", active_release_digest=release, catalog_revision=revision, runtime_audience="nexuscrm-runtime-api", security_epoch=security_epoch, active_publication_digest="sha256:" + "2" * 64, schema_generation=9, engine_generation=10, ) class ManageTest2AppReleaseTest(unittest.TestCase): def test_worker_state_is_invalidated_as_one_fail_closed_set(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) stems = ( "audit-history-fanout", "bulk-ingest", "file-upload-lifecycle", "permission-authoring-command-lifecycle", "platform-command", "record-command-lifecycle", "record-event-dispatch", "rollup-delta", "scheduled-flow", "layout-authoring", ) for stem in stems: role = root / f"config/workers/{stem}.json" token = root / f"secrets/workers/{stem}.token" role.parent.mkdir(parents=True, exist_ok=True) token.parent.mkdir(parents=True, exist_ok=True) role.write_text("{}") token.write_text("token") invalidate_worker_state(root) for stem in stems: self.assertFalse((root / f"config/workers/{stem}.json").exists()) token = root / f"secrets/workers/{stem}.token" if stem == "layout-authoring": self.assertFalse(token.exists()) else: self.assertTrue(token.is_file()) def test_apprelease_refuses_active_d3_before_mutation(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) require_no_active_d3({"ENV_ROOT": str(root)}, process_inventory=[]) active = root / ".runtime/serviceification-d3/active.json" active.parent.mkdir(parents=True) active.write_text("{}") with self.assertRaisesRegex(AppReleaseError, "active full-D3"): require_no_active_d3({"ENV_ROOT": str(root)}, process_inventory=[]) active.unlink() with self.assertRaisesRegex(AppReleaseError, "process inventory"): require_no_active_d3( {"ENV_ROOT": str(root)}, process_inventory=[{"Name": "/old-runtime"}], ) def test_deployed_worker_identity_does_not_rewrite_bom_policy(self) -> None: original = { "services": {"runtime_worker": { "digest": "sha256:" + "1" * 64, "source_commit": "2" * 40, }}, } selected = { "digest": "sha256:" + "3" * 64, "source_commit": "4" * 40, } self.assertEqual(selected, runtime_worker_release_identity(original, selected)) self.assertEqual("sha256:" + "1" * 64, original["services"]["runtime_worker"]["digest"]) with self.assertRaisesRegex(ValidationFailure, "unknown shape"): runtime_worker_release_identity(original, {**selected, "repository": "extra"}) def test_exact_checkout_rejects_a_symlink_before_resolving_it(self) -> None: with tempfile.TemporaryDirectory() as raw: root = Path(raw) target = root / "target" target.mkdir() checkout = root / "checkout" checkout.symlink_to(target, target_is_directory=True) with self.assertRaisesRegex(AppReleaseError, "incomplete"): exact_checkout(checkout, "0" * 40, "L2") def test_install_creates_the_first_catalog_authority(self) -> None: start = select_catalog_start( "install", None, release_digest="sha256:" + "1" * 64, installation_org_id="org-1", ) self.assertEqual("PREPARE", start.kind) self.assertEqual(0, start.catalog_revision) self.assertFalse(start.refresh_active_release) def test_install_retry_reuses_the_normal_same_release_upgrade(self) -> None: start = select_catalog_start( "install", authority(), release_digest="sha256:" + "1" * 64, installation_org_id="org-1", ) self.assertEqual("UPGRADE", start.kind) self.assertTrue(start.refresh_active_release) def test_install_rejects_a_different_existing_installation(self) -> None: with self.assertRaisesRegex(AppReleaseError, "use upgrade"): select_catalog_start( "install", authority(), release_digest="sha256:" + "3" * 64, installation_org_id="org-1", ) def test_upgrade_reuses_exact_database_authority(self) -> None: start = select_catalog_start( "upgrade", authority(), release_digest="sha256:" + "3" * 64, installation_org_id="ignored", ) self.assertEqual("UPGRADE", start.kind) self.assertEqual("installation-1", start.app_installation_id) self.assertEqual(7, start.catalog_revision) self.assertEqual(8, start.security_epoch) self.assertFalse(start.refresh_active_release) def test_same_release_upgrade_uses_the_supported_refresh_path(self) -> None: release = "sha256:" + "1" * 64 self.assertTrue(select_catalog_start( "upgrade", authority(release), release_digest=release, installation_org_id="ignored", ).refresh_active_release) def test_catalog_readback_must_match_every_serving_axis(self) -> None: release = "sha256:" + "1" * 64 output = { "appInstallationId": "installation-1", "orgId": "org-1", "catalogRevision": 8, "securityEpoch": 9, "schemaGeneration": 9, "engineGeneration": 10, } start = select_catalog_start( "upgrade", authority(release), release_digest=release, installation_org_id="ignored", ) require_published_authority( authority(release, revision=8, security_epoch=9), output, release, start, ) for field, wrong in ( ("appInstallationId", "other-installation"), ("catalogRevision", 11), ("securityEpoch", 12), ("schemaGeneration", 13), ("engineGeneration", 14), ): with self.subTest(field=field): changed = dict(output) changed[field] = wrong with self.assertRaisesRegex(AppReleaseError, "starting authority"): require_published_authority( authority(release, revision=8, security_epoch=9), changed, release, start, ) wrong_audience = authority(release, revision=8, security_epoch=9) wrong_audience = CatalogAuthority( **{ **wrong_audience.__dict__, "runtime_audience": "other-runtime", } ) with self.assertRaisesRegex(AppReleaseError, "runtime_audience"): require_published_authority( wrong_audience, output, release, start, ) def test_operation_identity_is_deterministic(self) -> None: first = deterministic_uuid7(b"same release") self.assertEqual(first, deterministic_uuid7(b"same release")) self.assertNotEqual(first, deterministic_uuid7(b"other release")) self.assertEqual("7", first.split("-")[2][0]) def test_workflow_has_one_public_operation_axis(self) -> None: self.assertIn("- install", WORKFLOW) self.assertIn("- upgrade", WORKFLOW) self.assertNotIn("operation: deploy", WORKFLOW) self.assertNotIn("apply all", WORKFLOW.lower()) self.assertNotIn("patch_dir", WORKFLOW) self.assertNotIn("refresh", WORKFLOW.lower()) def test_workflow_keeps_l1_and_l2_release_inputs_separate(self) -> None: dispatch = WORKFLOW.split("permissions:", 1)[0] self.assertIn("apprelease_reference:", dispatch) self.assertIn("l2_installation_reference:", dispatch) self.assertNotIn("resolved_bom_reference:", dispatch) self.assertIn("nexuscrm-sfcrm-apprelease@sha256", WORKFLOW) self.assertIn("nexuscrm-serviceification-bom@sha256", WORKFLOW) self.assertIn("apprelease-publication/v1", WORKFLOW) self.assertIn("source.commit", WORKFLOW) def test_workflow_uses_controlplane_but_never_operates_l0(self) -> None: self.assertIn("manage_test2_apprelease.py", WORKFLOW) self.assertNotIn("uses: ./.github/workflows/deploy-test2-l0.yml", WORKFLOW) self.assertNotIn("operation: deactivate", WORKFLOW) self.assertNotIn("operation: activate", WORKFLOW) self.assertNotIn("nexuscrm-l0-test2-runtime-api", WORKFLOW) self.assertNotIn("docker restart", WORKFLOW) self.assertNotIn("docker rm --force \"${L0", WORKFLOW) def test_candidate_is_verified_before_apply(self) -> None: verify = WORKFLOW.index("Validate complete AppRelease source closure") apply = WORKFLOW.index("Apply AppRelease through Controlplane") self.assertLess(verify, apply) self.assertIn('cmp "${release_root}/app-package.json"', WORKFLOW) self.assertIn('cmp "${release_root}/publisher-attestation.json"', WORKFLOW) self.assertIn('cmp "${release_root}/publisher-public-key.json"', WORKFLOW) self.assertNotIn("go -C preflight/app-packs", WORKFLOW) def test_single_job_is_main_only_and_has_no_handoff_artifacts(self) -> None: self.assertEqual(1, WORKFLOW.count("runs-on:")) job = WORKFLOW.split("jobs:", 1)[1].split("steps:", 1)[0] self.assertIn("if: ${{ github.ref == 'refs/heads/main' }}", job) self.assertLess( WORKFLOW.index("Checkout reviewed Test2 controller"), WORKFLOW.index("Require standalone Test2 ownership"), ) self.assertIn("refs/heads/main", WORKFLOW) self.assertNotIn("upload-artifact", WORKFLOW) self.assertNotIn("download-artifact", WORKFLOW) def test_preflight_reuses_lineage_owner_and_preserves_source_history(self) -> None: verifier = WORKFLOW.index("verify_serviceification_d1_metadata_authority.py") mutation = WORKFLOW.index("Apply AppRelease through Controlplane") self.assertLess(verifier, mutation) app_checkout = WORKFLOW.split( "Checkout exact AppRelease source for preflight", 1, )[1].split("Checkout exact L2 source for preflight", 1)[0] l2_checkout = WORKFLOW.split( "Checkout exact L2 source for preflight", 1, )[1].split("Validate complete AppRelease source closure", 1)[0] self.assertIn("fetch-depth: 0", app_checkout) self.assertIn("fetch-depth: 0", l2_checkout) def test_workflow_does_not_select_or_replace_an_l0_image(self) -> None: self.assertNotIn("RUNTIME_API_IMAGE", WORKFLOW) self.assertNotIn("l0_source", WORKFLOW) self.assertNotIn("standalone-", WORKFLOW) self.assertIn("L0_STATE_ROOT", WORKFLOW) self.assertIn("host-compatibility.json", IMPLEMENTATION) self.assertIn("require_l0_stopped_or_bootstrap", IMPLEMENTATION) self.assertIn('"docker", "container", "inspect"', IMPLEMENTATION) self.assertNotIn('"docker", "stop"', IMPLEMENTATION) self.assertNotIn("docker restart", IMPLEMENTATION) def test_apprelease_prepares_but_does_not_start_the_worker_closure(self) -> None: self.assertIn("prepare_worker_state(", IMPLEMENTATION) self.assertIn("render_worker_activation_inputs(", IMPLEMENTATION) self.assertIn("render_worker_role_inputs(", IMPLEMENTATION) self.assertNotIn("render_layout_authoring_catalog_template(", IMPLEMENTATION) self.assertIn("runtime_worker_release_identity(bom", IMPLEMENTATION) self.assertIn("worker_release=worker_release", IMPLEMENTATION) self.assertNotIn("worker_bom", IMPLEMENTATION) self.assertIn('f"nexuscrm-l0-test2-worker-{stem}"', IMPLEMENTATION) self.assertNotIn('state_root / "config/l0-release.json"', IMPLEMENTATION) invalidation = IMPLEMENTATION.index("invalidate_worker_state(state_root)") serving = IMPLEMENTATION.index("prepare_l0_state(serving_path, state_root") workers = IMPLEMENTATION.index("prepare_worker_state(\n bom", serving) self.assertLess(invalidation, serving) self.assertLess(serving, workers) self.assertIn('"work-ownership"', IMPLEMENTATION) self.assertNotIn('"docker", "restart"', IMPLEMENTATION) self.assertNotIn('"docker", "stop"', IMPLEMENTATION) self.assertNotIn('"docker", "rm", "--force", L0_CONTAINER', IMPLEMENTATION) def test_sensitive_catalog_evidence_has_an_explicit_lifetime(self) -> None: self.assertIn("with tempfile.TemporaryDirectory", IMPLEMENTATION) self.assertNotIn("tempfile.mkdtemp", IMPLEMENTATION) def test_each_workflow_run_owns_and_cleans_one_transient_input_root(self) -> None: self.assertIn( "APPRELEASE_INPUT_NAME: apprelease-input-${{ github.run_id }}-${{ github.run_attempt }}", WORKFLOW, ) self.assertNotIn("runner.temp", WORKFLOW) self.assertNotIn('${RUNNER_TEMP}/apprelease-input/', WORKFLOW) self.assertIn("Clean transient AppRelease inputs", WORKFLOW) self.assertIn('rm -rf -- "${input_root}"', WORKFLOW) def test_catalog_image_comes_from_the_raw_l0_release_shape(self) -> None: self.assertIn( ".l0_engine.release.catalog_reconcile.image_repository", WORKFLOW, ) self.assertIn( ".l0_engine.release.catalog_reconcile.image_digest", WORKFLOW, ) self.assertNotIn(".jobs.catalog_reconcile.image", WORKFLOW) def test_static_preconditions_precede_pack_apply_and_generation_binding_follows_it(self) -> None: body = IMPLEMENTATION.split("def apply_apprelease", 1)[1] mutation = body.index("run_catalog_offline_candidate_phase(") for evidence in ( "require_no_active_d3(envv)", "selected_l0_worker_release(L0_WORKER_STEMS)", "runtime_worker_release_identity(bom", "copy_image_file(", 'load_json(approval, "Approval publication")', "write_database_env(", "require_l0_stopped_or_bootstrap()", "require_staging_target_runtime_quiesced(", "validate_apprelease_compatibility(", ): with self.subTest(evidence=evidence): self.assertLess(body.index(evidence), mutation) self.assertIn( "for stem in (*L0_WORKER_STEMS, *RETIRED_L0_WORKER_STEMS)", IMPLEMENTATION, ) generation_binding = body.index( "render_serviceification_d3_integration_bindings.py", ) catalog_request = body.index("render_catalog_config(") catalog_publish = body.index('"docker", "run", "--rm", "--network", "host"') self.assertLess(mutation, generation_binding) self.assertLess(generation_binding, catalog_request) self.assertLess(catalog_request, catalog_publish) self.assertIn('run(("docker", "pull", "--quiet", image))', IMPLEMENTATION) class CatalogPublisherSelectionTest(unittest.TestCase): repository = "ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile" commit = "a" * 40 old_image = repository + "@sha256:" + "1" * 64 new_image = repository + "@sha256:" + "2" * 64 def selected_step(self) -> dict: steps = yaml.safe_load(WORKFLOW)["jobs"]["install-apprelease"]["steps"] selected = [step for step in steps if step.get("id") == "catalog"] self.assertEqual(1, len(selected), "Catalog publisher needs an independent verified selection") return selected[0] def select(self, commit: str, *, fail: bool = False, receipt: dict | None = None): body = self.selected_step()["run"] if receipt is None: receipt = { "oci_repository": self.repository, "digest": "sha256:" + "2" * 64, "source_repository": "pingcap-inc/nexuscrm-controlplane", "source_commit": self.commit, } with tempfile.TemporaryDirectory() as directory: root = Path(directory) verifier = root / "scripts/verify_serviceification_oci_publication.sh" verifier.parent.mkdir() # Only substitute the existing cryptographic verifier's process boundary. # Run the actual workflow shell, including its failure propagation. verifier.write_text(f"#!{sys.executable}\n" + """ import json, os, pathlib, sys pathlib.Path('verifier-args.json').write_text(json.dumps(sys.argv[1:])) if os.environ['FAIL_VERIFICATION'] == '1': sys.exit(1) pathlib.Path(sys.argv[sys.argv.index('--output') + 1]).write_text(os.environ['RECEIPT']) """) verifier.chmod(0o755) output = root / "output" summary = root / "summary" result = subprocess.run( ["bash", "-c", body + '\nprintf reached > apply-reached\n'], cwd=root, text=True, capture_output=True, timeout=10, env={**os.environ, "CATALOG_SOURCE_COMMIT": commit, "DEFAULT_CATALOG_IMAGE": self.old_image, "RUNNER_TEMP": directory, "APPRELEASE_INPUT_NAME": "inputs", "GITHUB_OUTPUT": str(output), "GITHUB_STEP_SUMMARY": str(summary), "FAIL_VERIFICATION": "1" if fail else "0", "RECEIPT": json.dumps(receipt)}, ) calls = root / "verifier-args.json" return (result, output.read_text() if output.exists() else "", json.loads(calls.read_text()) if calls.exists() else None, (root / "apply-reached").exists()) def test_no_selection_preserves_the_signed_installation_default(self) -> None: result, output, calls, reached = self.select("") self.assertEqual(0, result.returncode, result.stderr) self.assertEqual(f"catalog_image={self.old_image}\n", output) self.assertIsNone(calls) self.assertTrue(reached) def test_verified_publication_replaces_only_catalog_selection(self) -> None: result, output, calls, reached = self.select(self.commit) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual(f"catalog_image={self.new_image}\n", output) arguments = dict(zip(calls[::2], calls[1::2])) self.assertEqual("exact/catalog-publisher", arguments["--owner-root"]) self.assertEqual("pingcap-inc/nexuscrm-controlplane", arguments["--source-repository"]) self.assertEqual(self.commit, arguments["--source-commit"]) self.assertEqual(self.repository, arguments["--oci-repository"]) self.assertEqual("catalog_reconcile", arguments["--role"]) self.assertTrue(reached) def test_failed_verification_never_falls_back_or_reaches_apply(self) -> None: result, output, calls, reached = self.select(self.commit, fail=True) self.assertNotEqual(0, result.returncode) self.assertIsNotNone(calls) self.assertEqual("", output) self.assertFalse(reached) def test_selection_rejects_non_exact_commits_before_verification(self) -> None: for commit in ("main", "v0.2.109", "a" * 39, "a" * 40 + "\n", "$(touch bad)"): with self.subTest(commit=commit): result, output, calls, reached = self.select(commit) self.assertNotEqual(0, result.returncode) self.assertEqual("", output) self.assertIsNone(calls) self.assertFalse(reached) def test_invalid_verifier_output_never_falls_back(self) -> None: good = {"oci_repository": self.repository, "digest": "sha256:" + "2" * 64, "source_repository": "pingcap-inc/nexuscrm-controlplane", "source_commit": self.commit} for key, value in (("oci_repository", "ghcr.io/other/image"), ("digest", "latest"), ("source_repository", "other/repo"), ("source_commit", "b" * 40)): with self.subTest(key=key): result, output, _, reached = self.select(self.commit, receipt={**good, key: value}) self.assertNotEqual(0, result.returncode) self.assertEqual("", output) self.assertFalse(reached) def test_workflow_passes_the_verified_selection_to_the_existing_installer(self) -> None: value = yaml.safe_load(WORKFLOW) # PyYAML's YAML 1.1 reader parses the Actions 'on' key as True. dispatch = value[True]["workflow_dispatch"]["inputs"] self.assertFalse(dispatch["catalog_source_commit"]["required"]) job = value["jobs"]["install-apprelease"] self.assertEqual("${{ inputs.catalog_source_commit }}", job["env"]["CATALOG_SOURCE_COMMIT"]) selected = self.selected_step() self.assertEqual("${{ steps.release.outputs.catalog_image }}", selected["env"]["DEFAULT_CATALOG_IMAGE"]) steps = job["steps"] checkout = next(step for step in steps if step.get("with", {}).get("path") == "exact/catalog-publisher") self.assertEqual("${{ inputs.catalog_source_commit != '' }}", checkout["if"]) self.assertEqual("${{ inputs.catalog_source_commit }}", checkout["with"]["ref"]) self.assertEqual("pingcap-inc/nexuscrm-controlplane", checkout["with"]["repository"]) apply = next(step for step in steps if step.get("id") == "apply") self.assertIn('--catalog-image "${{ steps.catalog.outputs.catalog_image }}"', apply["run"]) self.assertNotIn("steps.release.outputs.catalog_image", apply["run"]) self.assertLess(steps.index(checkout), steps.index(selected)) self.assertLess(steps.index(selected), steps.index(apply)) validation = next(step for step in steps if step["name"] == "Validate immutable request") self.assertIn("CATALOG_SOURCE_COMMIT", validation["run"]) self.assertLess(steps.index(validation), steps.index(checkout)) class IndependentSourceAuthorityTest(unittest.TestCase): """Exercise the controller's real sealed-BOM read path, without Docker mocks.""" fixture = ROOT / "tests/fixtures/serviceification-independent-sources-v10.json" def test_frontend_and_metadata_keep_their_independent_source_bindings(self) -> None: sealed = json.loads(self.fixture.read_text()) self.assertNotEqual( sealed["frontend"]["source_tuple"]["l2_commit"], sealed["l2_metadata"]["commit"], ) execution = load_execution_bom(self.fixture) self.assertEqual(execution["frontend"], sealed["frontend"]) self.assertEqual(execution["l2_metadata"], sealed["l2_metadata"]) self.assertEqual(execution["l0_engine"], sealed["l0_engine"]) def test_each_source_role_is_bound_by_the_sealed_desired_digest(self) -> None: for fields, replacement in ( (("frontend", "source_tuple", "l2_commit"), "0" * 40), (("l2_metadata", "commit"), "0" * 40), (("l2_metadata", "tree_sha256"), "sha256:" + "0" * 64), ): with self.subTest(fields=fields), tempfile.TemporaryDirectory() as directory: changed = json.loads(self.fixture.read_text()) parent = changed for field in fields[:-1]: parent = parent[field] parent[fields[-1]] = replacement path = Path(directory) / "changed.json" path.write_text(json.dumps(changed)) with self.assertRaisesRegex(ValidationFailure, "desired_bom_sha256"): load_execution_bom(path) if __name__ == "__main__": unittest.main()