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 run_serviceification_d3_prod as prod from scripts import serviceification_d3_test_patch as replay 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: original = getattr(owner, name) stack.callback(setattr, owner, name, original) setattr(owner, name, value) class ProdPatchOrderTests(unittest.TestCase): """Exercise the installed Prod adapter and real stopped-upgrade ordering. Host/DB boundaries are explicit no-I/O fixtures. Patch sealing, replay, receipt persistence, and the engine's callback/checkpoint sequence are real; the Flow fixture models only the customized ownership gate, not the Go reconciler or the SQL interpreter. """ 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={}, ) executable = root / "setup-fixture" executable.write_text("not executed\n") executable.chmod(0o700) self.state = { "runId": "exact-run", "phase": "quiesced", "baselineReset": {}, "exactSources": { "coreRoot": str(root), "appPacksRoot": str(root), "l2Root": str(root), "coreCommit": "1" * 40, "appPacksCommit": "2" * 40, "l2Commit": "3" * 40, }, "prepared": { "setupBinary": str(executable), "packRuntimeMetadataBinary": str(executable), "setupCoreRevision": "1" * 40, }, } self.persist_state() self.events = [] self.flow = {"IsCustomized": True, "activeVersion": 1} self.stack = ExitStack() self.addCleanup(self.stack.close) # _install mutates the shared engine and its host adapters. Restore all # existing module bindings so combined Staging/Test suites stay isolated. for owner in (prod.impl, prod.impl.staging, prod.impl.test_deploy): self.stack.callback(owner.__dict__.update, owner.__dict__.copy()) prod._install() self.bind(prod, "REQUESTED_PATCH_DIR", PATCH_DIR) self.bind(prod, "LAST_PATCH_REPLAY", None) self.bind(prod, "quiesce_target_runtime", self.quiesce) self.bind(prod.impl.staging, "quiesce_staging_target_runtime", self.quiesce) self.bind(prod.impl, "require_state", self.require_state) self.bind(prod.impl, "replace_state", self.replace_state) self.bind(prod.impl, "read_state", lambda path: json.loads(path.read_text())) self.bind(prod.impl, "require_catalog_start", lambda state: {"mode": "INSTALL"}) self.bind(prod.impl, "owned_containers", lambda: []) self.bind(prod.impl, "read_optional_catalog_authority", lambda env: None) self.bind(prod.impl, "catalog_bundle_operation_id", lambda context: "fixture-operation") self.bind(prod.impl, "stopped_upgrade_runtime_env", lambda env, **kwargs: dict(env)) self.bind(prod.impl, "run_backend_setup", lambda *a: self.events.append("l0-schema")) self.bind(prod.impl, "read_optional_worker_ownership", lambda env: (True, None)) self.bind(prod.impl, "require_worker_ownership_before", lambda state: None) self.bind(prod, "_ORIG_APPLY_CATALOG", self.publish_catalog) self.bind(replay, "execute_patch_sql_file", self.execute_sql) replay.preflight_and_seal_patch( self.context, PATCH_DIR, source_root=REPO_ROOT, environment="prod", ) def bind(self, owner, name, value) -> None: bind_fixture(self.stack, owner, name, value) def persist_state(self) -> None: self.context.state_file.write_text(json.dumps(self.state)) def require_state(self, context, *, expected_phase): allowed = {expected_phase} if isinstance(expected_phase, str) else expected_phase self.assertIn(self.state["phase"], allowed) return dict(self.state) def replace_state(self, context, *, expected_phase, updates): self.require_state(context, expected_phase=expected_phase) self.state.update(updates) self.persist_state() return dict(self.state) def quiesce(self, env, run_dir, *, disable_autostart) -> None: self.assertTrue(disable_autostart) self.events.append("quiesce") def execute_sql(self, env, path) -> None: self.assertIn("quiesce", self.events) self.events.append(path.name) if path.name.startswith("002_"): self.flow["IsCustomized"] = False def apply_flow(self, env, l2_root, core_root, app_packs_root, **kwargs) -> None: self.events.append("pack-flow-apply") if not kwargs["candidate_already_applied"]: if not self.flow["IsCustomized"]: self.flow["activeVersion"] = 2 kwargs["on_candidate_applied"]() def apply_runtime(self, *, use_default=False): kwargs = {"apply_app_shell": lambda *a: {"status": "passed"}} if not use_default: kwargs["apply_runtime"] = self.apply_flow return prod.impl.apply_stopped_upgrade_runtime(self.context, **kwargs) def publish_catalog(self, context, **kwargs): self.events.append("catalog") return {"stage": "catalog-applied"} def finish_catalog(self): self.state["phase"] = "catalog-ready" self.persist_state() return prod.impl.apply_catalog(self.context) def assert_order(self) -> None: self.assertEqual(2, self.flow["activeVersion"]) self.assertFalse(self.flow["IsCustomized"]) receipt = prod.LAST_PATCH_REPLAY self.assertEqual("passed", receipt["status"]) self.assertEqual(PATCH_DIR, receipt["requestedPatchDir"]) for sql_file in receipt["sqlFiles"]: self.assertEqual(1, self.events.count(sql_file)) self.assertLess(self.events.index("l0-schema"), self.events.index(sql_file)) self.assertLess(self.events.index(sql_file), self.events.index("pack-flow-apply")) self.assertLess(self.events.index("quiesce"), self.events.index("l0-schema")) self.assertLess(self.events.index("pack-flow-apply"), self.events.index("catalog")) def test_fresh_prod_replays_patch_before_flow_apply_not_only_before_catalog(self) -> None: self.apply_runtime() self.finish_catalog() self.assert_order() def test_default_runtime_callback_also_replays_patch_first(self) -> None: self.assertIs(prod.impl.apply_stopped_upgrade_runtime, prod.apply_stopped_upgrade_runtime) self.bind(prod.impl.staging, "run_catalog_offline_candidate_phase", self.apply_flow) self.apply_runtime(use_default=True) self.finish_catalog() self.assert_order() def test_runtime_apply_retry_replays_patch_before_flow(self) -> None: self.state["phase"] = "runtime-apply-in-progress" self.persist_state() self.apply_runtime() self.finish_catalog() self.assert_order() def test_same_process_candidate_resume_and_catalog_do_not_replay_successful_sql(self) -> None: self.apply_runtime() receipt = prod.LAST_PATCH_REPLAY self.state["phase"] = "runtime-candidate-applied" self.persist_state() self.apply_runtime() self.finish_catalog() self.assertIs(receipt, prod.LAST_PATCH_REPLAY) self.assert_order() def test_no_patch_keeps_the_normal_runtime_path(self) -> None: self.bind(prod, "REQUESTED_PATCH_DIR", "") self.apply_runtime() self.finish_catalog() self.assertIsNone(prod.LAST_PATCH_REPLAY) self.assertEqual(["quiesce", "l0-schema", "pack-flow-apply", "quiesce", "catalog"], self.events) def test_patch_failure_stops_before_pack_flow_apply(self) -> None: def failed_sql(*args): raise prod.DeployError("fixture SQL failure") self.bind(replay, "execute_patch_sql_file", failed_sql) with self.assertRaisesRegex(prod.DeployError, "fixture SQL failure"): self.apply_runtime() self.assertNotIn("pack-flow-apply", self.events) self.assertIsNone(prod.LAST_PATCH_REPLAY) def test_missing_required_replay_stops_before_pack_flow_apply(self) -> None: self.bind(replay, "replay_patch_before_catalog", lambda *a, **kw: None) with self.assertRaisesRegex(prod.DeployError, "patch was not replayed"): self.apply_runtime() self.assertNotIn("pack-flow-apply", self.events) def test_quiescence_failure_prevents_patch_and_runtime_apply(self) -> None: def failed_quiesce(*args, **kwargs): raise prod.DeployError("fixture quiescence failure") self.bind(prod.impl.staging, "quiesce_staging_target_runtime", failed_quiesce) with self.assertRaisesRegex(prod.DeployError, "fixture quiescence failure"): self.apply_runtime() self.assertEqual([], self.events) self.assertIsNone(prod.LAST_PATCH_REPLAY) if __name__ == "__main__": unittest.main()