test_deploy_cutover.py
6.54 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
"""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()