diff options
Diffstat (limited to 'tests/test_backends_servers.py')
| -rw-r--r-- | tests/test_backends_servers.py | 146 |
1 files changed, 0 insertions, 146 deletions
diff --git a/tests/test_backends_servers.py b/tests/test_backends_servers.py deleted file mode 100644 index 02b65e6..0000000 --- a/tests/test_backends_servers.py +++ /dev/null @@ -1,146 +0,0 @@ -"""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() |