"""U-039 tests: c_deploy_supervisor_cutover. Real subprocesses throughout (old Live, health-check-mode candidate, and the post-cutover restarted process) -- this Unit's whole job is process lifecycle around a real port/DB swap, so mocking that away would test nothing real. Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v """ import os import shutil import subprocess import sys import tempfile import time import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from supervisor import deploy_cutover, deploy_health_check, deploy_stage, deploy_supervisor from src import paths, schema def _free_port(): return deploy_stage._find_free_port() def _commit_to_test_against(): """Same rationale as test_deploy_stage.py's own helper -- git worktree only sees committed content, and this Unit's own changes may not be committed yet at test time.""" stash_sha = subprocess.run( ["git", "stash", "create"], cwd=paths.repo_root(), capture_output=True, text=True, check=True ).stdout.strip() return stash_sha or deploy_stage._current_commit(paths.repo_root()) class DeployCutoverTestBase(unittest.TestCase): def setUp(self): self.home_tmp = tempfile.mkdtemp(prefix="deploy-cutover-home-") self.addCleanup(shutil.rmtree, self.home_tmp, ignore_errors=True) self.old_home = os.environ.get("GRAPHLEMON_HOME") os.environ["GRAPHLEMON_HOME"] = self.home_tmp self.prod_db_path = os.path.join(self.home_tmp, "prod.db") prod_conn = schema.connect(self.prod_db_path) schema.create_schema(prod_conn) prod_conn.close() self.conn = schema.connect(":memory:") schema.create_schema(self.conn) self.addCleanup(self.conn.close) self._cleanup_pids = [] self._cleanup_worktrees = [] def tearDown(self): if self.old_home is None: os.environ.pop("GRAPHLEMON_HOME", None) else: os.environ["GRAPHLEMON_HOME"] = self.old_home for pid in self._cleanup_pids: if deploy_supervisor.is_alive(pid): os.kill(pid, 9) for wt in self._cleanup_worktrees: subprocess.run(["git", "worktree", "remove", "--force", wt], cwd=paths.repo_root(), capture_output=True) def _stage_and_health_check(self, port=None): commit = _commit_to_test_against() result = deploy_stage.stage_deployment( self.conn, commit, repo_root=paths.repo_root(), prod_db_path=self.prod_db_path, port=port or _free_port(), ) self._cleanup_pids.append(result["candidate_pid"]) self._cleanup_worktrees.append(result["candidate_instance_dir"]) self._wait_for_port(result["candidate_port"]) deploy_health_check.run_health_check( self.conn, result["deployment_id"], result["candidate_pid"], result["candidate_port"] ) return result def _wait_for_port(self, port, timeout=10): deadline = time.time() + timeout while time.time() < deadline: if deploy_health_check.check_http_health(port, timeout=1)["status"] == "ok": return time.sleep(0.2) self.fail("candidate on port {} never came up".format(port)) class CutoverHappyPathTest(DeployCutoverTestBase): def test_first_ever_cutover_no_prior_live_process(self): staged = self._stage_and_health_check() live_port = _free_port() result = deploy_cutover.cutover( self.conn, staged["deployment_id"], live_pid=None, candidate_pid=staged["candidate_pid"], candidate_instance_dir=staged["candidate_instance_dir"], prod_db_path=self.prod_db_path, live_port=live_port, sleep=lambda s: time.sleep(min(s, 0.2)), ) self._cleanup_pids.append(result["live_pid"]) self.assertEqual(result["status"], "live") self.assertEqual(result["live_port"], live_port) self.assertTrue(deploy_supervisor.is_alive(result["live_pid"])) self.assertNotEqual(result["live_pid"], staged["candidate_pid"]) # old candidate process was retired self.assertFalse(deploy_supervisor.is_alive(staged["candidate_pid"])) row = self.conn.execute( "SELECT status, cutover_at FROM deployment WHERE deployment_id = ?", (staged["deployment_id"],) ).fetchone() self.assertEqual(row[0], "live") self.assertIsNotNone(row[1]) # real HTTP round trip against the NEW live process, on live_port, # backed by prod_db_path (not the candidate's one-time DB copy) self._wait_for_port(live_port) health = deploy_health_check.check_http_health(live_port) self.assertEqual(health["status"], "ok") def test_cutover_stops_the_old_live_process(self): staged = self._stage_and_health_check() old_live_port = _free_port() old_live = deploy_stage._launch_candidate(paths.repo_root(), old_live_port, self.prod_db_path) self._cleanup_pids.append(old_live.pid) self._wait_for_port(old_live_port) self.assertTrue(deploy_supervisor.is_alive(old_live.pid)) new_live_port = _free_port() result = deploy_cutover.cutover( self.conn, staged["deployment_id"], live_pid=old_live.pid, candidate_pid=staged["candidate_pid"], candidate_instance_dir=staged["candidate_instance_dir"], prod_db_path=self.prod_db_path, live_port=new_live_port, ) self._cleanup_pids.append(result["live_pid"]) deadline = time.time() + 5 while time.time() < deadline and deploy_supervisor.is_alive(old_live.pid): time.sleep(0.1) self.assertFalse(deploy_supervisor.is_alive(old_live.pid)) class CutoverFailedTest(DeployCutoverTestBase): def test_bind_timeout_marks_blocked_pending_human_and_raises(self): staged = self._stage_and_health_check() with self.assertRaises(deploy_cutover.CutoverFailedError): deploy_cutover.cutover( self.conn, staged["deployment_id"], live_pid=None, candidate_pid=staged["candidate_pid"], candidate_instance_dir="/nonexistent/checkout/dir/that/cannot/run/anything", prod_db_path=self.prod_db_path, live_port=_free_port(), bind_timeout=1, sleep=lambda s: time.sleep(min(s, 0.1)), ) row = self.conn.execute( "SELECT status FROM deployment WHERE deployment_id = ?", (staged["deployment_id"],) ).fetchone() self.assertEqual(row[0], "blocked_pending_human") if __name__ == "__main__": unittest.main()