#!/usr/bin/env python3 from __future__ import annotations import copy import hashlib import json import subprocess import tempfile import unittest from pathlib import Path from scripts.prod_promotion_contract import ( METADATA_FIRST_SHARED_STATE_POLICY, RUNTIME_ONLY_SHARED_STATE_POLICY, SHARED_STATE_INPUTS, build_active_baseline_comparison, build_prod_promotion_contract, build_prod_promotion_contract_from_git, final_pack_payload_hash_from_git, require_prod_baseline_transition, require_prod_promotion_contract, ) from scripts.prod_migrated_baseline_receipt import ( build_migrated_baseline_receipt, require_bound_migrated_baseline_receipt, require_migrated_baseline_receipt, ) FINAL_PACK_HASH = "a" * 64 OFFICIAL_COMMIT = "b" * 40 class ProdPromotionContractTests(unittest.TestCase): def make_repo(self, root: Path, *, marker: str = "v1") -> None: for relative_path in SHARED_STATE_INPUTS: path = root / relative_path if path.suffix: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(f"{relative_path}:{marker}\n") else: path.mkdir(parents=True, exist_ok=True) (path / "contract-input.txt").write_text( f"{relative_path}:{marker}\n" ) (root / "stack.lock.json").write_text( json.dumps( { "pack_build": { "final_pack_path": ( "packs/NOT-EDIT-final-merged-packs" ) } }, indent=2, sort_keys=True, ) + "\n" ) final_pack = root / "packs" / "NOT-EDIT-final-merged-packs" final_pack.mkdir(parents=True, exist_ok=True) (final_pack / "pack.json").write_text( json.dumps( { "name": "salesforce-pingcap-final", "version": "1.2.3", }, indent=2, sort_keys=True, ) + "\n" ) (final_pack / ".build-lineage.json").write_text( json.dumps( {"officialPack": {"ref": OFFICIAL_COMMIT}}, indent=2, sort_keys=True, ) + "\n" ) def build(self, root: Path) -> dict: return build_prod_promotion_contract( root, final_pack_hash=FINAL_PACK_HASH, final_pack_name="salesforce-pingcap-final", final_pack_version="1.2.3", official_app_packs_commit=OFFICIAL_COMMIT, ) def test_contract_is_path_independent_and_validates(self) -> None: with tempfile.TemporaryDirectory() as first_tmp, tempfile.TemporaryDirectory() as second_tmp: first = Path(first_tmp) second = Path(second_tmp) self.make_repo(first) self.make_repo(second) first_contract = self.build(first) second_contract = self.build(second) self.assertEqual(first_contract, second_contract) self.assertEqual( require_prod_promotion_contract(first_contract), first_contract, ) self.assertEqual( first_contract["shared_state_policy"], METADATA_FIRST_SHARED_STATE_POLICY, ) self.assertIn( "runtime-standby-health", first_contract["allowed_prod_effects"], ) self.assertIn( "setup-or-pack-lifecycle-apply", first_contract["allowed_prod_effects"], ) self.assertIn( "deploy-seed-or-reference-data-apply", first_contract["forbidden_prod_effects"], ) self.assertNotIn(first_tmp, str(first_contract)) with self.assertRaisesRegex( ValueError, "final pack does not match the candidate", ): require_prod_promotion_contract( first_contract, expected_final_pack_hash="f" * 64, ) with self.assertRaisesRegex( ValueError, "official app-packs source does not match the candidate", ): require_prod_promotion_contract( first_contract, expected_official_app_packs_commit="f" * 40, ) different_official_source = build_prod_promotion_contract( first, final_pack_hash=FINAL_PACK_HASH, final_pack_name="salesforce-pingcap-final", final_pack_version="1.2.3", official_app_packs_commit="c" * 40, ) self.assertEqual( first_contract["shared_state_digest"], different_official_source["shared_state_digest"], ) self.assertNotEqual( first_contract["contract_sha256"], different_official_source["contract_sha256"], ) def test_metadata_first_effects_are_the_default_v3_policy(self) -> None: with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) contract = build_prod_promotion_contract( root, final_pack_hash=FINAL_PACK_HASH, final_pack_name="salesforce-pingcap-final", final_pack_version="1.2.3", official_app_packs_commit=OFFICIAL_COMMIT, ) self.assertEqual(contract["schema_version"], "ordinary-prod-promotion/v3") self.assertEqual( contract["shared_state_policy"], METADATA_FIRST_SHARED_STATE_POLICY, ) self.assertIn( "setup-or-pack-lifecycle-apply", contract["allowed_prod_effects"], ) self.assertIn( "release-actor-shared-state-write", contract["allowed_prod_effects"], ) self.assertIn( "connector-replay-or-admission-pause", contract["allowed_prod_effects"], ) self.assertIn( "deploy-seed-or-reference-data-apply", contract["forbidden_prod_effects"], ) self.assertEqual(require_prod_promotion_contract(contract), contract) def test_each_l2_shared_state_input_change_changes_contract(self) -> None: for relative_path in SHARED_STATE_INPUTS: with self.subTest(relative_path=relative_path): with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) before = self.build(root) input_path = root / relative_path marker = ( input_path if input_path.is_file() else input_path / "contract-input.txt" ) marker.write_text("changed\n") after = self.build(root) self.assertNotEqual( before["components"]["l2_shared_state_inputs"]["digest"], after["components"]["l2_shared_state_inputs"]["digest"], ) self.assertNotEqual( before["shared_state_digest"], after["shared_state_digest"], ) self.assertNotEqual( before["contract_sha256"], after["contract_sha256"], ) def test_reference_data_and_owner_reconcile_inputs_are_governed(self) -> None: required_mutation_inputs = { "deployments/offline/runtime/reference-data", "scripts/apply_pack_reference_data.py", "scripts/apply_pack_reference_data.sh", "deployments/offline/runtime/reconcile_hubspot_owner_directory.py", } self.assertTrue(required_mutation_inputs.issubset(set(SHARED_STATE_INPUTS))) with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) contract = self.build(root) self.assertTrue( required_mutation_inputs.issubset( set( contract["components"]["l2_shared_state_inputs"]["paths"] ) ) ) def test_tampering_or_unknown_fields_fail_closed(self) -> None: with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) contract = self.build(root) tampered = copy.deepcopy(contract) tampered["components"]["final_pack"]["payload_sha256"] = "c" * 64 with self.assertRaisesRegex(ValueError, "shared_state_digest mismatch"): require_prod_promotion_contract(tampered) unknown = copy.deepcopy(contract) unknown["allow_setup_apply"] = True with self.assertRaisesRegex(ValueError, "unexpected or missing fields"): require_prod_promotion_contract(unknown) def test_missing_or_symlinked_input_fails_closed(self) -> None: with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) (root / "scripts" / "apply_deploy_seed.py").unlink() with self.assertRaisesRegex(ValueError, "input is missing"): self.build(root) with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) target = root / "real-seed.py" target.write_text("seed\n") symlink = root / "scripts" / "apply_deploy_seed.py" symlink.unlink() symlink.symlink_to(target) with self.assertRaisesRegex(ValueError, "must not be a symlink"): self.build(root) def test_active_baseline_comparison_distinguishes_match_and_mismatch(self) -> None: with tempfile.TemporaryDirectory() as active_tmp, tempfile.TemporaryDirectory() as candidate_tmp: active_root = Path(active_tmp) candidate_root = Path(candidate_tmp) self.make_repo(active_root) self.make_repo(candidate_root) active = self.build(active_root) candidate = self.build(candidate_root) matched = build_active_baseline_comparison( candidate, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-candidate", active_tag_target_commit="c" * 40, active_contract=active, ) self.assertEqual(matched["status"], "match") self.assertTrue(matched["baseline_match"]) self.assertEqual(matched["enforcement_status"], "enforced") self.assertEqual( require_prod_baseline_transition(matched), matched, ) distinct_runtime = build_active_baseline_comparison( candidate, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=active, ) self.assertEqual(distinct_runtime["status"], "match") self.assertTrue(distinct_runtime["baseline_match"]) self.assertEqual( distinct_runtime["next_action"], "run-runtime-only-blue-green-cutover", ) self.make_repo(candidate_root, marker="changed") changed_candidate = self.build(candidate_root) mismatched = build_active_baseline_comparison( changed_candidate, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=active, ) self.assertEqual(mismatched["status"], "mismatch") self.assertFalse(mismatched["baseline_match"]) self.assertEqual( require_prod_baseline_transition(mismatched), mismatched, ) self.assertEqual( mismatched["next_action"], "run-separate-governed-migration-and-restage", ) self.assertNotEqual( distinct_runtime["receipt_sha256"], mismatched["receipt_sha256"], ) tampered_relation = copy.deepcopy(mismatched) tampered_relation["active"]["shared_state_digest"] = ( tampered_relation["candidate"]["shared_state_digest"] ) unsigned = { key: item for key, item in tampered_relation.items() if key != "receipt_sha256" } tampered_relation["receipt_sha256"] = hashlib.sha256( json.dumps( unsigned, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") ).hexdigest() with self.assertRaisesRegex(ValueError, "shared-state relation"): require_prod_baseline_transition(tampered_relation) def test_active_baseline_comparison_records_legacy_or_unavailable_state(self) -> None: with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) candidate = self.build(root) for status in ( "no-active-release", "missing-active-contract", "invalid-active-contract", "unavailable-active-contract", ): receipt = build_active_baseline_comparison( candidate, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag=( "" if status == "no-active-release" else "staging-release-legacy" ), active_tag_target_commit="", active_contract=None, active_contract_status=status, ) self.assertEqual(receipt["status"], status) self.assertFalse(receipt["baseline_match"]) self.assertEqual(len(receipt["receipt_sha256"]), 64) with self.assertRaisesRegex( ValueError, "no exact active baseline", ): require_prod_baseline_transition(receipt) def test_catalog_scope_requires_reset_based_staging_proof_for_l2_changes(self) -> None: with ( tempfile.TemporaryDirectory() as active_tmp, tempfile.TemporaryDirectory() as candidate_tmp, ): active_root = Path(active_tmp) candidate_root = Path(candidate_tmp) self.make_repo(active_root) self.make_repo(candidate_root) active = self.build(active_root) runtime_only_candidate = build_prod_promotion_contract( candidate_root, final_pack_hash="c" * 64, final_pack_name="salesforce-pingcap-final", final_pack_version="1.2.4", official_app_packs_commit=OFFICIAL_COMMIT, shared_state_policy=RUNTIME_ONLY_SHARED_STATE_POLICY, ) with self.assertRaisesRegex(ValueError, "metadata-first"): build_active_baseline_comparison( runtime_only_candidate, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=active, catalog_migration=True, ) candidate = build_prod_promotion_contract( candidate_root, final_pack_hash="c" * 64, final_pack_name="salesforce-pingcap-final", final_pack_version="1.2.4", official_app_packs_commit=OFFICIAL_COMMIT, shared_state_policy=METADATA_FIRST_SHARED_STATE_POLICY, ) transition = build_active_baseline_comparison( candidate, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=active, catalog_migration=True, ) self.assertEqual(transition["status"], "mismatch") self.assertEqual( transition["next_action"], "run-default-catalog-migration-and-cutover", ) self.assertEqual( require_prod_baseline_transition( transition, catalog_migration=True, candidate_contract=candidate, active_contract=active, ), transition, ) with self.assertRaisesRegex(ValueError, "next_action"): require_prod_baseline_transition(transition) reference_data = ( candidate_root / "deployments" / "offline" / "runtime" / "reference-data" / "contract-input.txt" ) reference_data.write_text("changed outside Catalog migration\n") non_catalog_candidate = build_prod_promotion_contract( candidate_root, final_pack_hash="c" * 64, final_pack_name="salesforce-pingcap-final", final_pack_version="1.2.4", official_app_packs_commit=OFFICIAL_COMMIT, shared_state_policy=METADATA_FIRST_SHARED_STATE_POLICY, ) non_catalog_transition = build_active_baseline_comparison( non_catalog_candidate, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=active, catalog_migration=True, ) self.assertEqual( non_catalog_transition["status"], "non-catalog-shared-state-mismatch", ) with self.assertRaisesRegex(ValueError, "no exact active baseline"): require_prod_baseline_transition( non_catalog_transition, catalog_migration=True, candidate_contract=non_catalog_candidate, active_contract=active, ) staging_proved_transition = build_active_baseline_comparison( non_catalog_candidate, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=active, catalog_migration=True, staging_proved_l2_migration=True, ) self.assertEqual( staging_proved_transition["status"], "staging-proved-shared-state-mismatch", ) self.assertFalse(staging_proved_transition["baseline_match"]) self.assertEqual( staging_proved_transition["next_action"], "run-default-catalog-migration-and-cutover", ) with self.assertRaisesRegex( ValueError, "requires Catalog migration mode", ): require_prod_baseline_transition(staging_proved_transition) with self.assertRaisesRegex( ValueError, "reset-based Formal Staging proof", ): require_prod_baseline_transition( staging_proved_transition, catalog_migration=True, candidate_contract=non_catalog_candidate, active_contract=active, ) self.assertEqual( require_prod_baseline_transition( staging_proved_transition, catalog_migration=True, candidate_contract=non_catalog_candidate, active_contract=active, staging_proved_l2_migration=True, ), staging_proved_transition, ) forged_catalog_transition = copy.deepcopy(non_catalog_transition) forged_catalog_transition["status"] = "mismatch" forged_catalog_transition["next_action"] = ( "run-default-catalog-migration-and-cutover" ) unsigned = { key: item for key, item in forged_catalog_transition.items() if key != "receipt_sha256" } forged_catalog_transition["receipt_sha256"] = hashlib.sha256( json.dumps( unsigned, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") ).hexdigest() with self.assertRaisesRegex(ValueError, "outside the default Catalog"): require_prod_baseline_transition( forged_catalog_transition, catalog_migration=True, candidate_contract=non_catalog_candidate, active_contract=active, ) def test_migrated_baseline_receipt_is_bound_to_exact_mismatch(self) -> None: with ( tempfile.TemporaryDirectory() as active_tmp, tempfile.TemporaryDirectory() as candidate_tmp, ): active_root = Path(active_tmp) candidate_root = Path(candidate_tmp) self.make_repo(active_root, marker="active") self.make_repo(candidate_root, marker="candidate") transition = build_active_baseline_comparison( self.build(candidate_root), candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=self.build(active_root), ) self.assertEqual(transition["status"], "mismatch") migrated = build_migrated_baseline_receipt( migration_id="20260729-companion-flow-payload", active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, candidate_contract_sha256=transition["candidate"][ "contract_sha256" ], candidate_shared_state_digest=transition["candidate"][ "shared_state_digest" ], candidate_final_pack_hash="e" * 64, installed_packs=[ { "name": "pingcap-hubspot-companion", "source_revision": "f" * 40, "desired_payload_hash": ( "nexus-pack-sha256-v1:" + "a" * 64 ), "applied_payload_hash": ( "nexus-pack-sha256-v1:" + "a" * 64 ), }, { "name": "salesforce-crm-standard", "source_revision": "f" * 40, "desired_payload_hash": ( "nexus-pack-sha256-v1:" + "b" * 64 ), "applied_payload_hash": ( "nexus-pack-sha256-v1:" + "b" * 64 ), }, ], migration_command_id="11111111-1111-1111-1111-111111111111", verification_command_id="22222222-2222-2222-2222-222222222222", preflight_plan_sha256="1" * 64, rehearsal_evidence_sha256="2" * 64, source_reader_stop_command_id=( "33333333-3333-3333-3333-333333333333" ), zero_nonterminal_workitems_command_id=( "44444444-4444-4444-4444-444444444444" ), ) self.assertEqual( require_bound_migrated_baseline_receipt( migrated, baseline_transition=transition, expected_final_pack_hash="e" * 64, ), migrated, ) candidate_contract = self.build(candidate_root) matched_transition = build_active_baseline_comparison( candidate_contract, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=candidate_contract, ) with self.assertRaisesRegex(ValueError, "allowed baseline transition"): require_bound_migrated_baseline_receipt( migrated, baseline_transition=matched_transition, expected_final_pack_hash="e" * 64, ) self.assertEqual( require_bound_migrated_baseline_receipt( migrated, baseline_transition=matched_transition, expected_final_pack_hash="e" * 64, allow_baseline_match=True, ), migrated, ) self.assertEqual( migrated["migration_id"], "20260729-companion-flow-payload", ) self.assertEqual( migrated["schema_version"], "ordinary-prod-migrated-baseline/v4", ) self.assertEqual( migrated["evidence"], { "migration_command_id": ( "11111111-1111-1111-1111-111111111111" ), "verification_command_id": ( "22222222-2222-2222-2222-222222222222" ), "preflight_plan_sha256": "1" * 64, "rehearsal_evidence_sha256": "2" * 64, "source_reader_stop_command_id": ( "33333333-3333-3333-3333-333333333333" ), "zero_nonterminal_workitems_command_id": ( "44444444-4444-4444-4444-444444444444" ), }, ) self.assertEqual( [ pack["name"] for pack in migrated["database"]["installed_packs"] ], ["pingcap-hubspot-companion", "salesforce-crm-standard"], ) tampered = copy.deepcopy(migrated) tampered["database"]["installed_packs"][0]["name"] = "another-pack" with self.assertRaisesRegex(ValueError, "receipt_sha256 mismatch"): require_migrated_baseline_receipt(tampered) with self.assertRaisesRegex(ValueError, "exact candidate"): require_bound_migrated_baseline_receipt( migrated, baseline_transition=transition, expected_final_pack_hash="9" * 64, ) def test_active_contract_can_be_derived_from_exact_git_tree(self) -> None: with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) subprocess.run(["git", "init", "-q", str(root)], check=True) subprocess.run( ["git", "-C", str(root), "config", "user.name", "Test"], check=True, ) subprocess.run( ["git", "-C", str(root), "config", "user.email", "test@example.invalid"], check=True, ) subprocess.run(["git", "-C", str(root), "add", "."], check=True) subprocess.run( ["git", "-C", str(root), "commit", "-qm", "baseline"], check=True, ) revision = subprocess.run( ["git", "-C", str(root), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ).stdout.strip() derived = build_prod_promotion_contract_from_git( root, revision=revision, final_pack_name="salesforce-pingcap-final", final_pack_version="1.2.3", official_app_packs_commit=OFFICIAL_COMMIT, ) exact_pack_hash = final_pack_payload_hash_from_git(root, revision) filesystem = build_prod_promotion_contract( root, final_pack_hash=exact_pack_hash, final_pack_name="salesforce-pingcap-final", final_pack_version="1.2.3", official_app_packs_commit=OFFICIAL_COMMIT, ) self.assertEqual(derived, filesystem) with self.assertRaisesRegex( ValueError, "declared final pack hash does not match the exact Git tree", ): build_prod_promotion_contract_from_git( root, revision=revision, final_pack_hash=FINAL_PACK_HASH, official_app_packs_commit=OFFICIAL_COMMIT, ) with self.assertRaisesRegex( ValueError, "declared official app-packs commit does not match the exact Git tree", ): build_prod_promotion_contract_from_git( root, revision=revision, official_app_packs_commit="c" * 40, ) receipt = build_active_baseline_comparison( filesystem, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-legacy", active_tag_target_commit=revision, active_contract=derived, active_contract_source="derived-exact-active-tag", ) self.assertEqual( receipt["active"]["contract_source"], "derived-exact-active-tag", ) self.assertEqual( require_prod_baseline_transition(receipt), receipt, ) def test_baseline_admission_rejects_tampered_receipt(self) -> None: with tempfile.TemporaryDirectory() as tempdir: root = Path(tempdir) self.make_repo(root) contract = self.build(root) receipt = build_active_baseline_comparison( contract, candidate_release_tag="staging-release-candidate", candidate_tag_target_commit="c" * 40, active_release_tag="staging-release-active", active_tag_target_commit="d" * 40, active_contract=contract, ) receipt["active"]["shared_state_digest"] = "f" * 64 with self.assertRaisesRegex(ValueError, "receipt_sha256 mismatch"): require_prod_baseline_transition(receipt) if __name__ == "__main__": unittest.main()