File size: 10,783 Bytes
b5b9c2e | 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 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 | """Tests for hermes_cli.gateway."""
from types import SimpleNamespace
from unittest.mock import patch, call
import hermes_cli.gateway as gateway
class TestSystemdLingerStatus:
def test_reports_enabled(self, monkeypatch):
monkeypatch.setattr(gateway, "is_linux", lambda: True)
monkeypatch.setattr(gateway, "is_termux", lambda: False)
monkeypatch.setenv("USER", "alice")
monkeypatch.setattr(
gateway.subprocess,
"run",
lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="yes\n", stderr=""),
)
monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/loginctl")
assert gateway.get_systemd_linger_status() == (True, "")
def test_reports_disabled(self, monkeypatch):
monkeypatch.setattr(gateway, "is_linux", lambda: True)
monkeypatch.setattr(gateway, "is_termux", lambda: False)
monkeypatch.setenv("USER", "alice")
monkeypatch.setattr(
gateway.subprocess,
"run",
lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="no\n", stderr=""),
)
monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/loginctl")
assert gateway.get_systemd_linger_status() == (False, "")
def test_reports_termux_as_not_supported(self, monkeypatch):
monkeypatch.setattr(gateway, "is_termux", lambda: True)
assert gateway.get_systemd_linger_status() == (None, "not supported in Termux")
def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys):
unit_path = tmp_path / "hermes-gateway.service"
unit_path.write_text("[Unit]\n")
monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (False, ""))
def fake_run(cmd, capture_output=False, text=False, check=False, **kwargs):
if cmd[:4] == ["systemctl", "--user", "status", gateway.get_service_name()]:
return SimpleNamespace(returncode=0, stdout="", stderr="")
if cmd[:3] == ["systemctl", "--user", "is-active"]:
return SimpleNamespace(returncode=0, stdout="active\n", stderr="")
raise AssertionError(f"Unexpected command: {cmd}")
monkeypatch.setattr(gateway.subprocess, "run", fake_run)
gateway.systemd_status(deep=False)
out = capsys.readouterr().out
assert "gateway service is running" in out
assert "Systemd linger is disabled" in out
assert "loginctl enable-linger" in out
def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys):
unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service"
monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path)
calls = []
helper_calls = []
def fake_run(cmd, check=False, **kwargs):
calls.append((cmd, check))
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway.subprocess, "run", fake_run)
monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True))
gateway.systemd_install(force=False)
out = capsys.readouterr().out
assert unit_path.exists()
assert [cmd for cmd, _ in calls] == [
["systemctl", "--user", "daemon-reload"],
["systemctl", "--user", "enable", gateway.get_service_name()],
]
assert helper_calls == [True]
assert "User service installed and enabled" in out
def test_systemd_install_system_scope_skips_linger_and_uses_systemctl(monkeypatch, tmp_path, capsys):
unit_path = tmp_path / "etc" / "systemd" / "system" / "hermes-gateway.service"
monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(
gateway,
"generate_systemd_unit",
lambda system=False, run_as_user=None: f"scope={system} user={run_as_user}\n",
)
monkeypatch.setattr(gateway, "_require_root_for_system_service", lambda action: None)
calls = []
helper_calls = []
def fake_run(cmd, check=False, **kwargs):
calls.append((cmd, check))
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway.subprocess, "run", fake_run)
monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True))
gateway.systemd_install(force=False, system=True, run_as_user="alice")
out = capsys.readouterr().out
assert unit_path.exists()
assert unit_path.read_text(encoding="utf-8") == "scope=True user=alice\n"
assert [cmd for cmd, _ in calls] == [
["systemctl", "daemon-reload"],
["systemctl", "enable", gateway.get_service_name()],
]
assert helper_calls == []
assert "Configured to run as: alice" not in out # generated test unit has no User= line
assert "System service installed and enabled" in out
def test_conflicting_systemd_units_warning(monkeypatch, tmp_path, capsys):
user_unit = tmp_path / "user" / "hermes-gateway.service"
system_unit = tmp_path / "system" / "hermes-gateway.service"
user_unit.parent.mkdir(parents=True)
system_unit.parent.mkdir(parents=True)
user_unit.write_text("[Unit]\n", encoding="utf-8")
system_unit.write_text("[Unit]\n", encoding="utf-8")
monkeypatch.setattr(
gateway,
"get_systemd_unit_path",
lambda system=False: system_unit if system else user_unit,
)
gateway.print_systemd_scope_conflict_warning()
out = capsys.readouterr().out
assert "Both user and system gateway services are installed" in out
assert "hermes gateway uninstall" in out
assert "--system" in out
def test_install_linux_gateway_from_setup_system_choice_without_root_prints_followup(monkeypatch, capsys):
monkeypatch.setattr(gateway, "prompt_linux_gateway_install_scope", lambda: "system")
monkeypatch.setattr(gateway.os, "geteuid", lambda: 1000)
monkeypatch.setattr(gateway, "_default_system_service_user", lambda: "alice")
monkeypatch.setattr(gateway, "systemd_install", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not install")))
scope, did_install = gateway.install_linux_gateway_from_setup(force=False)
out = capsys.readouterr().out
assert (scope, did_install) == ("system", False)
assert "sudo hermes gateway install --system --run-as-user alice" in out
assert "sudo hermes gateway start --system" in out
def test_install_linux_gateway_from_setup_system_choice_as_root_installs(monkeypatch):
monkeypatch.setattr(gateway, "prompt_linux_gateway_install_scope", lambda: "system")
monkeypatch.setattr(gateway.os, "geteuid", lambda: 0)
monkeypatch.setattr(gateway, "_default_system_service_user", lambda: "alice")
calls = []
monkeypatch.setattr(
gateway,
"systemd_install",
lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)),
)
scope, did_install = gateway.install_linux_gateway_from_setup(force=True)
assert (scope, did_install) == ("system", True)
assert calls == [(True, True, "alice")]
# ---------------------------------------------------------------------------
# _wait_for_gateway_exit
# ---------------------------------------------------------------------------
class TestWaitForGatewayExit:
"""PID-based wait with force-kill on timeout."""
def test_returns_immediately_when_no_pid(self, monkeypatch):
"""If get_running_pid returns None, exit instantly."""
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
# Should return without sleeping at all.
gateway._wait_for_gateway_exit(timeout=1.0, force_after=0.5)
def test_returns_when_process_exits_gracefully(self, monkeypatch):
"""Process exits after a couple of polls — no SIGKILL needed."""
poll_count = 0
def mock_get_running_pid():
nonlocal poll_count
poll_count += 1
return 12345 if poll_count <= 2 else None
monkeypatch.setattr("gateway.status.get_running_pid", mock_get_running_pid)
monkeypatch.setattr("time.sleep", lambda _: None)
gateway._wait_for_gateway_exit(timeout=10.0, force_after=999.0)
# Should have polled until None was returned.
assert poll_count == 3
def test_force_kills_after_grace_period(self, monkeypatch):
"""When the process doesn't exit, force-kill the saved PID."""
# Simulate monotonic time advancing past force_after
call_num = 0
def fake_monotonic():
nonlocal call_num
call_num += 1
# First two calls: initial deadline + force_deadline setup (time 0)
# Then each loop iteration advances time
return call_num * 2.0 # 2, 4, 6, 8, ...
kills = []
def mock_terminate(pid, force=False):
kills.append((pid, force))
# get_running_pid returns the PID until kill is sent, then None
def mock_get_running_pid():
return None if kills else 42
monkeypatch.setattr("time.monotonic", fake_monotonic)
monkeypatch.setattr("time.sleep", lambda _: None)
monkeypatch.setattr("gateway.status.get_running_pid", mock_get_running_pid)
monkeypatch.setattr(gateway, "terminate_pid", mock_terminate)
gateway._wait_for_gateway_exit(timeout=10.0, force_after=5.0)
assert (42, True) in kills
def test_handles_process_already_gone_on_kill(self, monkeypatch):
"""ProcessLookupError during force-kill is not fatal."""
call_num = 0
def fake_monotonic():
nonlocal call_num
call_num += 1
return call_num * 3.0 # Jump past force_after quickly
def mock_terminate(pid, force=False):
raise ProcessLookupError
monkeypatch.setattr("time.monotonic", fake_monotonic)
monkeypatch.setattr("time.sleep", lambda _: None)
monkeypatch.setattr("gateway.status.get_running_pid", lambda: 99)
monkeypatch.setattr(gateway, "terminate_pid", mock_terminate)
# Should not raise — ProcessLookupError means it's already gone.
gateway._wait_for_gateway_exit(timeout=10.0, force_after=2.0)
def test_kill_gateway_processes_force_uses_helper(self, monkeypatch):
calls = []
monkeypatch.setattr(gateway, "find_gateway_pids", lambda exclude_pids=None, all_profiles=False: [11, 22])
monkeypatch.setattr(gateway, "terminate_pid", lambda pid, force=False: calls.append((pid, force)))
killed = gateway.kill_gateway_processes(force=True)
assert killed == 2
assert calls == [(11, True), (22, True)]
|