test_deploy_supervisor.py
7.02 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
188
189
190
191
192
193
"""U-036 tests: Deploy Supervisor process skeleton -- lifecycle only (PID
file, status file, start/stop), no actual deploy logic (that's U-037+).
GRAPHLEMON_HOME is always overridden to a throwaway tempdir so these tests
never touch the real user's ~/.graphlemon.
Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v
"""
import os
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import unittest
from supervisor import deploy_supervisor as ds
from src import paths
class DeploySupervisorTestBase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix="deploy-supervisor-test-")
self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
self.old_home = os.environ.get("GRAPHLEMON_HOME")
os.environ["GRAPHLEMON_HOME"] = self.tmp
def tearDown(self):
if self.old_home is None:
os.environ.pop("GRAPHLEMON_HOME", None)
else:
os.environ["GRAPHLEMON_HOME"] = self.old_home
class RuntimePathsTest(DeploySupervisorTestBase):
def test_runtime_dir_created_under_platform_home(self):
d = ds.runtime_dir()
self.assertTrue(os.path.isdir(d))
self.assertTrue(d.startswith(paths.platform_home()))
def test_pid_and_status_file_paths_live_under_runtime_dir(self):
self.assertEqual(os.path.dirname(ds.pid_file_path()), ds.runtime_dir())
self.assertEqual(os.path.dirname(ds.status_file_path()), ds.runtime_dir())
class IsAliveTest(unittest.TestCase):
def test_current_process_is_alive(self):
self.assertTrue(ds.is_alive(os.getpid()))
def test_zero_or_negative_pid_is_not_alive(self):
self.assertFalse(ds.is_alive(0))
self.assertFalse(ds.is_alive(-1))
def test_non_int_is_not_alive(self):
self.assertFalse(ds.is_alive("not-a-pid"))
def test_a_pid_that_does_not_exist_is_not_alive(self):
# PID_MAX on Linux/macOS is well below this; astronomically unlikely
# to collide with a real running process.
self.assertFalse(ds.is_alive(2**30))
class StatusFileTest(DeploySupervisorTestBase):
def test_write_then_read_status_round_trips(self):
ds.write_status("running", detail="test detail")
status = ds.read_status()
self.assertEqual(status["status"], "running")
self.assertEqual(status["detail"], "test detail")
self.assertEqual(status["pid"], os.getpid())
def test_read_status_returns_none_when_no_file_yet(self):
self.assertIsNone(ds.read_status())
def test_write_status_is_atomic_no_tmp_file_left_behind(self):
ds.write_status("running")
entries = os.listdir(ds.runtime_dir())
self.assertEqual(entries, ["supervisor.status.json"])
class IsRunningTest(DeploySupervisorTestBase):
def test_false_when_no_pid_file(self):
self.assertFalse(ds.is_running())
def test_true_when_pid_file_points_to_a_live_process(self):
with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
handle.write(str(os.getpid()))
self.assertTrue(ds.is_running())
def test_false_when_pid_file_points_to_a_dead_process(self):
with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
handle.write(str(2**30))
self.assertFalse(ds.is_running())
def test_false_when_pid_file_is_corrupt(self):
with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
handle.write("not-a-number")
self.assertFalse(ds.is_running())
class RunForegroundTest(DeploySupervisorTestBase):
def test_writes_pid_file_and_running_status_during_the_loop(self):
seen_running = []
def poll_fn():
seen_running.append(ds.is_running())
seen_running.append(ds.read_status()["status"])
ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=1)
self.assertEqual(seen_running, [True, "running"])
def test_calls_poll_fn_exactly_iterations_times(self):
calls = []
ds.run_foreground(poll_fn=lambda: calls.append(1), poll_interval=0, iterations=3)
self.assertEqual(len(calls), 3)
def test_writes_stopped_status_and_removes_pid_file_on_normal_completion(self):
ds.run_foreground(poll_fn=None, poll_interval=0, iterations=1)
self.assertEqual(ds.read_status()["status"], "stopped")
self.assertFalse(os.path.exists(ds.pid_file_path()))
def test_sigterm_breaks_the_loop_and_still_writes_stopped_status(self):
calls = []
def poll_fn():
calls.append(1)
if len(calls) == 2:
os.kill(os.getpid(), signal.SIGTERM)
# iterations=None (would run forever) -- SIGTERM is the only exit.
ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=None)
self.assertEqual(len(calls), 2)
self.assertEqual(ds.read_status()["status"], "stopped")
self.assertFalse(os.path.exists(ds.pid_file_path()))
def test_poll_fn_exception_propagates_but_cleanup_still_runs(self):
def poll_fn():
raise RuntimeError("boom")
with self.assertRaises(RuntimeError):
ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=1)
self.assertEqual(ds.read_status()["status"], "stopped")
self.assertFalse(os.path.exists(ds.pid_file_path()))
class RealSubprocessLifecycleTest(DeploySupervisorTestBase):
"""Not mocked -- actually spawns `python3 -m supervisor.deploy_supervisor`
as a real independent OS process and stops it for real, per this Unit's
Done criterion ("骨架能独立启停")."""
def test_start_detached_then_stop(self):
env = dict(os.environ)
# start_detached itself calls is_running()/Popen using the current
# process's env var (already patched by setUp); the child process
# inherits it via Popen's default env=None -> os.environ passthrough.
pid = ds.start_detached(poll_interval=0.1)
try:
deadline = time.time() + 5
while time.time() < deadline and not ds.is_running():
time.sleep(0.05)
self.assertTrue(ds.is_running())
self.assertTrue(ds.is_alive(pid))
stopped = ds.stop(timeout=5)
self.assertTrue(stopped)
self.assertFalse(ds.is_running())
finally:
if ds.is_alive(pid):
os.kill(pid, signal.SIGKILL)
def test_start_detached_twice_raises_while_already_running(self):
pid = ds.start_detached(poll_interval=0.1)
try:
deadline = time.time() + 5
while time.time() < deadline and not ds.is_running():
time.sleep(0.05)
self.assertTrue(ds.is_running())
with self.assertRaises(RuntimeError):
ds.start_detached(poll_interval=0.1)
finally:
ds.stop(timeout=5)
if ds.is_alive(pid):
os.kill(pid, signal.SIGKILL)
def test_stop_returns_false_when_nothing_is_running(self):
self.assertFalse(ds.stop(timeout=1))
if __name__ == "__main__":
unittest.main()