from __future__ import annotations import json import os import re import subprocess import tempfile import textwrap import unittest from contextlib import ExitStack from pathlib import Path from types import SimpleNamespace from deployments.online.runtime.prod_d3_authority import IN_PROGRESS_SCHEMA from deployments.online.runtime import prepare_multi_prod_d3_runtime_env as prod_prepare from scripts import run_serviceification_d3_prod as prod def bind_fixture(stack, owner, name, value): original = getattr(owner, name) stack.callback(setattr, owner, name, original) setattr(owner, name, value) class ProdRetirementTests(unittest.TestCase): """Real Prod closeout/authority files; explicit no-I/O Docker/systemd fixtures.""" def setUp(self): temp = tempfile.TemporaryDirectory() self.addCleanup(temp.cleanup) self.root = Path(temp.name) / "d3" release_root = self.root / "releases" / "staging-d3-release-123" state_file = release_root / ".runtime" / "serviceification-d3" / "active.json" state_file.parent.mkdir(parents=True) self.pending = { "schema": IN_PROGRESS_SCHEMA, "releaseTag": "staging-d3-release-123", "tagTargetCommit": "a" * 40, "resolvedBomReference": "ghcr.io/pingcap-inc/nexuscrm-serviceification-bom@sha256:" + "b" * 64, "previousBackendPort": 3104, "previousRouteKind": "d3", "targetSlot": "slot-3105", "targetBackendPort": 3105, "releaseRoot": str(release_root), } self.context = SimpleNamespace( env_root=release_root, run_id="exact-run", state_file=state_file, bom={"desired_bom_sha256": "sha256:" + "c" * 64}, envv={ "ENV_NAME": "multi-prod-d3-slot-3105", "RELEASE_TAG": self.pending["releaseTag"], "TAG_TARGET_COMMIT": self.pending["tagTargetCommit"], "D3_RESOLVED_BOM_REFERENCE": self.pending["resolvedBomReference"], "PROD_D3_PREVIOUS_BACKEND_PORT": "3104", "PROD_D3_PREVIOUS_ROUTE_KIND": "d3", }, ) self.state = {"phase": "serving", "runId": "exact-run", "publishedNginxConfigDigest": "fixture"} self.persist() self.live = prod.ProdRoute(3104, "slot-3104", True) self.target = prod.ProdRoute(3105, "slot-3105", True) self.events, self.commands, self.verified_ports = [], [], [] self.public_calls = 0 self.fail_public_call = None self.fail_stop = False self.ineffective_commands = set() self.occupied_ports = set() self.inspect_overrides = {} self.containers = { digit * 64: { "id": digit * 64, "name": f"/nexuscrm-d3-prod-slot-{port}-old-runtime-api", "owner": f"prod-slot-{port}", "running": True, "restarting": False, "restartPolicy": "unless-stopped", } for digit, port in (("1", 3104), ("2", 3105), ("3", 3106)) } self.units = { unit: {"load_state": "loaded", "active_state": "active", "unit_file_state": "enabled"} for port in (3104, 3105, 3106) for unit in ( f"nexuscrm-multi-prod-d3-slot-{port}-hubspot-options-adapter.service", f"nexuscrm-multi-prod-slot-{port}-backend.service", ) } self.stack = ExitStack() self.addCleanup(self.stack.close) self.bind(prod, "REQUESTED_PATCH_DIR", "") self.bind(prod, "LAST_PATCH_REPLAY", None) self.bind(prod, "live_route", lambda: self.live) self.bind(prod, "preflight_companion_authority", lambda *a, **kw: None) self.bind(prod, "archive_target_generation", lambda context: self.events.append("archive")) self.bind(prod, "prepare_target_for_convergence", lambda *a, **kw: self.events.append("prepare")) self.bind(prod, "_ORIG_CONVERGE", self.converge_target) self.bind(prod, "verify_public_target", self.public_readback) 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, "default_capture", self.capture) self.bind(prod.impl, "default_execute", self.execute) self.bind(prod.impl.staging, "systemd_unit_state", self.unit_state) self.bind(prod.impl.staging, "require_staging_target_port_listener_free", self.port_free) def bind(self, owner, name, value): bind_fixture(self.stack, owner, name, value) def persist(self): (self.root / "in-progress.json").write_text(json.dumps(self.pending)) self.context.state_file.write_text(json.dumps(self.state)) def require_state(self, context, *, expected_phase, **kwargs): self.assertEqual(expected_phase, self.state["phase"]) return dict(self.state) def replace_state(self, context, *, expected_phase, updates): self.require_state(context, expected_phase=expected_phase) self.events.append("checkpoint") self.state.update(updates) context.state_file.write_text(json.dumps(self.state)) return dict(self.state) def converge_target(self, *args, **kwargs): self.events.append("converge") self.live = self.target return {"stage": "serving"} def public_readback(self, route): self.assertEqual(self.target, route) self.public_calls += 1 self.events.append("public-readback") if self.fail_public_call == self.public_calls: raise prod.DeployError("fixture public readback failure") return {"backendPort": route.backend_port} def capture(self, args): if args[:3] == ("docker", "ps", "-a"): owner = args[args.index("--filter") + 1].split("=", 2)[2] self.assertEqual("prod-slot-3104", owner) return "\n".join(key for key, item in self.containers.items() if item["owner"] == owner) self.assertEqual(("docker", "inspect", "--format"), args[:3]) self.assertNotIn(".Config.Env", args[3]) return json.dumps({**self.containers[args[-1]], **self.inspect_overrides}) def unit_state(self, unit): return dict(self.units.get(unit, { "load_state": "not-found", "active_state": "inactive", "unit_file_state": "not-found", })) def execute(self, args): self.assertGreater(self.public_calls, 0, "stop must follow public target proof") self.commands.append(tuple(args)) self.events.append("retire") if args[:2] in self.ineffective_commands or args[:4] in self.ineffective_commands: return if args[:2] == ("docker", "update"): self.assertEqual("1" * 64, args[-1]) self.containers[args[-1]]["restartPolicy"] = "no" elif args[:2] == ("docker", "stop"): self.assertEqual("1" * 64, args[-1]) if self.fail_stop: raise subprocess.CalledProcessError(1, args) self.containers[args[-1]]["running"] = False self.containers[args[-1]]["restarting"] = False else: self.assertEqual(("sudo", "-n", "systemctl"), args[:3]) self.assertNotIn("3105", args[-1]) self.assertNotIn("3106", args[-1]) if args[3] == "disable": self.units[args[-1]]["unit_file_state"] = "disabled" else: self.assertEqual("stop", args[3]) if self.fail_stop: raise subprocess.CalledProcessError(1, args) self.units[args[-1]]["active_state"] = "inactive" def port_free(self, port, label): self.verified_ports.append(int(port)) if int(port) in self.occupied_ports: raise prod.DeployError("fixture unknown port listener remains") def assert_pending(self): self.assertTrue((self.root / "in-progress.json").is_file()) self.assertFalse((self.root / "current.json").exists()) self.assertFalse((self.root / "current").exists()) def test_d3_cutover_stops_only_predecessor_then_publishes_real_marker(self): result = prod.converge_prod(self.context) self.assertFalse(self.containers["1" * 64]["running"]) self.assertEqual("no", self.containers["1" * 64]["restartPolicy"]) for identifier in ("2" * 64, "3" * 64): self.assertTrue(self.containers[identifier]["running"]) self.assertEqual("unless-stopped", self.containers[identifier]["restartPolicy"]) self.assertLess(self.events.index("converge"), self.events.index("public-readback")) self.assertLess(self.events.index("public-readback"), self.events.index("retire")) self.assertEqual(["public-readback", "checkpoint"], self.events[-2:]) self.assertTrue(result["previousRuntimeRetirement"]["verified"]) self.assertEqual(3104, result["previousRuntimeRetirement"]["backendPort"]) self.assertIn(3801, self.verified_ports) self.assertIn(3810, self.verified_ports) self.assertNotIn(3105, self.verified_ports) self.assertFalse((self.root / "in-progress.json").exists()) self.assertEqual(self.context.env_root, (self.root / "current").resolve()) self.assertEqual(3105, json.loads((self.root / "current.json").read_text())["backendPort"]) self.assertFalse(any("rm" in command for command in self.commands)) def test_classic_handoff_stops_reviewed_units_without_docker_mutation(self): self.pending["previousRouteKind"] = "legacy" self.context.envv["PROD_D3_PREVIOUS_ROUTE_KIND"] = "legacy" self.live = prod.ProdRoute(3104, "slot-3104", False) for unit in prod.previous_runtime_units(self.live): self.units[unit] = {"load_state": "loaded", "active_state": "active", "unit_file_state": "enabled"} self.persist() result = prod.converge_prod(self.context) self.assertEqual("legacy", result["previousRuntimeRetirement"]["kind"]) self.assertEqual([3104, 3205], self.verified_ports) self.assertFalse(any(command[0] == "docker" for command in self.commands)) for unit in result["previousRuntimeRetirement"]["retiredUnits"]: self.assertEqual("inactive", self.units[unit]["active_state"]) self.assertEqual("disabled", self.units[unit]["unit_file_state"]) def test_target_failure_does_not_stop_predecessor_or_publish(self): def failed(*args, **kwargs): raise prod.DeployError("fixture target failure") self.bind(prod, "_ORIG_CONVERGE", failed) with self.assertRaises(prod.DeployError): prod.converge_prod(self.context) self.assertEqual([], self.commands) self.assert_pending() def test_public_readback_failure_does_not_stop_predecessor(self): self.fail_public_call = 1 with self.assertRaisesRegex(prod.DeployError, "public readback"): prod.converge_prod(self.context) self.assertEqual([], self.commands) self.assert_pending() def test_partial_stop_failure_preserves_pending_and_exact_retry_finishes(self): self.fail_stop = True with self.assertRaisesRegex(prod.DeployError, "stop or verify"): prod.converge_prod(self.context) self.assert_pending() self.fail_stop = False result = prod.converge_prod(self.context) self.assertEqual(3104, result["previousRuntimeRetirement"]["backendPort"]) self.assertFalse((self.root / "in-progress.json").exists()) self.assertTrue(self.containers["2" * 64]["running"]) def test_unknown_listener_blocks_closeout_without_killing_it(self): self.occupied_ports.add(3801) with self.assertRaisesRegex(prod.DeployError, "listener"): prod.converge_prod(self.context) self.assert_pending() self.assertFalse(any("kill" in command for command in self.commands)) def test_successful_command_without_actual_retirement_blocks_publication(self): for command, error in ( (("docker", "stop"), "container remains active"), (("docker", "update"), "container remains active or restartable"), (("sudo", "-n", "systemctl", "stop"), "unit remains active"), (("sudo", "-n", "systemctl", "disable"), "unit remains enabled"), ): with self.subTest(command=command): self.containers["1" * 64].update(running=True, restartPolicy="unless-stopped") for unit in prod.previous_runtime_units(prod.ProdRoute(3104, "slot-3104", True)): self.units[unit].update(active_state="active", unit_file_state="enabled") self.ineffective_commands = {command} with self.assertRaisesRegex(prod.DeployError, error): prod.converge_prod(self.context) self.assert_pending() def test_unknown_unit_restart_state_is_rejected_before_any_stop(self): self.units["nexuscrm-multi-prod-slot-3104-backend.service"]["unit_file_state"] = "unknown" with self.assertRaisesRegex(prod.DeployError, "autostart state"): prod.converge_prod(self.context) self.assertEqual([], self.commands) self.assert_pending() def test_target_failure_after_retirement_preserves_pending_and_retries_idempotently(self): self.fail_public_call = 2 with self.assertRaisesRegex(prod.DeployError, "public readback"): prod.converge_prod(self.context) self.assert_pending() before = list(self.commands) self.fail_public_call = None prod.converge_prod(self.context) self.assertEqual(before, self.commands) def test_foreign_container_identity_is_rejected_before_any_stop(self): self.inspect_overrides["name"] = "/nexuscrm-d3-prod-slot-3105-new-runtime-api" with self.assertRaisesRegex(prod.DeployError, "not slot-owned"): prod.converge_prod(self.context) self.assertEqual([], self.commands) self.assert_pending() def test_same_target_predecessor_is_rejected_before_mutation(self): self.pending["previousBackendPort"] = 3105 self.context.envv["PROD_D3_PREVIOUS_BACKEND_PORT"] = "3105" self.persist() with self.assertRaisesRegex(prod.DeployError, "invalid slot binding"): prod.converge_prod(self.context) self.assertEqual([], self.commands) self.assert_pending() def test_changed_pending_authority_is_rejected_after_cutover(self): original = self.public_readback def changed(route): result = original(route) self.pending["previousBackendPort"] = 3103 self.persist() return result self.bind(prod, "verify_public_target", changed) with self.assertRaisesRegex(prod.DeployError, "durable target authority"): prod.converge_prod(self.context) self.assertEqual([], self.commands) self.assert_pending() def test_publication_requires_a_matching_verified_retirement_receipt(self): for receipt in (None, {"verified": True}, { "schema": "nexuscrm.prod-predecessor-retirement/v1", "verified": True, "kind": "d3", "backendPort": 3103, "targetBackendPort": 3105, }): with self.subTest(receipt=receipt): self.state["previousRuntimeRetirement"] = receipt with self.assertRaisesRegex(prod.DeployError, "verified predecessor retirement"): prod.publish_current(self.context, self.target) self.assert_pending() def workflow_receipt_check(self, prepared): workflow = (Path(__file__).resolve().parents[1] / ".github/workflows/deploy-multi-prod-d3-environment.yml").read_text() step = workflow.split("- name: Verify cutover readback and current marker", 1)[1].split("\n - name:", 1)[0] script = "set -euo pipefail\n" + textwrap.dedent(step[step.index(" jq -e '.previousRuntimeRetirement'"):]) evidence = self.root / "evidence" evidence.mkdir(exist_ok=True) # The deploy command's stdout can contain Docker output; it is not JSON authority. (evidence / "deployment.json").write_text("fixture Docker output\n") bindings = re.findall(r"^ (\w+): \$\{\{ steps\.prepare\.outputs\.(\w+) \}\}$", step, re.MULTILINE) env = {**os.environ, "D3_EVIDENCE_ROOT": str(evidence)} env.update({name: prepared[output] for name, output in bindings}) return subprocess.run( ["bash", "-c", script], capture_output=True, text=True, env=env, timeout=10, ) def test_workflow_validates_the_durable_retirement_receipt_not_command_output(self): valid = { "schema": "nexuscrm.prod-predecessor-retirement/v1", "verified": True, "kind": "d3", "backendPort": 3104, "targetBackendPort": 3105, } for receipt, expected in ( (valid, True), (None, False), ({**valid, "verified": False}, False), ({**valid, "backendPort": 3106}, False), ({**valid, "targetBackendPort": 3104}, False), ): with self.subTest(receipt=receipt): self.state["previousRuntimeRetirement"] = receipt self.persist() result = self.workflow_receipt_check({ "release_root": str(self.context.env_root), "active_backend_port": "3104", "previous_backend_port": "3104", "target_backend_port": "3105", }) self.assertEqual(expected, result.returncode == 0, result.stderr) def test_d3_prepare_and_workflow_preserve_predecessor_after_cutover_retry(self): self.exercise_prepared_cutover_retry("d3") def test_classic_prepare_and_workflow_preserve_predecessor_after_cutover_retry(self): self.exercise_prepared_cutover_retry("legacy") def exercise_prepared_cutover_retry(self, previous_kind): """Real prepare -> interrupted closeout -> prepare retry -> workflow receipt.""" previous_is_d3 = previous_kind == "d3" self.live = prod.ProdRoute(3104, "slot-3104", previous_is_d3) self.context.envv["PROD_D3_PREVIOUS_ROUTE_KIND"] = previous_kind if not previous_is_d3: for unit in prod.previous_runtime_units(self.live): self.units[unit] = {"load_state": "loaded", "active_state": "active", "unit_file_state": "enabled"} previous_authority = self.root if previous_is_d3 else self.root.parent previous_root = previous_authority / "releases" / "staging-d3-release-previous" previous_env = previous_root / "env" / "runtime.env" previous_env.parent.mkdir(parents=True) (previous_root / "frontend-dist").mkdir() authority = { **{key: "fixture-value" for key in prod_prepare.REQUIRED_RUNTIME_KEYS}, "TIDB_HOST": "fixture.invalid", "TIDB_PORT": "4000", "TIDB_USER": "fixture-runtime", "TIDB_DATABASE": "nexuscrm", "ENV_ROOT": str(previous_root), "ENV_NAME": "multi-prod-d3-slot-3104", "BACKEND_PORT": "3104", "RELEASE_TAG": previous_root.name, "D3_RESOLVED_BOM_REFERENCE": self.pending["resolvedBomReference"], } previous_env.write_text("".join(f"{key}={value}\n" for key, value in authority.items())) (previous_authority / "current").symlink_to(previous_root) if previous_is_d3: (self.root / "current.json").write_text(json.dumps({ "schema": prod_prepare.CURRENT_SCHEMA, "slot": "slot-3104", "backendPort": 3104, "releaseTag": previous_root.name, "releaseRoot": str(previous_root), "resolvedBomReference": self.pending["resolvedBomReference"], })) nginx_conf = self.root / "nginx.conf" def write_route(port, is_d3): marker = "# NexusCRM D3 direct ingress run=fixture\n" if is_d3 else "" nginx_conf.write_text( marker + "server {\n server_name nexuscrm-new-prod.pingcap.com;\n" + f" location /api/ {{ proxy_pass http://127.0.0.1:{port}; }}\n}}\n" ) write_route(3104, previous_is_d3) # Let the real prepare create pending authority for the first promotion. (self.root / "in-progress.json").unlink() def verify_bom(reference, commit): self.assertEqual(self.pending["resolvedBomReference"], reference) self.assertEqual(self.pending["tagTargetCommit"], commit) with ExitStack() as stack: bind_fixture(stack, prod_prepare, "_require_tag_bom_reference", verify_bom) bind_fixture(stack, prod_prepare.os, "environ", { "PROD_TIDB_HOST": "fixture.invalid", "PROD_RUNTIME_TIDB_USER": "fixture-runtime", "D3_RESOLVED_BOM_REFERENCE": self.pending["resolvedBomReference"], }) def prepare(): return prod_prepare.prepare( release_tag=self.pending["releaseTag"], tag_target_commit=self.pending["tagTargetCommit"], multi_prod_root=self.root.parent, nginx_conf=nginx_conf, ) initial = prepare() self.assertEqual("3104", initial["active_backend_port"]) self.fail_public_call = 2 with self.assertRaisesRegex(prod.DeployError, "public readback"): prod.converge_prod(self.context) self.assertTrue((self.root / "in-progress.json").is_file()) write_route(3105, True) resumed = prepare() self.assertEqual("3105", resumed["active_backend_port"]) self.assertEqual(initial["target_slot"], resumed["target_slot"]) self.fail_public_call = None prod.converge_prod(self.context) self.assertFalse((self.root / "in-progress.json").exists()) self.assertEqual(3105, json.loads((self.root / "current.json").read_text())["backendPort"]) for prepared in (resumed, initial): result = self.workflow_receipt_check(prepared) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual("3104", prepared["previous_backend_port"]) if __name__ == "__main__": unittest.main()