from __future__ import annotations import sys import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] WORKFLOW = (ROOT / ".github/workflows/manage-test1-apprelease.yml").read_text() IMPLEMENTATION = (ROOT / "scripts/manage_test1_apprelease.py").read_text() sys.path.insert(0, str(ROOT / "scripts")) sys.path.insert(0, str(ROOT / "deployments/offline/runtime")) from manage_test1_apprelease import ( # noqa: E402 AppReleaseError, deterministic_uuid7, exact_checkout, invalidate_worker_state, 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 ManageTest1AppReleaseTest(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="container-id", ) 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_test1_apprelease.py", WORKFLOW) self.assertNotIn("uses: ./.github/workflows/deploy-test-l0.yml", WORKFLOW) self.assertNotIn("operation: deactivate", WORKFLOW) self.assertNotIn("operation: activate", WORKFLOW) self.assertNotIn("nexuscrm-l0-test1-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("Validate AppRelease controller"), WORKFLOW.index("Checkout AppRelease controller"), ) 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-test1-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) if __name__ == "__main__": unittest.main()