"""U-041 tests: c_deploy_supervisor_rollback. Real subprocesses, real git repos, real filesystem operations throughout -- this is the highest-rework- risk Unit on PH5's critical path (development-plan.md's own note: "回滚本身 失败没有再下一层的自动兜底"), so its own "rollback itself fails" fallback is tested for real (a genuinely broken previous_instance_dir, a genuinely conflicting git revert), not assumed to work because the happy path does. 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_rollback, deploy_stage, deploy_supervisor from src import paths, schema def _init_git_repo(path): subprocess.run(["git", "init", "-q"], cwd=path, check=True) subprocess.run(["git", "config", "user.email", "test@graphlemon.local"], cwd=path, check=True) subprocess.run(["git", "config", "user.name", "test"], cwd=path, check=True) def _write(path, name, content): with open(os.path.join(path, name), "w", encoding="utf-8") as f: f.write(content) def _commit_all(path, message): subprocess.run(["git", "add", "-A"], cwd=path, check=True) subprocess.run(["git", "commit", "-q", "-m", message], cwd=path, check=True) return subprocess.run( ["git", "rev-parse", "HEAD"], cwd=path, capture_output=True, text=True, check=True ).stdout.strip() class DeployRollbackTestBase(unittest.TestCase): def setUp(self): self.home_tmp = tempfile.mkdtemp(prefix="deploy-rollback-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 = [] 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) def _seed_deployment(self, deployment_id, status, **extra): cols = ["deployment_id", "git_commit_hash", "previous_commit_hash", "status", "previous_instance_dir"] vals = [deployment_id, extra.get("git_commit_hash", "c1"), extra.get("previous_commit_hash", "c0"), status, extra.get("previous_instance_dir", "/prev")] self.conn.execute( "INSERT INTO deployment ({}) VALUES ({})".format(", ".join(cols), ", ".join("?" * len(cols))), vals ) self.conn.commit() class RollbackPathAHappyPathTest(DeployRollbackTestBase): def setUp(self): super().setUp() self.repo_dir = tempfile.mkdtemp(prefix="rollback-a-repo-") self.addCleanup(shutil.rmtree, self.repo_dir, ignore_errors=True) _init_git_repo(self.repo_dir) _write(self.repo_dir, "f.txt", "v1\n") self.commit_v1 = _commit_all(self.repo_dir, "v1") _write(self.repo_dir, "f.txt", "v2 (self-modification)\n") self.commit_v2 = _commit_all(self.repo_dir, "v2") def test_stops_candidate_deletes_db_reverts_commit_records_bug(self): candidate_db = os.path.join(self.home_tmp, "candidate.db") shutil.copy2(self.prod_db_path, candidate_db) proc = subprocess.Popen(["sleep", "300"]) self._cleanup_pids.append(proc.pid) self._seed_deployment("dep-a1", "health_check_failed", git_commit_hash=self.commit_v2) result = deploy_rollback.rollback_path_a( self.conn, "dep-a1", proc.pid, candidate_db, self.commit_v2, repo_root=self.repo_dir, description="canary_task failed: connection refused", ) self.assertEqual(result["status"], "rolled_back") self.assertFalse(deploy_supervisor.is_alive(proc.pid)) self.assertFalse(os.path.exists(candidate_db)) with open(os.path.join(self.repo_dir, "f.txt"), encoding="utf-8") as f: self.assertEqual(f.read(), "v1\n") # revert landed row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-a1'").fetchone() self.assertEqual(row[0], "rolled_back") bug = self.conn.execute( "SELECT root_cause_category, status, description FROM bug_record" ).fetchone() self.assertEqual(bug[0], "core_engine") self.assertEqual(bug[1], "resolved") self.assertIn("canary_task failed", bug[2]) class RollbackPathAFailureTest(DeployRollbackTestBase): def test_git_revert_conflict_marks_blocked_pending_human_and_raises(self): repo_dir = tempfile.mkdtemp(prefix="rollback-a-conflict-repo-") self.addCleanup(shutil.rmtree, repo_dir, ignore_errors=True) _init_git_repo(repo_dir) _write(repo_dir, "f.txt", "v1\n") _commit_all(repo_dir, "v1") _write(repo_dir, "f.txt", "v2 (self-modification)\n") commit_v2 = _commit_all(repo_dir, "v2") # A later commit that touches the SAME line v2 changed -- reverting # v2 now conflicts, the exact scenario design doc §10 open question # 3 describes ("主分支又有新的提交，revert 可能冲突"). _write(repo_dir, "f.txt", "v3 (unrelated later change, same line)\n") _commit_all(repo_dir, "v3") self._seed_deployment("dep-a2", "health_check_failed", git_commit_hash=commit_v2) with self.assertRaises(deploy_rollback.RollbackFailedError): deploy_rollback.rollback_path_a( self.conn, "dep-a2", candidate_pid=None, candidate_db_path=None, git_commit_hash=commit_v2, repo_root=repo_dir, ) row = self.conn.execute( "SELECT status, rollback_reason FROM deployment WHERE deployment_id = 'dep-a2'" ).fetchone() self.assertEqual(row[0], "blocked_pending_human") self.assertIn("git revert failed", row[1]) # repo left clean, not mid-conflict, for whoever investigates status_output = subprocess.run( ["git", "status", "--porcelain"], cwd=repo_dir, capture_output=True, text=True, check=True ).stdout self.assertEqual(status_output.strip(), "") class RollbackPathBHappyPathTest(DeployRollbackTestBase): def setUp(self): super().setUp() self.repo_dir = tempfile.mkdtemp(prefix="rollback-b-repo-") self.addCleanup(shutil.rmtree, self.repo_dir, ignore_errors=True) _init_git_repo(self.repo_dir) _write(self.repo_dir, "f.txt", "v1\n") self.commit_v1 = _commit_all(self.repo_dir, "v1") _write(self.repo_dir, "f.txt", "v2 (self-modification, buggy)\n") self.commit_v2 = _commit_all(self.repo_dir, "v2") # The "previously known-good" checkout -- a REAL worktree of THIS # actual graphlemon repo (needs real src/backend_server.py to relaunch), # not repo_dir (which has no real graphlemon source in it). self._prev_stash = subprocess.run( ["git", "stash", "create"], cwd=paths.repo_root(), capture_output=True, text=True, check=True ).stdout.strip() or deploy_stage._current_commit(paths.repo_root()) self.prev_dir = os.path.join(self.home_tmp, "previous-instance") subprocess.run( ["git", "worktree", "add", "--detach", self.prev_dir, self._prev_stash], cwd=paths.repo_root(), check=True, capture_output=True, text=True, ) self.addCleanup( lambda: subprocess.run( ["git", "worktree", "remove", "--force", self.prev_dir], cwd=paths.repo_root(), capture_output=True ) ) def test_stops_bad_live_revives_previous_reverts_commit_records_bug(self): bad_live_port = deploy_stage._find_free_port() bad_live = deploy_stage._launch_candidate(paths.repo_root(), bad_live_port, self.prod_db_path) self._cleanup_pids.append(bad_live.pid) self._wait_for_port(bad_live_port) self._seed_deployment( "dep-b1", "anomaly_detected", git_commit_hash=self.commit_v2, previous_instance_dir=self.prev_dir, ) rollback_port = deploy_stage._find_free_port() result = deploy_rollback.rollback_path_b( self.conn, "dep-b1", bad_live.pid, self.prev_dir, self.prod_db_path, self.commit_v2, repo_root=self.repo_dir, live_port=rollback_port, ) self._cleanup_pids.append( # find the revived process's pid via port not tracked -- best-effort cleanup bad_live.pid ) self.assertEqual(result["status"], "rolled_back") self.assertFalse(deploy_supervisor.is_alive(bad_live.pid)) with open(os.path.join(self.repo_dir, "f.txt"), encoding="utf-8") as f: self.assertEqual(f.read(), "v1\n") row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-b1'").fetchone() self.assertEqual(row[0], "rolled_back") bug = self.conn.execute("SELECT root_cause_category FROM bug_record").fetchone() self.assertEqual(bug[0], "core_engine") # revived instance genuinely serving on rollback_port health = deploy_health_check.check_http_health(rollback_port) self.assertEqual(health["status"], "ok") 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("port {} never came up".format(port)) class RollbackPathBFailureTest(DeployRollbackTestBase): """The Unit's own highest-risk scenario, tested for real: the old version ALSO can't start.""" def test_previous_instance_also_fails_to_start_marks_blocked_pending_human(self): repo_dir = tempfile.mkdtemp(prefix="rollback-b-fail-repo-") self.addCleanup(shutil.rmtree, repo_dir, ignore_errors=True) _init_git_repo(repo_dir) _write(repo_dir, "f.txt", "v1\n") _commit_all(repo_dir, "v1") _write(repo_dir, "f.txt", "v2\n") commit_v2 = _commit_all(repo_dir, "v2") broken_previous_dir = os.path.join(self.home_tmp, "broken-previous") os.makedirs(broken_previous_dir) # a real directory, but with NO python package in it at all self._seed_deployment( "dep-b2", "anomaly_detected", git_commit_hash=commit_v2, previous_instance_dir=broken_previous_dir, ) with self.assertRaises(deploy_rollback.RollbackFailedError): deploy_rollback.rollback_path_b( self.conn, "dep-b2", live_pid=None, previous_instance_dir=broken_previous_dir, prod_db_path=self.prod_db_path, git_commit_hash=commit_v2, repo_root=repo_dir, live_port=deploy_stage._find_free_port(), bind_timeout=3, ) row = self.conn.execute( "SELECT status, rollback_reason FROM deployment WHERE deployment_id = 'dep-b2'" ).fetchone() self.assertEqual(row[0], "blocked_pending_human") self.assertIn("rollback_failed", row[1]) # commit was NOT reverted -- rollback failed before it ever got there with open(os.path.join(repo_dir, "f.txt"), encoding="utf-8") as f: self.assertEqual(f.read(), "v2\n") if __name__ == "__main__": unittest.main()