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
|
"""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 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()
|
