File size: 5,967 Bytes
36dada9 | 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 | import signal
from types import SimpleNamespace
import pytest
import TerraFin.interface.server as server_module
def test_stop_server_waits_for_process_exit_and_port_release(monkeypatch) -> None:
kill_calls: list[tuple[int, int]] = []
wait_process_calls: list[tuple[int, float, float]] = []
wait_port_calls: list[tuple[str, int, float, float]] = []
removed: list[str] = []
monkeypatch.setattr(server_module, "_read_pid", lambda: 1234)
monkeypatch.setattr(server_module, "_is_process_alive", lambda pid: pid == 1234)
monkeypatch.setattr(
server_module,
"get_runtime_config",
lambda: SimpleNamespace(host="127.0.0.1", port=8001),
)
monkeypatch.setattr(server_module, "_find_listener_pid", lambda port: 1234)
monkeypatch.setattr(server_module.os, "kill", lambda pid, sig: kill_calls.append((pid, sig)))
monkeypatch.setattr(
server_module,
"_wait_for_process_exit",
lambda pid, timeout_s=5.0, poll_interval_s=0.1: (
wait_process_calls.append((pid, timeout_s, poll_interval_s)) or True
),
)
monkeypatch.setattr(
server_module,
"_wait_for_port_release",
lambda host, port, timeout_s=5.0, poll_interval_s=0.1: (
wait_port_calls.append((host, port, timeout_s, poll_interval_s)) or True
),
)
monkeypatch.setattr(server_module, "_remove_pid_file", lambda: removed.append("removed"))
assert server_module.stop_server() is True
assert kill_calls == [(1234, signal.SIGTERM)]
assert wait_process_calls == [(1234, 5.0, 0.1)]
assert wait_port_calls == [("127.0.0.1", 8001, 5.0, 0.1)]
assert removed == ["removed"]
def test_stop_server_escalates_to_sigkill_when_process_does_not_exit(monkeypatch) -> None:
kill_calls: list[tuple[int, int]] = []
wait_process_calls: list[tuple[int, float, float]] = []
wait_results = iter([False, True])
monkeypatch.setattr(server_module, "_read_pid", lambda: 5678)
monkeypatch.setattr(server_module, "_is_process_alive", lambda pid: pid == 5678)
monkeypatch.setattr(
server_module,
"get_runtime_config",
lambda: SimpleNamespace(host="127.0.0.1", port=8001),
)
monkeypatch.setattr(server_module, "_find_listener_pid", lambda port: 5678)
monkeypatch.setattr(server_module.os, "kill", lambda pid, sig: kill_calls.append((pid, sig)))
monkeypatch.setattr(
server_module,
"_wait_for_process_exit",
lambda pid, timeout_s=5.0, poll_interval_s=0.1: (
wait_process_calls.append((pid, timeout_s, poll_interval_s)) or next(wait_results)
),
)
monkeypatch.setattr(
server_module, "_wait_for_port_release", lambda host, port, timeout_s=5.0, poll_interval_s=0.1: True
)
monkeypatch.setattr(server_module, "_remove_pid_file", lambda: None)
assert server_module.stop_server() is True
assert kill_calls == [(5678, signal.SIGTERM), (5678, signal.SIGKILL)]
assert wait_process_calls == [(5678, 5.0, 0.1), (5678, 2.0, 0.1)]
def test_server_status_recovers_listener_when_pid_file_is_missing(monkeypatch) -> None:
monkeypatch.setattr(server_module, "_read_pid", lambda: None)
monkeypatch.setattr(
server_module,
"get_runtime_config",
lambda: SimpleNamespace(host="127.0.0.1", port=8001),
)
monkeypatch.setattr(server_module, "_find_listener_pid", lambda port: 2468)
assert server_module.server_status() == (True, 2468)
def test_stop_server_uses_listener_pid_when_pid_file_is_missing(monkeypatch) -> None:
kill_calls: list[tuple[int, int]] = []
removed: list[str] = []
monkeypatch.setattr(server_module, "_read_pid", lambda: None)
monkeypatch.setattr(
server_module,
"get_runtime_config",
lambda: SimpleNamespace(host="127.0.0.1", port=8001),
)
monkeypatch.setattr(server_module, "_find_listener_pid", lambda port: 2468)
monkeypatch.setattr(server_module, "_is_process_alive", lambda pid: pid == 2468)
monkeypatch.setattr(server_module.os, "kill", lambda pid, sig: kill_calls.append((pid, sig)))
monkeypatch.setattr(
server_module,
"_wait_for_process_exit",
lambda pid, timeout_s=5.0, poll_interval_s=0.1: True,
)
monkeypatch.setattr(
server_module,
"_wait_for_port_release",
lambda host, port, timeout_s=5.0, poll_interval_s=0.1: True,
)
monkeypatch.setattr(server_module, "_remove_pid_file", lambda: removed.append("removed"))
assert server_module.stop_server() is True
assert kill_calls == [(2468, signal.SIGTERM)]
assert removed == ["removed"]
def test_start_server_raises_when_process_never_binds_port(monkeypatch, tmp_path) -> None:
class _FakeProc:
pid = 4321
@staticmethod
def poll():
return None
writes: list[int] = []
removed: list[str] = []
monkeypatch.setattr(
server_module,
"get_runtime_config",
lambda: SimpleNamespace(host="127.0.0.1", port=8001),
)
monkeypatch.setattr(server_module, "_resolve_server_pid", lambda runtime_config=None: None)
monkeypatch.setattr(server_module, "_port_has_listener", lambda host, port: False)
monkeypatch.setattr(server_module.subprocess, "Popen", lambda *args, **kwargs: _FakeProc())
monkeypatch.setattr(
server_module,
"_wait_for_server_startup",
lambda proc, host, port, timeout_s=5.0, poll_interval_s=0.1: False,
)
monkeypatch.setattr(server_module, "_write_pid", lambda pid: writes.append(pid))
monkeypatch.setattr(server_module, "_remove_pid_file", lambda: removed.append("removed"))
monkeypatch.setattr(server_module, "SERVER_LOG_FILE", tmp_path / "interface_server.log")
with pytest.raises(RuntimeError, match="Check interface_server.log"):
server_module.start_server()
assert writes == [4321]
assert removed == ["removed"]
|