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
194
195
196
197
|
"""Tests for the server lifecycle module (backends/servers.py)."""
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from backends import ServerSpec, servers
class StartTests(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.dir = Path(self._tmp.name)
# A fake executable so Path(argv[0]).exists() passes.
self.exe = self.dir / "fake_server"
self.exe.write_bytes(b"#!/bin/sh\n")
self.spec = ServerSpec("test", "http://127.0.0.1:9999",
[str(self.exe), "--port", "9999"])
def tearDown(self):
self._tmp.cleanup()
def test_returns_false_when_executable_missing(self):
spec = ServerSpec("nope", "http://127.0.0.1:1", ["/no/such/binary"])
with patch.object(servers, "LOG_DIR", self.dir):
self.assertFalse(servers.start(spec))
def test_noop_when_already_running(self):
with patch.object(servers, "LOG_DIR", self.dir), \
patch("backends.common.server_running", return_value=True), \
patch("subprocess.Popen") as mk:
self.assertTrue(servers.start(self.spec))
mk.assert_not_called()
def test_happy_path_spawns_and_polls_until_ready(self):
proc = MagicMock()
proc.pid = 4242
proc.poll.return_value = None # process still running
# server_running: False on the pre-check, True once inside the loop.
with patch.object(servers, "LOG_DIR", self.dir), \
patch("subprocess.Popen", return_value=proc) as mk, \
patch("backends.common.server_running",
side_effect=[False, True]), \
patch("time.sleep"):
ok = servers.start(self.spec)
self.assertTrue(ok)
mk.assert_called_once()
# Pid file written.
self.assertEqual(
(self.dir / "test-server.pid").read_text(encoding="utf-8"),
"4242")
def test_returns_false_when_process_exits_early(self):
proc = MagicMock()
proc.pid = 99
proc.poll.return_value = 1 # exited with code 1
with patch.object(servers, "LOG_DIR", self.dir), \
patch("subprocess.Popen", return_value=proc), \
patch("backends.common.server_running", return_value=False), \
patch("time.sleep"):
ok = servers.start(self.spec)
self.assertFalse(ok)
# Pid file cleaned up after early exit.
self.assertFalse((self.dir / "test-server.pid").exists())
def test_returns_false_on_timeout(self):
proc = MagicMock()
proc.pid = 7
proc.poll.return_value = None
# time.time: first call < deadline loop entry, then past deadline.
times = iter([0.0, float(servers.SERVER_START_TIMEOUT + 1)])
with patch.object(servers, "LOG_DIR", self.dir), \
patch("subprocess.Popen", return_value=proc), \
patch("backends.common.server_running", return_value=False), \
patch("time.sleep"), \
patch("time.time", side_effect=lambda: next(times)):
ok = servers.start(self.spec)
self.assertFalse(ok)
class StopTests(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.dir = Path(self._tmp.name)
def tearDown(self):
self._tmp.cleanup()
def _write_pid(self, name, pid):
(self.dir / f"{name}-server.pid").write_text(str(pid),
encoding="utf-8")
def test_returns_false_when_no_pid_file(self):
with patch.object(servers, "LOG_DIR", self.dir):
self.assertFalse(servers.stop("test"))
def test_stops_alive_process_and_removes_pid_file(self):
self._write_pid("test", 1234)
with patch.object(servers, "LOG_DIR", self.dir), \
patch.object(servers, "_pid_alive", return_value=True), \
patch.object(servers, "_kill_pid", return_value=True) as mk:
ok = servers.stop("test")
self.assertTrue(ok)
mk.assert_called_once_with(1234)
self.assertFalse((self.dir / "test-server.pid").exists())
def test_already_dead_returns_true_and_cleans_pid_file(self):
self._write_pid("test", 1234)
with patch.object(servers, "LOG_DIR", self.dir), \
patch.object(servers, "_pid_alive", return_value=False), \
patch.object(servers, "_kill_pid") as mk:
ok = servers.stop("test")
self.assertTrue(ok)
mk.assert_not_called()
self.assertFalse((self.dir / "test-server.pid").exists())
def test_corrupt_pid_file_returns_false_and_cleans(self):
(self.dir / "test-server.pid").write_text("not-a-number",
encoding="utf-8")
with patch.object(servers, "LOG_DIR", self.dir):
self.assertFalse(servers.stop("test"))
self.assertFalse((self.dir / "test-server.pid").exists())
class ManagesTests(unittest.TestCase):
"""manages(): a live recorded pid marks a server as ours."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.dir = Path(self._tmp.name)
self.specs = [ServerSpec("test", "http://127.0.0.1:9999", [])]
def tearDown(self):
self._tmp.cleanup()
def _write_pid(self, name, pid):
(self.dir / f"{name}-server.pid").write_text(str(pid),
encoding="utf-8")
def test_false_without_pid_file(self):
with patch.object(servers, "LOG_DIR", self.dir):
self.assertFalse(servers.manages(self.specs))
def test_true_with_live_recorded_pid(self):
self._write_pid("test", 4242)
with patch.object(servers, "LOG_DIR", self.dir), \
patch.object(servers, "_pid_alive", return_value=True):
self.assertTrue(servers.manages(self.specs))
def test_false_with_dead_recorded_pid(self):
self._write_pid("test", 4242)
with patch.object(servers, "LOG_DIR", self.dir), \
patch.object(servers, "_pid_alive", return_value=False):
self.assertFalse(servers.manages(self.specs))
def test_false_with_corrupt_pid_file(self):
(self.dir / "test-server.pid").write_text("junk",
encoding="utf-8")
with patch.object(servers, "LOG_DIR", self.dir), \
patch.object(servers, "_pid_alive",
return_value=True) as mk_alive:
self.assertFalse(servers.manages(self.specs))
mk_alive.assert_not_called()
def test_true_when_any_spec_is_ours(self):
other = ServerSpec("other", "http://127.0.0.1:9998", [])
with patch.object(servers, "LOG_DIR", self.dir), \
patch.object(servers, "_pid_alive", return_value=True):
self._write_pid("test", 4242)
self.assertTrue(servers.manages([other] + self.specs))
# The live pid belongs to 'test'; 'other' alone stays unmanaged.
with patch.object(servers, "LOG_DIR", self.dir):
self.assertFalse(servers.manages([other]))
class PidForTests(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.dir = Path(self._tmp.name)
def tearDown(self):
self._tmp.cleanup()
def test_none_when_no_pid_file(self):
with patch.object(servers, "LOG_DIR", self.dir):
self.assertIsNone(servers.pid_for("test"))
def test_returns_pid_from_file(self):
(self.dir / "test-server.pid").write_text("555\n",
encoding="utf-8")
with patch.object(servers, "LOG_DIR", self.dir):
self.assertEqual(servers.pid_for("test"), 555)
if __name__ == "__main__":
unittest.main()
|
