test_deploy_health_check.py
7.94 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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
"""U-038 tests: c_deploy_supervisor_health_check. Real HTTP server (in-process
thread, same pattern src/test_backend_server.py already uses), real project/
task creation round trips for the canary check -- nothing here is mocked
except process liveness timing (injectable sleep/now, same convention as
work_node_run's own quota-wait tests).
Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v
"""
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import threading
from supervisor import deploy_health_check as dhc
from src import backend_server, schema
class RealServerTestBase(unittest.TestCase):
def setUp(self):
self.server = backend_server.create_server(db_path=":memory:")
self.port = self.server.server_address[1]
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
class CheckProcessAliveTest(unittest.TestCase):
def test_alive_process_returns_ok_immediately(self):
signal = dhc.check_process_alive(os.getpid())
self.assertEqual(signal, {"signal_type": "process_alive", "status": "ok", "detail": None})
def test_dead_pid_times_out_and_fails(self):
clock = {"t": 0.0}
sleeps = []
signal = dhc.check_process_alive(
2**30, timeout=5, sleep=lambda s: (sleeps.append(s), clock.__setitem__("t", clock["t"] + s)),
now=lambda: clock["t"],
)
self.assertEqual(signal["signal_type"], "process_alive")
self.assertEqual(signal["status"], "fail")
self.assertTrue(sleeps)
def test_process_that_becomes_alive_mid_poll_still_passes(self):
calls = {"n": 0}
def fake_is_alive(pid):
calls["n"] += 1
return calls["n"] >= 3
with patch("supervisor.deploy_supervisor.is_alive", side_effect=fake_is_alive):
signal = dhc.check_process_alive(99999, timeout=5, sleep=lambda s: None, now=lambda: 0.0)
self.assertEqual(signal["status"], "ok")
self.assertEqual(calls["n"], 3)
class CheckHttpHealthTest(RealServerTestBase):
def test_healthy_server_returns_ok(self):
signal = dhc.check_http_health(self.port)
self.assertEqual(signal, {"signal_type": "http_health", "status": "ok", "detail": None})
def test_broken_connection_returns_fail(self):
self.server.conn.close()
signal = dhc.check_http_health(self.port)
self.assertEqual(signal["signal_type"], "http_health")
self.assertEqual(signal["status"], "fail")
def test_nothing_listening_on_port_returns_fail(self):
clock = {"t": 0.0}
signal = dhc.check_http_health(
1, timeout=5, # privileged/unbound port, nothing listening
sleep=lambda s: clock.__setitem__("t", clock["t"] + s), now=lambda: clock["t"],
)
self.assertEqual(signal["status"], "fail")
def test_connection_refused_polls_until_server_actually_listening(self):
# 2026-08-03 regression: a real self-hosting run showed
# process_alive can read 'ok' before the candidate has finished
# binding its listen socket -- check_http_health must tolerate
# that same startup window instead of failing on the first
# ECONNREFUSED, same as check_process_alive already tolerates a
# not-yet-existing pid.
server = backend_server.create_server(db_path=":memory:")
real_port = server.server_address[1]
self.addCleanup(server.server_close)
def start_server_after_delay():
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
self.addCleanup(lambda: (server.shutdown(), thread.join(timeout=5)))
timer = threading.Timer(0.3, start_server_after_delay)
timer.start()
self.addCleanup(timer.cancel)
signal = dhc.check_http_health(real_port, timeout=5)
self.assertEqual(signal, {"signal_type": "http_health", "status": "ok", "detail": None})
def test_real_http_error_response_fails_immediately_without_polling(self):
# a genuine non-2xx response IS a definitive answer from a live
# server -- must not be treated the same as "not up yet".
sleeps = []
with patch.object(dhc.urllib.request, "urlopen") as mock_urlopen:
error = dhc.urllib.error.HTTPError("url", 500, "Internal Server Error", {}, None)
mock_urlopen.side_effect = error
signal = dhc.check_http_health(self.port, timeout=5, sleep=lambda s: sleeps.append(s), now=lambda: 0.0)
self.assertEqual(signal["status"], "fail")
self.assertIn("500", signal["detail"])
self.assertEqual(sleeps, [])
mock_urlopen.assert_called_once()
class CheckCanaryTaskTest(RealServerTestBase):
def test_real_project_and_task_round_trip_passes(self):
signal = dhc.check_canary_task(self.port)
self.assertEqual(signal, {"signal_type": "canary_task", "status": "ok", "detail": None})
conn = self.server.conn
count = conn.execute("SELECT COUNT(*) FROM task WHERE goal_text LIKE 'deploy-supervisor canary%'").fetchone()[0]
self.assertEqual(count, 1)
def test_nothing_listening_returns_fail(self):
signal = dhc.check_canary_task(1)
self.assertEqual(signal["signal_type"], "canary_task")
self.assertEqual(signal["status"], "fail")
class RunHealthCheckTest(RealServerTestBase):
def setUp(self):
super().setUp()
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
self.conn.execute(
"INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
"previous_instance_dir) VALUES ('dep-1', 'c1', 'c0', 'candidate_starting', '/prev')"
)
self.conn.commit()
self.addCleanup(self.conn.close)
def test_all_checks_pass_marks_health_check_passed(self):
result = dhc.run_health_check(self.conn, "dep-1", os.getpid(), self.port)
self.assertEqual(result["status"], "health_check_passed")
self.assertEqual([s["signal_type"] for s in result["signals"]], ["process_alive", "http_health", "canary_task"])
self.assertTrue(all(s["status"] == "ok" for s in result["signals"]))
row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone()
self.assertEqual(row[0], "health_check_passed")
signal_count = self.conn.execute(
"SELECT COUNT(*) FROM deploy_health_signal WHERE deployment_id = 'dep-1'"
).fetchone()[0]
self.assertEqual(signal_count, 3)
def test_process_alive_failure_stops_before_http_and_canary_checks(self):
result = dhc.run_health_check(
self.conn, "dep-1", 2**30, self.port,
startup_timeout=0, sleep=lambda s: None, now=lambda: 0.0,
)
self.assertEqual(result["status"], "health_check_failed")
self.assertEqual(len(result["signals"]), 1)
self.assertEqual(result["signals"][0]["signal_type"], "process_alive")
row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone()
self.assertEqual(row[0], "health_check_failed")
signal_count = self.conn.execute(
"SELECT COUNT(*) FROM deploy_health_signal WHERE deployment_id = 'dep-1'"
).fetchone()[0]
self.assertEqual(signal_count, 1) # http_health/canary_task never ran
def test_http_health_failure_stops_before_canary_check(self):
self.server.conn.close()
result = dhc.run_health_check(self.conn, "dep-1", os.getpid(), self.port)
self.assertEqual(result["status"], "health_check_failed")
self.assertEqual([s["signal_type"] for s in result["signals"]], ["process_alive", "http_health"])
if __name__ == "__main__":
unittest.main()