from __future__ import annotations import json import tempfile import unittest from contextlib import ExitStack from pathlib import Path from types import SimpleNamespace from scripts import serviceification_d3_test_patch as replay from scripts import run_serviceification_d3_prod as prod from scripts import run_serviceification_d3_staging as staging from scripts import run_serviceification_d3_test as test_deploy REPO_ROOT = Path(__file__).resolve().parents[1] PATCH_DIR = "docs/migration/patch-bundles/20260819_task_status_open_and_lead_relate_flow" def bind_fixture(stack: ExitStack, owner, name: str, value) -> None: """Install explicit no-I/O fixture dependencies and restore module state.""" original = getattr(owner, name) stack.callback(setattr, owner, name, original) setattr(owner, name, value) def forbidden_io(*args, **kwargs): raise AssertionError("post-Catalog admission must not perform this operation") class CompletedPatchResumeTests(unittest.TestCase): def setUp(self) -> None: temp = tempfile.TemporaryDirectory() self.addCleanup(temp.cleanup) root = Path(temp.name) self.context = SimpleNamespace( env_root=root, run_root=root / "runs" / "exact-run", run_id="exact-run", state_file=root / "active.json", envv={}, bom={}, ) self.context.state_file.write_text(json.dumps({ "runId": "exact-run", "phase": "serving-start-in-progress", })) self.sealed = replay.preflight_and_seal_patch( self.context, PATCH_DIR, source_root=REPO_ROOT, environment="prod", ) stack = ExitStack() self.addCleanup(stack.close) self.sql_files = [] bind_fixture(stack, replay, "execute_patch_sql_file", lambda env, path: self.sql_files.append(path.name)) self.receipt = replay.apply_sealed_patch_bundle( self.context, self.sealed, requested_patch_dir=PATCH_DIR, ) self.assertEqual(self.receipt["sqlFiles"], self.sql_files) bind_fixture(stack, replay, "execute_patch_sql_file", forbidden_io) self.summary = self.context.run_root / "patch-replay" / "patch-summary.json" def resume(self, requested: str = PATCH_DIR): return replay.require_active_run_can_replay_patch( self.context, requested, read_state=lambda path: json.loads(path.read_text()), ) def test_exact_post_catalog_retry_uses_completed_receipt_without_sql(self) -> None: before = self.summary.read_bytes() for phase in sorted(replay.PATCH_SQL_BLOCKED_PHASES): with self.subTest(phase=phase): self.context.state_file.write_text(json.dumps({"runId": "exact-run", "phase": phase})) self.assertEqual(self.receipt, self.resume()) self.assertIsNone(replay.replay_patch_before_catalog( self.context, requested_patch_dir=PATCH_DIR, read_state=lambda path: json.loads(path.read_text()), )) self.assertEqual(before, self.summary.read_bytes()) def test_pre_catalog_retry_still_requires_normal_sql_replay(self) -> None: self.context.state_file.write_text(json.dumps({"runId": "exact-run", "phase": "quiesced"})) self.assertIsNone(self.resume()) def test_trailing_slash_receipt_resumes_without_rewriting_identity_or_sql(self) -> None: for requested in (PATCH_DIR + "/", " " + PATCH_DIR + "/ "): with self.subTest(requested=requested): self.context.state_file.write_text(json.dumps({ "runId": "exact-run", "phase": "quiesced", })) replay.preflight_and_seal_patch( self.context, requested, source_root=REPO_ROOT, environment="prod", ) executed = [] with ExitStack() as stack: bind_fixture(stack, replay, "execute_patch_sql_file", lambda env, path: executed.append(path.name)) self.receipt = replay.replay_patch_before_catalog( self.context, requested_patch_dir=requested, read_state=lambda path: json.loads(path.read_text()), ) self.assertEqual(requested.strip(), self.receipt["requestedPatchDir"]) self.assertEqual(self.receipt["sqlFiles"], executed) before = self.summary.read_bytes() for phase in sorted(replay.PATCH_SQL_BLOCKED_PHASES): self.context.state_file.write_text(json.dumps({ "runId": "exact-run", "phase": phase, })) self.assertEqual(self.receipt, self.resume(requested)) self.assertIsNone(replay.replay_patch_before_catalog( self.context, requested_patch_dir=requested, read_state=lambda path: json.loads(path.read_text()), )) # Accepted spelling is persisted identity, not permission to # substitute another spelling on an existing run's retry. with self.assertRaises(replay.DeployError): self.resume(PATCH_DIR) self.context.state_file.write_text(json.dumps({ "runId": "exact-run", "phase": "serving-start-in-progress", })) self.assert_adapter_closeout(requested) self.assertEqual(before, self.summary.read_bytes()) def test_new_run_does_not_reuse_previous_run_receipt(self) -> None: self.context.state_file.write_text(json.dumps({"runId": "previous-run", "phase": "serving"})) self.assertIsNone(self.resume()) def test_post_catalog_retry_without_receipt_fails_closed(self) -> None: self.summary.unlink() with self.assertRaises(replay.DeployError): self.resume() def test_post_catalog_retry_refuses_symlinked_receipt(self) -> None: original = self.summary.with_name("other-run.json") self.summary.rename(original) self.summary.symlink_to(original) with self.assertRaises(replay.DeployError): self.resume() def test_post_catalog_retry_rejects_incomplete_or_foreign_receipt(self) -> None: for field, value in ( ("schema", "unknown"), ("status", "failed"), ("requestedPatchDir", PATCH_DIR + "-other"), ("patchDir", "/other-run/patch-bundle"), ("manifestPath", "/other-run/patch-bundle/patch-bundle.json"), ("treeDigest", "sha256:" + "0" * 64), ("sqlFiles", self.receipt["sqlFiles"][:-1]), ("results", self.receipt["results"][:-1]), ("results", [{"file": x, "status": "failed"} for x in self.receipt["sqlFiles"]]), ): with self.subTest(field=field, value=value): self.summary.write_text(json.dumps({**self.receipt, field: value})) with self.assertRaises(replay.DeployError): self.resume() def test_post_catalog_retry_rejects_changed_sealed_tree(self) -> None: (self.sealed / self.receipt["sqlFiles"][0]).write_text("SELECT 1;\n") with self.assertRaises(replay.DeployError): self.resume() def test_all_deploy_adapters_retain_the_verified_receipt_in_closeout(self) -> None: self.assert_adapter_closeout(PATCH_DIR) def assert_adapter_closeout(self, requested: str) -> None: for adapter in (prod, staging, test_deploy): with self.subTest(adapter=adapter.__name__), ExitStack() as stack: bind_fixture(stack, adapter, "REQUESTED_PATCH_DIR", requested) bind_fixture(stack, adapter, "LAST_PATCH_REPLAY", None) bind_fixture(stack, adapter, "_ORIG_CONVERGE", lambda *a, **kw: {"verified": True}) bind_fixture(stack, adapter.impl, "read_state", lambda path: json.loads(path.read_text())) if adapter is prod: bind_fixture(stack, prod, "EXPECTED_PATCH_TREE_DIGEST", self.receipt["treeDigest"]) for name, value in ( ("require_prod_in_progress", {"targetSlot": "slot-3105"}), ("require_target_is_standby", "slot-3105"), ("preflight_companion_authority", None), ("archive_target_generation", None), ("prepare_target_for_convergence", None), ("require_live_target", "target"), ("verify_public_target", {"passed": True}), ("retire_previous_runtime", {"verified": True}), ("publish_current", None), ): bind_fixture(stack, prod, name, lambda *a, result=value, **kw: result) bind_fixture(stack, prod.impl, "replace_state", lambda *a, **kw: None) result = prod.converge_prod(self.context, l2_root=REPO_ROOT) elif adapter is staging: for name, value in ( ("live_nginx_slot", "green"), ("require_target_is_standby", "blue"), ("preflight_companion_authority", None), ("exact_l2_patch_source", REPO_ROOT), ("archive_standby_d3_generation", None), ("require_live_nginx_is_target", None), ("retire_previous_live_after_cutover", None), ): bind_fixture(stack, staging, name, lambda *a, result=value, **kw: result) bind_fixture(stack, staging.impl, "require_state", lambda *a, **kw: {"phase": "serving-start-in-progress"}) result = staging._converge_deployment(self.context, l2_root=REPO_ROOT) else: bind_fixture(stack, test_deploy, "require_no_uncertain_database_production_refresh", lambda *a, **kw: None) bind_fixture(stack, test_deploy, "test_slot", lambda *a: "test1") bind_fixture(stack, test_deploy, "read_production_refresh_active_state", lambda path: json.loads(path.read_text())) bind_fixture(stack, test_deploy.impl, "installation_org_id", lambda: "fixture-org") result = test_deploy._converge_deployment(self.context) self.assertTrue(result["verified"]) self.assertEqual(self.receipt, adapter.LAST_PATCH_REPLAY) self.assertEqual(self.receipt, result["patchReplay"]) def test_prod_retry_still_rejects_a_different_tag_patch_digest_before_convergence(self) -> None: with ExitStack() as stack: for name, value in ( ("REQUESTED_PATCH_DIR", PATCH_DIR), ("LAST_PATCH_REPLAY", None), ("EXPECTED_PATCH_TREE_DIGEST", "sha256:" + "0" * 64), ): bind_fixture(stack, prod, name, value) bind_fixture(stack, prod, "require_prod_in_progress", lambda *a: {"targetSlot": "slot-3105"}) bind_fixture(stack, prod, "require_target_is_standby", lambda *a, **kw: "slot-3105") bind_fixture(stack, prod, "preflight_companion_authority", lambda *a, **kw: None) bind_fixture(stack, prod.impl, "read_state", lambda path: json.loads(path.read_text())) bind_fixture(stack, prod, "archive_target_generation", forbidden_io) bind_fixture(stack, prod, "_ORIG_CONVERGE", forbidden_io) with self.assertRaisesRegex(replay.DeployError, "promoted digest"): prod.converge_prod(self.context, l2_root=REPO_ROOT) self.assertIsNone(prod.LAST_PATCH_REPLAY) if __name__ == "__main__": unittest.main()