test_deploy_health_check.py
5.96 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
"""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):
signal = dhc.check_http_health(1) # privileged/unbound port, nothing listening
self.assertEqual(signal["status"], "fail")
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()