test_gateway.py
1 """Tests for hermes_cli.gateway.""" 2 3 import sys 4 from types import ModuleType, SimpleNamespace 5 from unittest.mock import patch, call 6 7 import pytest 8 9 import hermes_cli.gateway as gateway 10 11 12 def _install_fake_gateway_run(monkeypatch, start_gateway): 13 module = ModuleType("gateway.run") 14 module.start_gateway = start_gateway 15 monkeypatch.setitem(sys.modules, "gateway.run", module) 16 17 18 def test_run_gateway_exits_cleanly_on_keyboard_interrupt(monkeypatch, capsys): 19 calls = [] 20 21 def fake_start_gateway(*, replace, verbosity): 22 calls.append((replace, verbosity)) 23 return object() 24 25 def fake_asyncio_run(coro): 26 raise KeyboardInterrupt 27 28 _install_fake_gateway_run(monkeypatch, fake_start_gateway) 29 monkeypatch.setattr(gateway.asyncio, "run", fake_asyncio_run) 30 31 gateway.run_gateway() 32 33 out = capsys.readouterr().out 34 assert calls == [(False, 0)] 35 assert "Press Ctrl+C to stop" in out 36 assert "Gateway stopped." in out 37 38 39 def test_run_gateway_exits_nonzero_when_start_gateway_reports_failure(monkeypatch): 40 calls = [] 41 42 def fake_start_gateway(*, replace, verbosity): 43 calls.append((replace, verbosity)) 44 return object() 45 46 _install_fake_gateway_run(monkeypatch, fake_start_gateway) 47 monkeypatch.setattr(gateway.asyncio, "run", lambda coro: False) 48 49 with pytest.raises(SystemExit) as exc_info: 50 gateway.run_gateway(verbose=1, quiet=True, replace=True) 51 52 assert exc_info.value.code == 1 53 assert calls == [(True, None)] 54 55 56 class TestSystemdLingerStatus: 57 def test_reports_enabled(self, monkeypatch): 58 monkeypatch.setattr(gateway, "is_linux", lambda: True) 59 monkeypatch.setattr(gateway, "is_termux", lambda: False) 60 monkeypatch.setenv("USER", "alice") 61 monkeypatch.setattr( 62 gateway.subprocess, 63 "run", 64 lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="yes\n", stderr=""), 65 ) 66 monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/loginctl") 67 68 assert gateway.get_systemd_linger_status() == (True, "") 69 70 def test_reports_disabled(self, monkeypatch): 71 monkeypatch.setattr(gateway, "is_linux", lambda: True) 72 monkeypatch.setattr(gateway, "is_termux", lambda: False) 73 monkeypatch.setenv("USER", "alice") 74 monkeypatch.setattr( 75 gateway.subprocess, 76 "run", 77 lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="no\n", stderr=""), 78 ) 79 monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/loginctl") 80 81 assert gateway.get_systemd_linger_status() == (False, "") 82 83 def test_reports_termux_as_not_supported(self, monkeypatch): 84 monkeypatch.setattr(gateway, "is_termux", lambda: True) 85 86 assert gateway.get_systemd_linger_status() == (None, "not supported in Termux") 87 88 89 class TestContainerSystemdSupport: 90 def test_supports_systemd_services_in_container_with_user_manager(self, monkeypatch): 91 monkeypatch.setattr(gateway, "is_linux", lambda: True) 92 monkeypatch.setattr(gateway, "is_termux", lambda: False) 93 monkeypatch.setattr(gateway, "is_wsl", lambda: False) 94 monkeypatch.setattr(gateway, "is_container", lambda: True) 95 monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl") 96 monkeypatch.setattr(gateway, "_systemd_operational", lambda system=False: not system) 97 98 assert gateway.supports_systemd_services() is True 99 100 def test_supports_systemd_services_in_container_with_system_manager(self, monkeypatch): 101 monkeypatch.setattr(gateway, "is_linux", lambda: True) 102 monkeypatch.setattr(gateway, "is_termux", lambda: False) 103 monkeypatch.setattr(gateway, "is_wsl", lambda: False) 104 monkeypatch.setattr(gateway, "is_container", lambda: True) 105 monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl") 106 monkeypatch.setattr(gateway, "_systemd_operational", lambda system=False: system) 107 108 assert gateway.supports_systemd_services() is True 109 110 def test_supports_systemd_services_in_container_without_systemd(self, monkeypatch): 111 monkeypatch.setattr(gateway, "is_linux", lambda: True) 112 monkeypatch.setattr(gateway, "is_termux", lambda: False) 113 monkeypatch.setattr(gateway, "is_wsl", lambda: False) 114 monkeypatch.setattr(gateway, "is_container", lambda: True) 115 monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl") 116 monkeypatch.setattr(gateway, "_systemd_operational", lambda system=False: False) 117 118 assert gateway.supports_systemd_services() is False 119 120 121 def test_gateway_install_in_container_with_operational_systemd_uses_systemd(monkeypatch): 122 monkeypatch.setattr(gateway, "supports_systemd_services", lambda: True) 123 monkeypatch.setattr(gateway, "is_wsl", lambda: False) 124 monkeypatch.setattr(gateway, "is_macos", lambda: False) 125 monkeypatch.setattr(gateway, "is_managed", lambda: False) 126 127 calls = [] 128 monkeypatch.setattr( 129 gateway, 130 "systemd_install", 131 lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)), 132 ) 133 134 args = SimpleNamespace( 135 gateway_command="install", 136 force=False, 137 system=False, 138 run_as_user=None, 139 ) 140 gateway.gateway_command(args) 141 142 assert calls == [(False, False, None)] 143 144 145 def test_gateway_start_in_container_with_operational_systemd_uses_systemd(monkeypatch): 146 monkeypatch.setattr(gateway, "supports_systemd_services", lambda: True) 147 monkeypatch.setattr(gateway, "is_wsl", lambda: False) 148 monkeypatch.setattr(gateway, "is_macos", lambda: False) 149 150 calls = [] 151 monkeypatch.setattr(gateway, "systemd_start", lambda system=False: calls.append(system)) 152 153 args = SimpleNamespace(gateway_command="start", system=False, all=False) 154 gateway.gateway_command(args) 155 156 assert calls == [False] 157 158 159 def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys): 160 unit_path = tmp_path / "hermes-gateway.service" 161 unit_path.write_text("[Unit]\n") 162 163 monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path) 164 monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (False, "")) 165 166 def fake_run(cmd, capture_output=False, text=False, check=False, **kwargs): 167 if cmd[:4] == ["systemctl", "--user", "status", gateway.get_service_name()]: 168 return SimpleNamespace(returncode=0, stdout="", stderr="") 169 if cmd[:3] == ["systemctl", "--user", "is-active"]: 170 return SimpleNamespace(returncode=0, stdout="active\n", stderr="") 171 if cmd[:3] == ["systemctl", "--user", "show"]: 172 return SimpleNamespace( 173 returncode=0, 174 stdout="ActiveState=active\nSubState=running\nResult=success\nExecMainStatus=0\n", 175 stderr="", 176 ) 177 raise AssertionError(f"Unexpected command: {cmd}") 178 179 monkeypatch.setattr(gateway.subprocess, "run", fake_run) 180 181 gateway.systemd_status(deep=False) 182 183 out = capsys.readouterr().out 184 assert "gateway service is running" in out 185 assert "Systemd linger is disabled" in out 186 assert "loginctl enable-linger" in out 187 188 189 def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys): 190 unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service" 191 192 monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path) 193 194 calls = [] 195 helper_calls = [] 196 197 def fake_run(cmd, check=False, **kwargs): 198 calls.append((cmd, check)) 199 return SimpleNamespace(returncode=0, stdout="", stderr="") 200 201 monkeypatch.setattr(gateway.subprocess, "run", fake_run) 202 monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True)) 203 204 gateway.systemd_install(force=False) 205 206 out = capsys.readouterr().out 207 assert unit_path.exists() 208 assert [cmd for cmd, _ in calls] == [ 209 ["systemctl", "--user", "daemon-reload"], 210 ["systemctl", "--user", "enable", gateway.get_service_name()], 211 ] 212 assert helper_calls == [True] 213 assert "User service installed and enabled" in out 214 215 216 def test_systemd_install_system_scope_skips_linger_and_uses_systemctl(monkeypatch, tmp_path, capsys): 217 unit_path = tmp_path / "etc" / "systemd" / "system" / "hermes-gateway.service" 218 219 monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path) 220 monkeypatch.setattr( 221 gateway, 222 "generate_systemd_unit", 223 lambda system=False, run_as_user=None: f"scope={system} user={run_as_user}\n", 224 ) 225 monkeypatch.setattr(gateway, "_require_root_for_system_service", lambda action: None) 226 227 calls = [] 228 helper_calls = [] 229 230 def fake_run(cmd, check=False, **kwargs): 231 calls.append((cmd, check)) 232 return SimpleNamespace(returncode=0, stdout="", stderr="") 233 234 monkeypatch.setattr(gateway.subprocess, "run", fake_run) 235 monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True)) 236 237 gateway.systemd_install(force=False, system=True, run_as_user="alice") 238 239 out = capsys.readouterr().out 240 assert unit_path.exists() 241 assert unit_path.read_text(encoding="utf-8") == "scope=True user=alice\n" 242 assert [cmd for cmd, _ in calls] == [ 243 ["systemctl", "daemon-reload"], 244 ["systemctl", "enable", gateway.get_service_name()], 245 ] 246 assert helper_calls == [] 247 assert "Configured to run as: alice" not in out # generated test unit has no User= line 248 assert "System service installed and enabled" in out 249 250 251 def test_conflicting_systemd_units_warning(monkeypatch, tmp_path, capsys): 252 user_unit = tmp_path / "user" / "hermes-gateway.service" 253 system_unit = tmp_path / "system" / "hermes-gateway.service" 254 user_unit.parent.mkdir(parents=True) 255 system_unit.parent.mkdir(parents=True) 256 user_unit.write_text("[Unit]\n", encoding="utf-8") 257 system_unit.write_text("[Unit]\n", encoding="utf-8") 258 259 monkeypatch.setattr( 260 gateway, 261 "get_systemd_unit_path", 262 lambda system=False: system_unit if system else user_unit, 263 ) 264 265 gateway.print_systemd_scope_conflict_warning() 266 267 out = capsys.readouterr().out 268 assert "Both user and system gateway services are installed" in out 269 assert "hermes gateway uninstall" in out 270 assert "--system" in out 271 272 273 def test_install_linux_gateway_from_setup_system_choice_without_root_prints_followup(monkeypatch, capsys): 274 monkeypatch.setattr(gateway, "prompt_linux_gateway_install_scope", lambda: "system") 275 monkeypatch.setattr(gateway.os, "geteuid", lambda: 1000) 276 monkeypatch.setattr(gateway, "_default_system_service_user", lambda: "alice") 277 monkeypatch.setattr(gateway, "systemd_install", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not install"))) 278 279 scope, did_install = gateway.install_linux_gateway_from_setup(force=False) 280 281 out = capsys.readouterr().out 282 assert (scope, did_install) == ("system", False) 283 assert "sudo hermes gateway install --system --run-as-user alice" in out 284 assert "sudo hermes gateway start --system" in out 285 286 287 def test_install_linux_gateway_from_setup_system_choice_as_root_installs(monkeypatch): 288 monkeypatch.setattr(gateway, "prompt_linux_gateway_install_scope", lambda: "system") 289 monkeypatch.setattr(gateway.os, "geteuid", lambda: 0) 290 monkeypatch.setattr(gateway, "_default_system_service_user", lambda: "alice") 291 292 calls = [] 293 monkeypatch.setattr( 294 gateway, 295 "systemd_install", 296 lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)), 297 ) 298 299 scope, did_install = gateway.install_linux_gateway_from_setup(force=True) 300 301 assert (scope, did_install) == ("system", True) 302 assert calls == [(True, True, "alice")] 303 304 305 def test_find_gateway_pids_falls_back_to_pid_file_when_process_scan_fails(monkeypatch): 306 monkeypatch.setattr(gateway, "_get_service_pids", lambda: set()) 307 monkeypatch.setattr(gateway, "is_windows", lambda: False) 308 monkeypatch.setattr("gateway.status.get_running_pid", lambda: 321) 309 310 def fake_run(cmd, **kwargs): 311 if cmd[:4] == ["ps", "-A", "eww", "-o"]: 312 return SimpleNamespace(returncode=1, stdout="", stderr="ps failed") 313 if cmd[:3] == ["ps", "-o", "ppid="]: 314 # _get_ancestor_pids() walks up the tree; return "no parent" so 315 # the loop terminates cleanly. 316 return SimpleNamespace(returncode=1, stdout="", stderr="") 317 raise AssertionError(f"Unexpected command: {cmd}") 318 319 monkeypatch.setattr(gateway.subprocess, "run", fake_run) 320 321 assert gateway.find_gateway_pids() == [321] 322 323 324 # --------------------------------------------------------------------------- 325 # _wait_for_gateway_exit 326 # --------------------------------------------------------------------------- 327 328 329 class TestWaitForGatewayExit: 330 """PID-based wait with force-kill on timeout.""" 331 332 def test_returns_immediately_when_no_pid(self, monkeypatch): 333 """If get_running_pid returns None, exit instantly.""" 334 monkeypatch.setattr("gateway.status.get_running_pid", lambda: None) 335 # Should return without sleeping at all. 336 gateway._wait_for_gateway_exit(timeout=1.0, force_after=0.5) 337 338 def test_returns_when_process_exits_gracefully(self, monkeypatch): 339 """Process exits after a couple of polls — no SIGKILL needed.""" 340 poll_count = 0 341 342 def mock_get_running_pid(): 343 nonlocal poll_count 344 poll_count += 1 345 return 12345 if poll_count <= 2 else None 346 347 monkeypatch.setattr("gateway.status.get_running_pid", mock_get_running_pid) 348 monkeypatch.setattr("time.sleep", lambda _: None) 349 350 gateway._wait_for_gateway_exit(timeout=10.0, force_after=999.0) 351 # Should have polled until None was returned. 352 assert poll_count == 3 353 354 def test_force_kills_after_grace_period(self, monkeypatch): 355 """When the process doesn't exit, force-kill the saved PID.""" 356 357 # Simulate monotonic time advancing past force_after 358 call_num = 0 359 def fake_monotonic(): 360 nonlocal call_num 361 call_num += 1 362 # First two calls: initial deadline + force_deadline setup (time 0) 363 # Then each loop iteration advances time 364 return call_num * 2.0 # 2, 4, 6, 8, ... 365 366 kills = [] 367 def mock_terminate(pid, force=False): 368 kills.append((pid, force)) 369 370 # get_running_pid returns the PID until kill is sent, then None 371 def mock_get_running_pid(): 372 return None if kills else 42 373 374 monkeypatch.setattr("time.monotonic", fake_monotonic) 375 monkeypatch.setattr("time.sleep", lambda _: None) 376 monkeypatch.setattr("gateway.status.get_running_pid", mock_get_running_pid) 377 monkeypatch.setattr(gateway, "terminate_pid", mock_terminate) 378 379 gateway._wait_for_gateway_exit(timeout=10.0, force_after=5.0) 380 assert (42, True) in kills 381 382 def test_handles_process_already_gone_on_kill(self, monkeypatch): 383 """ProcessLookupError during force-kill is not fatal.""" 384 385 call_num = 0 386 def fake_monotonic(): 387 nonlocal call_num 388 call_num += 1 389 return call_num * 3.0 # Jump past force_after quickly 390 391 def mock_terminate(pid, force=False): 392 raise ProcessLookupError 393 394 monkeypatch.setattr("time.monotonic", fake_monotonic) 395 monkeypatch.setattr("time.sleep", lambda _: None) 396 monkeypatch.setattr("gateway.status.get_running_pid", lambda: 99) 397 monkeypatch.setattr(gateway, "terminate_pid", mock_terminate) 398 399 # Should not raise — ProcessLookupError means it's already gone. 400 gateway._wait_for_gateway_exit(timeout=10.0, force_after=2.0) 401 402 def test_kill_gateway_processes_force_uses_helper(self, monkeypatch): 403 calls = [] 404 405 monkeypatch.setattr(gateway, "find_gateway_pids", lambda exclude_pids=None, all_profiles=False: [11, 22]) 406 monkeypatch.setattr(gateway, "terminate_pid", lambda pid, force=False: calls.append((pid, force))) 407 408 killed = gateway.kill_gateway_processes(force=True) 409 410 assert killed == 2 411 assert calls == [(11, True), (22, True)] 412 413 414 class TestStopProfileGateway: 415 def test_stop_profile_gateway_keeps_pid_file_when_process_still_running(self, monkeypatch): 416 calls = {"kill": 0, "remove": 0} 417 418 monkeypatch.setattr("gateway.status.get_running_pid", lambda: 12345) 419 monkeypatch.setattr( 420 gateway.os, 421 "kill", 422 lambda pid, sig: calls.__setitem__("kill", calls["kill"] + 1), 423 ) 424 monkeypatch.setattr("time.sleep", lambda _: None) 425 monkeypatch.setattr( 426 "gateway.status.remove_pid_file", 427 lambda: calls.__setitem__("remove", calls["remove"] + 1), 428 ) 429 430 assert gateway.stop_profile_gateway() is True 431 assert calls["kill"] == 21 432 assert calls["remove"] == 0