1  """Tests for gateway service management helpers."""
   2  
   3  import os
   4  import pwd
   5  from pathlib import Path
   6  from types import SimpleNamespace
   7  
   8  import pytest
   9  
  10  import hermes_cli.gateway as gateway_cli
  11  from gateway.restart import (
  12      DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
  13      GATEWAY_SERVICE_RESTART_EXIT_CODE,
  14  )
  15  
  16  
  17  class TestUserSystemdPrivateSocketPreflight:
  18      def test_preflight_accepts_private_socket_without_dbus_bus(self, monkeypatch):
  19          monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: None)
  20          monkeypatch.setattr(gateway_cli, "_user_dbus_socket_path", lambda: Path("/tmp/missing-bus"))
  21          monkeypatch.setattr(gateway_cli, "_user_systemd_private_socket_path", lambda: Path("/tmp/private-socket"))
  22          monkeypatch.setattr(Path, "exists", lambda self: str(self) == "/tmp/private-socket")
  23  
  24          gateway_cli._preflight_user_systemd(auto_enable_linger=False)
  25  
  26      def test_wait_for_user_dbus_socket_accepts_private_socket(self, monkeypatch):
  27          calls = []
  28          monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("env"))
  29          monkeypatch.setattr(gateway_cli, "_user_dbus_socket_path", lambda: Path("/tmp/missing-bus"))
  30          monkeypatch.setattr(gateway_cli, "_user_systemd_private_socket_path", lambda: Path("/tmp/private-socket"))
  31          monkeypatch.setattr(Path, "exists", lambda self: str(self) == "/tmp/private-socket")
  32  
  33          assert gateway_cli._wait_for_user_dbus_socket(timeout=0.1) is True
  34          assert calls == ["env"]
  35  
  36  
  37  class TestSystemdServiceRefresh:
  38      def test_systemd_install_repairs_outdated_unit_without_force(self, tmp_path, monkeypatch):
  39          unit_path = tmp_path / "hermes-gateway.service"
  40          unit_path.write_text("old unit\n", encoding="utf-8")
  41  
  42          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
  43          monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
  44  
  45          calls = []
  46  
  47          def fake_run(cmd, check=True, **kwargs):
  48              calls.append(cmd)
  49              return SimpleNamespace(returncode=0, stdout="", stderr="")
  50  
  51          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
  52  
  53          gateway_cli.systemd_install()
  54  
  55          assert unit_path.read_text(encoding="utf-8") == "new unit\n"
  56          assert calls[:2] == [
  57              ["systemctl", "--user", "daemon-reload"],
  58              ["systemctl", "--user", "enable", gateway_cli.get_service_name()],
  59          ]
  60  
  61      def test_systemd_start_refreshes_outdated_unit(self, tmp_path, monkeypatch):
  62          unit_path = tmp_path / "hermes-gateway.service"
  63          unit_path.write_text("old unit\n", encoding="utf-8")
  64  
  65          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
  66          monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
  67  
  68          calls = []
  69  
  70          def fake_run(cmd, check=True, **kwargs):
  71              calls.append(cmd)
  72              return SimpleNamespace(returncode=0, stdout="", stderr="")
  73  
  74          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
  75  
  76          gateway_cli.systemd_start()
  77  
  78          assert unit_path.read_text(encoding="utf-8") == "new unit\n"
  79          assert calls[:2] == [
  80              ["systemctl", "--user", "daemon-reload"],
  81              ["systemctl", "--user", "start", gateway_cli.get_service_name()],
  82          ]
  83  
  84      def test_systemd_restart_refreshes_outdated_unit(self, tmp_path, monkeypatch):
  85          unit_path = tmp_path / "hermes-gateway.service"
  86          unit_path.write_text("old unit\n", encoding="utf-8")
  87  
  88          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
  89          monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
  90  
  91          calls = []
  92  
  93          def fake_run(cmd, check=True, **kwargs):
  94              calls.append(cmd)
  95              return SimpleNamespace(returncode=0, stdout="", stderr="")
  96  
  97          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
  98  
  99          gateway_cli.systemd_restart()
 100  
 101          assert unit_path.read_text(encoding="utf-8") == "new unit\n"
 102          assert calls[:4] == [
 103              ["systemctl", "--user", "daemon-reload"],
 104              ["systemctl", "--user", "show", gateway_cli.get_service_name(), "--no-pager", "--property", "ActiveState,SubState,Result,ExecMainStatus"],
 105              ["systemctl", "--user", "reset-failed", gateway_cli.get_service_name()],
 106              ["systemctl", "--user", "reload-or-restart", gateway_cli.get_service_name()],
 107          ]
 108  
 109  
 110      def test_run_gateway_refreshes_outdated_unit_on_boot(self, tmp_path, monkeypatch):
 111          """run_gateway() should refresh the systemd unit on boot so that
 112          restart settings take effect even when the process was respawned
 113          via exit-code-75 (bypassing `hermes gateway restart`)."""
 114          unit_path = tmp_path / "hermes-gateway.service"
 115          unit_path.write_text("old unit\n", encoding="utf-8")
 116  
 117          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
 118          monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
 119          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 120  
 121          calls = []
 122  
 123          def fake_run(cmd, check=True, **kwargs):
 124              calls.append(cmd)
 125              return SimpleNamespace(returncode=0, stdout="", stderr="")
 126  
 127          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 128  
 129          # Prevent run_gateway from actually starting the gateway
 130          async def fake_start_gateway(**kwargs):
 131              return True
 132  
 133          monkeypatch.setattr("gateway.run.start_gateway", fake_start_gateway)
 134  
 135          gateway_cli.run_gateway()
 136  
 137          assert unit_path.read_text(encoding="utf-8") == "new unit\n"
 138          assert ["systemctl", "--user", "daemon-reload"] in calls
 139  
 140  
 141  class TestRequireServiceInstalled:
 142      def test_exits_with_install_hint_when_unit_missing(self, tmp_path, monkeypatch, capsys):
 143          unit_path = tmp_path / "hermes-gateway.service"
 144          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
 145  
 146          with pytest.raises(SystemExit) as exc_info:
 147              gateway_cli._require_service_installed("start")
 148  
 149          assert exc_info.value.code == 1
 150          out = capsys.readouterr().out
 151          assert "not installed" in out
 152          assert "hermes gateway install" in out
 153  
 154      def test_passes_when_unit_exists(self, tmp_path, monkeypatch):
 155          unit_path = tmp_path / "hermes-gateway.service"
 156          unit_path.write_text("[Unit]\n", encoding="utf-8")
 157          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
 158  
 159          gateway_cli._require_service_installed("start")
 160  
 161  
 162  class TestGeneratedSystemdUnits:
 163      def test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self, monkeypatch):
 164          monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 60)
 165          unit = gateway_cli.generate_systemd_unit(system=False)
 166  
 167          assert "ExecStart=" in unit
 168          assert "ExecStop=" not in unit
 169          assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
 170          assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
 171          # TimeoutStopSec must exceed the default drain_timeout (60s) so
 172          # systemd doesn't SIGKILL the cgroup before post-interrupt cleanup
 173          # (tool subprocess kill, adapter disconnect) runs — issue #8202.
 174          assert "TimeoutStopSec=90" in unit
 175  
 176      def test_user_unit_includes_resolved_node_directory_in_path(self, monkeypatch):
 177          monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: "/home/test/.nvm/versions/node/v24.14.0/bin/node" if cmd == "node" else None)
 178  
 179          unit = gateway_cli.generate_systemd_unit(system=False)
 180  
 181          assert "/home/test/.nvm/versions/node/v24.14.0/bin" in unit
 182  
 183      def test_user_unit_includes_resolved_bun_directory_in_path(self, monkeypatch):
 184          def _which(cmd):
 185              if cmd == "node":
 186                  return None
 187              if cmd == "bun":
 188                  return "/home/test/.bun/bin/bun"
 189              return None
 190  
 191          monkeypatch.setattr(gateway_cli.shutil, "which", _which)
 192  
 193          unit = gateway_cli.generate_systemd_unit(system=False)
 194  
 195          assert "/home/test/.bun/bin" in unit
 196  
 197      def test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self, monkeypatch):
 198          monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 60)
 199          unit = gateway_cli.generate_systemd_unit(system=True)
 200  
 201          assert "ExecStart=" in unit
 202          assert "ExecStop=" not in unit
 203          assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
 204          assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
 205          # TimeoutStopSec must exceed the default drain_timeout (60s) so
 206          # systemd doesn't SIGKILL the cgroup before post-interrupt cleanup
 207          # (tool subprocess kill, adapter disconnect) runs — issue #8202.
 208          assert "TimeoutStopSec=90" in unit
 209          assert "WantedBy=multi-user.target" in unit
 210  
 211  
 212  class TestGatewayStopCleanup:
 213      def test_stop_only_kills_current_profile_by_default(self, tmp_path, monkeypatch):
 214          """Without --all, stop uses systemd (if available) and does NOT call
 215          the global kill_gateway_processes()."""
 216          unit_path = tmp_path / "hermes-gateway.service"
 217          unit_path.write_text("unit\n", encoding="utf-8")
 218  
 219          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 220          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 221          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 222          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
 223  
 224          service_calls = []
 225          kill_calls = []
 226  
 227          monkeypatch.setattr(gateway_cli, "systemd_stop", lambda system=False: service_calls.append("stop"))
 228          monkeypatch.setattr(
 229              gateway_cli,
 230              "kill_gateway_processes",
 231              lambda force=False, all_profiles=False: kill_calls.append(force) or 2,
 232          )
 233  
 234          gateway_cli.gateway_command(SimpleNamespace(gateway_command="stop"))
 235  
 236          assert service_calls == ["stop"]
 237          # Global kill should NOT be called without --all
 238          assert kill_calls == []
 239  
 240      def test_stop_all_sweeps_all_gateway_processes(self, tmp_path, monkeypatch):
 241          """With --all, stop uses systemd AND calls the global kill_gateway_processes()."""
 242          unit_path = tmp_path / "hermes-gateway.service"
 243          unit_path.write_text("unit\n", encoding="utf-8")
 244  
 245          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 246          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 247          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 248          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
 249  
 250          service_calls = []
 251          kill_calls = []
 252  
 253          monkeypatch.setattr(gateway_cli, "systemd_stop", lambda system=False: service_calls.append("stop"))
 254          monkeypatch.setattr(
 255              gateway_cli,
 256              "kill_gateway_processes",
 257              lambda force=False, all_profiles=False: kill_calls.append(force) or 2,
 258          )
 259  
 260          gateway_cli.gateway_command(SimpleNamespace(gateway_command="stop", **{"all": True}))
 261  
 262          assert service_calls == ["stop"]
 263          assert kill_calls == [False]
 264  
 265  
 266  class TestLaunchdServiceRecovery:
 267      def test_get_restart_drain_timeout_prefers_env_then_config_then_default(self, monkeypatch):
 268          monkeypatch.delenv("HERMES_RESTART_DRAIN_TIMEOUT", raising=False)
 269          monkeypatch.setattr(gateway_cli, "read_raw_config", lambda: {})
 270  
 271          assert (
 272              gateway_cli._get_restart_drain_timeout()
 273              == DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
 274          )
 275  
 276          monkeypatch.setattr(
 277              gateway_cli,
 278              "read_raw_config",
 279              lambda: {"agent": {"restart_drain_timeout": 14}},
 280          )
 281          assert gateway_cli._get_restart_drain_timeout() == 14.0
 282  
 283          monkeypatch.setenv("HERMES_RESTART_DRAIN_TIMEOUT", "9")
 284          assert gateway_cli._get_restart_drain_timeout() == 9.0
 285  
 286          monkeypatch.setenv("HERMES_RESTART_DRAIN_TIMEOUT", "invalid")
 287          assert (
 288              gateway_cli._get_restart_drain_timeout()
 289              == DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
 290          )
 291  
 292      def test_launchd_install_repairs_outdated_plist_without_force(self, tmp_path, monkeypatch):
 293          plist_path = tmp_path / "ai.hermes.gateway.plist"
 294          plist_path.write_text("<plist>old content</plist>", encoding="utf-8")
 295  
 296          monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
 297  
 298          calls = []
 299  
 300          def fake_run(cmd, check=False, **kwargs):
 301              calls.append(cmd)
 302              return SimpleNamespace(returncode=0, stdout="", stderr="")
 303  
 304          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 305  
 306          gateway_cli.launchd_install()
 307  
 308          label = gateway_cli.get_launchd_label()
 309          domain = gateway_cli._launchd_domain()
 310          assert "--replace" in plist_path.read_text(encoding="utf-8")
 311          assert calls[:2] == [
 312              ["launchctl", "bootout", f"{domain}/{label}"],
 313              ["launchctl", "bootstrap", domain, str(plist_path)],
 314          ]
 315  
 316      def test_launchd_start_reloads_unloaded_job_and_retries(self, tmp_path, monkeypatch):
 317          plist_path = tmp_path / "ai.hermes.gateway.plist"
 318          plist_path.write_text(gateway_cli.generate_launchd_plist(), encoding="utf-8")
 319          label = gateway_cli.get_launchd_label()
 320  
 321          calls = []
 322          domain = gateway_cli._launchd_domain()
 323          target = f"{domain}/{label}"
 324  
 325          def fake_run(cmd, check=False, **kwargs):
 326              if cmd and cmd[0] == "launchctl":
 327                  calls.append(cmd)
 328              if cmd == ["launchctl", "kickstart", target] and calls.count(cmd) == 1:
 329                  raise gateway_cli.subprocess.CalledProcessError(3, cmd, stderr="Could not find service")
 330              return SimpleNamespace(returncode=0, stdout="", stderr="")
 331  
 332          monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
 333          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 334  
 335          gateway_cli.launchd_start()
 336  
 337          assert calls == [
 338              ["launchctl", "kickstart", target],
 339              ["launchctl", "bootstrap", domain, str(plist_path)],
 340              ["launchctl", "kickstart", target],
 341          ]
 342  
 343      def test_launchd_start_reloads_on_kickstart_exit_code_113(self, tmp_path, monkeypatch):
 344          """Exit code 113 (\"Could not find service\") should also trigger bootstrap recovery."""
 345          plist_path = tmp_path / "ai.hermes.gateway.plist"
 346          plist_path.write_text(gateway_cli.generate_launchd_plist(), encoding="utf-8")
 347          label = gateway_cli.get_launchd_label()
 348  
 349          calls = []
 350          domain = gateway_cli._launchd_domain()
 351          target = f"{domain}/{label}"
 352  
 353          def fake_run(cmd, check=False, **kwargs):
 354              if cmd and cmd[0] == "launchctl":
 355                  calls.append(cmd)
 356              if cmd == ["launchctl", "kickstart", target] and calls.count(cmd) == 1:
 357                  raise gateway_cli.subprocess.CalledProcessError(113, cmd, stderr="Could not find service")
 358              return SimpleNamespace(returncode=0, stdout="", stderr="")
 359  
 360          monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
 361          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 362  
 363          gateway_cli.launchd_start()
 364  
 365          assert calls == [
 366              ["launchctl", "kickstart", target],
 367              ["launchctl", "bootstrap", domain, str(plist_path)],
 368              ["launchctl", "kickstart", target],
 369          ]
 370  
 371      def test_launchd_restart_drains_running_gateway_before_kickstart(self, monkeypatch):
 372          calls = []
 373          target = f"{gateway_cli._launchd_domain()}/{gateway_cli.get_launchd_label()}"
 374  
 375          monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 12.0)
 376          monkeypatch.setattr(gateway_cli, "_request_gateway_self_restart", lambda pid: False)
 377          monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", lambda timeout, force_after=None: True)
 378          monkeypatch.setattr(gateway_cli, "terminate_pid", lambda pid, force=False: calls.append(("term", pid, force)))
 379          monkeypatch.setattr(
 380              "gateway.status.get_running_pid",
 381              lambda: 321,
 382          )
 383  
 384          def fake_run(cmd, check=False, **kwargs):
 385              calls.append(cmd)
 386              return SimpleNamespace(returncode=0, stdout="", stderr="")
 387  
 388          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 389  
 390          gateway_cli.launchd_restart()
 391  
 392          assert calls == [
 393              ("term", 321, False),
 394              ["launchctl", "kickstart", "-k", target],
 395          ]
 396  
 397      def test_launchd_restart_self_requests_graceful_restart_without_kickstart(self, monkeypatch, capsys):
 398          calls = []
 399  
 400          monkeypatch.setattr(
 401              "gateway.status.get_running_pid",
 402              lambda: 321,
 403          )
 404          monkeypatch.setattr(
 405              gateway_cli,
 406              "_request_gateway_self_restart",
 407              lambda pid: calls.append(("self", pid)) or True,
 408          )
 409          monkeypatch.setattr(
 410              gateway_cli.subprocess,
 411              "run",
 412              lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("launchctl should not run")),
 413          )
 414  
 415          gateway_cli.launchd_restart()
 416  
 417          assert calls == [("self", 321)]
 418          assert "restart requested" in capsys.readouterr().out.lower()
 419  
 420      def test_launchd_stop_uses_bootout_not_kill(self, monkeypatch):
 421          """launchd_stop must bootout the service so KeepAlive doesn't respawn it."""
 422          label = gateway_cli.get_launchd_label()
 423          domain = gateway_cli._launchd_domain()
 424          target = f"{domain}/{label}"
 425  
 426          calls = []
 427  
 428          def fake_run(cmd, check=False, **kwargs):
 429              calls.append(cmd)
 430              return SimpleNamespace(returncode=0, stdout="", stderr="")
 431  
 432          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 433          monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", lambda **kw: None)
 434  
 435          gateway_cli.launchd_stop()
 436  
 437          assert calls == [["launchctl", "bootout", target]]
 438  
 439      def test_launchd_stop_tolerates_already_unloaded(self, monkeypatch, capsys):
 440          """launchd_stop silently handles exit codes 3/113 (job not loaded)."""
 441          label = gateway_cli.get_launchd_label()
 442          domain = gateway_cli._launchd_domain()
 443          target = f"{domain}/{label}"
 444  
 445          def fake_run(cmd, check=False, **kwargs):
 446              if "bootout" in cmd:
 447                  raise gateway_cli.subprocess.CalledProcessError(3, cmd, stderr="Could not find service")
 448              return SimpleNamespace(returncode=0, stdout="", stderr="")
 449  
 450          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 451          monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", lambda **kw: None)
 452  
 453          # Should not raise — exit code 3 means already unloaded
 454          gateway_cli.launchd_stop()
 455  
 456          output = capsys.readouterr().out
 457          assert "stopped" in output.lower()
 458  
 459      def test_launchd_stop_waits_for_process_exit(self, monkeypatch):
 460          """launchd_stop calls _wait_for_gateway_exit after bootout."""
 461          wait_called = []
 462  
 463          def fake_run(cmd, check=False, **kwargs):
 464              return SimpleNamespace(returncode=0, stdout="", stderr="")
 465  
 466          def fake_wait(**kwargs):
 467              wait_called.append(kwargs)
 468  
 469          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 470          monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", fake_wait)
 471  
 472          gateway_cli.launchd_stop()
 473  
 474          assert len(wait_called) == 1
 475          assert wait_called[0] == {"timeout": 10.0, "force_after": 5.0}
 476  
 477      def test_launchd_status_reports_local_stale_plist_when_unloaded(self, tmp_path, monkeypatch, capsys):
 478          plist_path = tmp_path / "ai.hermes.gateway.plist"
 479          plist_path.write_text("<plist>old content</plist>", encoding="utf-8")
 480  
 481          monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
 482          monkeypatch.setattr(
 483              gateway_cli.subprocess,
 484              "run",
 485              lambda *args, **kwargs: SimpleNamespace(returncode=113, stdout="", stderr="Could not find service"),
 486          )
 487  
 488          gateway_cli.launchd_status()
 489  
 490          output = capsys.readouterr().out
 491          assert str(plist_path) in output
 492          assert "stale" in output.lower()
 493          assert "not loaded" in output.lower()
 494  
 495  
 496  class TestGatewayServiceDetection:
 497      def test_supports_systemd_services_requires_systemctl_binary(self, monkeypatch):
 498          monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
 499          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 500          monkeypatch.setattr(gateway_cli.shutil, "which", lambda name: None)
 501  
 502          assert gateway_cli.supports_systemd_services() is False
 503  
 504      def test_supports_systemd_services_returns_true_when_systemctl_present(self, monkeypatch):
 505          monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
 506          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 507          monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
 508          monkeypatch.setattr(gateway_cli, "is_container", lambda: False)
 509          monkeypatch.setattr(gateway_cli.shutil, "which", lambda name: "/usr/bin/systemctl")
 510  
 511          assert gateway_cli.supports_systemd_services() is True
 512  
 513      def test_is_service_running_checks_system_scope_when_user_scope_is_inactive(self, monkeypatch):
 514          user_unit = SimpleNamespace(exists=lambda: True)
 515          system_unit = SimpleNamespace(exists=lambda: True)
 516  
 517          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 518          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 519          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 520          monkeypatch.setattr(
 521              gateway_cli,
 522              "get_systemd_unit_path",
 523              lambda system=False: system_unit if system else user_unit,
 524          )
 525  
 526          def fake_run(cmd, capture_output=True, text=True, **kwargs):
 527              if cmd == ["systemctl", "--user", "is-active", gateway_cli.get_service_name()]:
 528                  return SimpleNamespace(returncode=0, stdout="inactive\n", stderr="")
 529              if cmd == ["systemctl", "is-active", gateway_cli.get_service_name()]:
 530                  return SimpleNamespace(returncode=0, stdout="active\n", stderr="")
 531              raise AssertionError(f"Unexpected command: {cmd}")
 532  
 533          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 534  
 535          assert gateway_cli._is_service_running() is True
 536  
 537      def test_is_service_running_returns_false_when_systemctl_missing(self, monkeypatch):
 538          unit = SimpleNamespace(exists=lambda: True)
 539  
 540          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 541          monkeypatch.setattr(
 542              gateway_cli,
 543              "get_systemd_unit_path",
 544              lambda system=False: unit,
 545          )
 546  
 547          def fake_run(*args, **kwargs):
 548              raise FileNotFoundError("systemctl")
 549  
 550          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
 551  
 552          assert gateway_cli._is_service_running() is False
 553  
 554  class TestGatewaySystemServiceRouting:
 555      def test_systemd_restart_self_requests_graceful_restart_and_waits(self, monkeypatch, capsys):
 556          calls = []
 557  
 558          monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
 559          monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
 560          monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: calls.append(("refresh", system)))
 561          monkeypatch.setattr(
 562              "gateway.status.get_running_pid",
 563              lambda: 654,
 564          )
 565          monkeypatch.setattr(
 566              gateway_cli,
 567              "_request_gateway_self_restart",
 568              lambda pid: calls.append(("self", pid)) or True,
 569          )
 570  
 571          # Simulate: old process dies immediately, new process becomes active
 572          kill_call_count = [0]
 573          def fake_kill(pid, sig):
 574              kill_call_count[0] += 1
 575              if kill_call_count[0] >= 2:  # first call checks, second = dead
 576                  raise ProcessLookupError()
 577          monkeypatch.setattr(os, "kill", fake_kill)
 578  
 579          # Simulate systemctl reset-failed/start followed by an active unit
 580          new_pid = [None]
 581          def fake_subprocess_run(cmd, **kwargs):
 582              if "reset-failed" in cmd:
 583                  calls.append(("reset-failed", cmd))
 584                  return SimpleNamespace(stdout="", returncode=0)
 585              if "start" in cmd:
 586                  calls.append(("start", cmd))
 587                  return SimpleNamespace(stdout="", returncode=0)
 588              if "show" in cmd:
 589                  new_pid[0] = 999
 590                  return SimpleNamespace(
 591                      stdout="ActiveState=active\nSubState=running\nResult=success\nExecMainStatus=0\n",
 592                      returncode=0,
 593                  )
 594              raise AssertionError(f"Unexpected systemctl call: {cmd}")
 595  
 596          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_subprocess_run)
 597          # get_running_pid returns new PID after restart
 598          pid_calls = [0]
 599          def fake_get_pid():
 600              pid_calls[0] += 1
 601              return 999 if pid_calls[0] > 1 else 654
 602          monkeypatch.setattr("gateway.status.get_running_pid", fake_get_pid)
 603  
 604          gateway_cli.systemd_restart()
 605  
 606          assert ("self", 654) in calls
 607          assert any(call[0] == "reset-failed" for call in calls)
 608          assert any(call[0] == "start" for call in calls)
 609          out = capsys.readouterr().out.lower()
 610          assert "restarted" in out
 611  
 612      def test_systemd_restart_recovers_failed_planned_restart(self, monkeypatch, capsys):
 613          monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
 614          monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
 615          monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
 616          monkeypatch.setattr(
 617              "gateway.status.read_runtime_status",
 618              lambda: {"restart_requested": True, "gateway_state": "stopped"},
 619          )
 620          monkeypatch.setattr(gateway_cli, "_request_gateway_self_restart", lambda pid: False)
 621  
 622          calls = []
 623          started = {"value": False}
 624  
 625          def fake_subprocess_run(cmd, **kwargs):
 626              if "show" in cmd:
 627                  if not started["value"]:
 628                      return SimpleNamespace(
 629                          stdout=(
 630                              "ActiveState=failed\n"
 631                              "SubState=failed\n"
 632                              "Result=exit-code\n"
 633                              f"ExecMainStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}\n"
 634                          ),
 635                          returncode=0,
 636                      )
 637                  return SimpleNamespace(
 638                      stdout="ActiveState=active\nSubState=running\nResult=success\nExecMainStatus=0\n",
 639                      returncode=0,
 640                  )
 641              if "reset-failed" in cmd:
 642                  calls.append(("reset-failed", cmd))
 643                  return SimpleNamespace(stdout="", returncode=0)
 644              if "start" in cmd:
 645                  started["value"] = True
 646                  calls.append(("start", cmd))
 647                  return SimpleNamespace(stdout="", returncode=0)
 648              raise AssertionError(f"Unexpected command: {cmd}")
 649  
 650          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_subprocess_run)
 651          monkeypatch.setattr(
 652              "gateway.status.get_running_pid",
 653              lambda: 999 if started["value"] else None,
 654          )
 655  
 656          gateway_cli.systemd_restart()
 657  
 658          assert any(call[0] == "reset-failed" for call in calls)
 659          assert any(call[0] == "start" for call in calls)
 660          out = capsys.readouterr().out.lower()
 661          assert "restarted" in out
 662  
 663      def test_systemd_status_surfaces_planned_restart_failure(self, monkeypatch, capsys):
 664          unit = SimpleNamespace(exists=lambda: True)
 665          monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
 666          monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit)
 667          monkeypatch.setattr(gateway_cli, "has_conflicting_systemd_units", lambda: False)
 668          monkeypatch.setattr(gateway_cli, "has_legacy_hermes_units", lambda: False)
 669          monkeypatch.setattr(gateway_cli, "systemd_unit_is_current", lambda system=False: True)
 670          monkeypatch.setattr(gateway_cli, "_runtime_health_lines", lambda: ["⚠ Last shutdown reason: Gateway restart requested"])
 671          monkeypatch.setattr(gateway_cli, "get_systemd_linger_status", lambda: (True, ""))
 672          monkeypatch.setattr(gateway_cli, "_read_systemd_unit_properties", lambda system=False: {
 673              "ActiveState": "failed",
 674              "SubState": "failed",
 675              "Result": "exit-code",
 676              "ExecMainStatus": str(GATEWAY_SERVICE_RESTART_EXIT_CODE),
 677          })
 678  
 679          calls = []
 680  
 681          def fake_run_systemctl(args, **kwargs):
 682              calls.append(args)
 683              if args[:2] == ["status", gateway_cli.get_service_name()]:
 684                  return SimpleNamespace(returncode=0, stdout="", stderr="")
 685              if args[:2] == ["is-active", gateway_cli.get_service_name()]:
 686                  return SimpleNamespace(returncode=3, stdout="failed\n", stderr="")
 687              raise AssertionError(f"Unexpected args: {args}")
 688  
 689          monkeypatch.setattr(gateway_cli, "_run_systemctl", fake_run_systemctl)
 690  
 691          gateway_cli.systemd_status()
 692  
 693          out = capsys.readouterr().out
 694          assert "Planned restart is stuck in systemd failed state" in out
 695  
 696      def test_gateway_status_dispatches_full_flag(self, monkeypatch):
 697          user_unit = SimpleNamespace(exists=lambda: True)
 698          system_unit = SimpleNamespace(exists=lambda: False)
 699  
 700          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 701          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 702          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 703          monkeypatch.setattr(
 704              gateway_cli,
 705              "get_systemd_unit_path",
 706              lambda system=False: system_unit if system else user_unit,
 707          )
 708          monkeypatch.setattr(
 709              gateway_cli,
 710              "get_gateway_runtime_snapshot",
 711              lambda system=False: gateway_cli.GatewayRuntimeSnapshot(
 712                  manager="systemd (user)",
 713                  service_installed=True,
 714                  service_running=False,
 715                  gateway_pids=(),
 716                  service_scope="user",
 717              ),
 718          )
 719  
 720          calls = []
 721          monkeypatch.setattr(
 722              gateway_cli,
 723              "systemd_status",
 724              lambda deep=False, system=False, full=False: calls.append((deep, system, full)),
 725          )
 726  
 727          gateway_cli.gateway_command(
 728              SimpleNamespace(gateway_command="status", deep=False, system=False, full=True)
 729          )
 730  
 731          assert calls == [(False, False, True)]
 732  
 733      def test_gateway_install_passes_system_flags(self, monkeypatch):
 734          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 735          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 736          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 737  
 738          calls = []
 739          monkeypatch.setattr(
 740              gateway_cli,
 741              "systemd_install",
 742              lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)),
 743          )
 744  
 745          gateway_cli.gateway_command(
 746              SimpleNamespace(gateway_command="install", force=True, system=True, run_as_user="alice")
 747          )
 748  
 749          assert calls == [(True, True, "alice")]
 750  
 751      def test_gateway_install_reports_termux_manual_mode(self, monkeypatch, capsys):
 752          monkeypatch.setattr(gateway_cli, "is_termux", lambda: True)
 753          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
 754          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 755  
 756          try:
 757              gateway_cli.gateway_command(
 758                  SimpleNamespace(gateway_command="install", force=False, system=False, run_as_user=None)
 759              )
 760          except SystemExit as exc:
 761              assert exc.code == 1
 762          else:
 763              raise AssertionError("Expected gateway_command to exit on unsupported Termux service install")
 764  
 765          out = capsys.readouterr().out
 766          assert "not supported on Termux" in out
 767          assert "Run manually: hermes gateway" in out
 768  
 769      def test_gateway_status_prefers_system_service_when_only_system_unit_exists(self, monkeypatch):
 770          user_unit = SimpleNamespace(exists=lambda: False)
 771          system_unit = SimpleNamespace(exists=lambda: True)
 772  
 773          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 774          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 775          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 776          monkeypatch.setattr(
 777              gateway_cli,
 778              "get_systemd_unit_path",
 779              lambda system=False: system_unit if system else user_unit,
 780          )
 781  
 782          calls = []
 783          monkeypatch.setattr(
 784              gateway_cli,
 785              "systemd_status",
 786              lambda deep=False, system=False, full=False: calls.append((deep, system, full)),
 787          )
 788  
 789          gateway_cli.gateway_command(SimpleNamespace(gateway_command="status", deep=False, system=False))
 790  
 791          assert calls == [(False, False, False)]
 792  
 793      def test_gateway_status_reports_manual_process_when_service_is_stopped(self, monkeypatch, capsys):
 794          user_unit = SimpleNamespace(exists=lambda: True)
 795          system_unit = SimpleNamespace(exists=lambda: False)
 796  
 797          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
 798          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
 799          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 800          monkeypatch.setattr(
 801              gateway_cli,
 802              "get_systemd_unit_path",
 803              lambda system=False: system_unit if system else user_unit,
 804          )
 805          monkeypatch.setattr(
 806              gateway_cli,
 807              "systemd_status",
 808              lambda deep=False, system=False, full=False: print("service stopped"),
 809          )
 810          monkeypatch.setattr(
 811              gateway_cli,
 812              "get_gateway_runtime_snapshot",
 813              lambda system=False: gateway_cli.GatewayRuntimeSnapshot(
 814                  manager="systemd (user)",
 815                  service_installed=True,
 816                  service_running=False,
 817                  gateway_pids=(4321,),
 818                  service_scope="user",
 819              ),
 820          )
 821  
 822          gateway_cli.gateway_command(SimpleNamespace(gateway_command="status", deep=False, system=False))
 823  
 824          out = capsys.readouterr().out
 825          assert "service stopped" in out
 826          assert "Gateway process is running for this profile" in out
 827          assert "PID(s): 4321" in out
 828  
 829      def test_gateway_status_on_termux_shows_manual_guidance(self, monkeypatch, capsys):
 830          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
 831          monkeypatch.setattr(gateway_cli, "is_termux", lambda: True)
 832          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
 833          monkeypatch.setattr(gateway_cli, "find_gateway_pids", lambda exclude_pids=None: [])
 834          monkeypatch.setattr(gateway_cli, "_runtime_health_lines", lambda: [])
 835  
 836          gateway_cli.gateway_command(SimpleNamespace(gateway_command="status", deep=False, system=False))
 837  
 838          out = capsys.readouterr().out
 839          assert "Gateway is not running" in out
 840          assert "nohup hermes gateway" in out
 841          assert "install as user service" not in out
 842  
 843      def test_gateway_restart_does_not_fallback_to_foreground_when_launchd_restart_fails(self, tmp_path, monkeypatch):
 844          plist_path = tmp_path / "ai.hermes.gateway.plist"
 845          plist_path.write_text("plist\n", encoding="utf-8")
 846  
 847          monkeypatch.setattr(gateway_cli, "is_linux", lambda: False)
 848          monkeypatch.setattr(gateway_cli, "is_macos", lambda: True)
 849          monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
 850          monkeypatch.setattr(
 851              gateway_cli,
 852              "launchd_restart",
 853              lambda: (_ for _ in ()).throw(
 854                  gateway_cli.subprocess.CalledProcessError(5, ["launchctl", "kickstart", "-k", "gui/501/ai.hermes.gateway"])
 855              ),
 856          )
 857  
 858          run_calls = []
 859          monkeypatch.setattr(gateway_cli, "run_gateway", lambda verbose=0, quiet=False, replace=False: run_calls.append((verbose, quiet, replace)))
 860          monkeypatch.setattr(gateway_cli, "kill_gateway_processes", lambda force=False: 0)
 861  
 862          try:
 863              gateway_cli.gateway_command(SimpleNamespace(gateway_command="restart", system=False))
 864          except SystemExit as exc:
 865              assert exc.code == 1
 866          else:
 867              raise AssertionError("Expected gateway_command to exit when service restart fails")
 868  
 869          assert run_calls == []
 870  
 871  
 872  class TestDetectVenvDir:
 873      """Tests for _detect_venv_dir() virtualenv detection."""
 874  
 875      def test_detects_active_virtualenv_via_sys_prefix(self, tmp_path, monkeypatch):
 876          venv_path = tmp_path / "my-custom-venv"
 877          venv_path.mkdir()
 878          monkeypatch.setattr("sys.prefix", str(venv_path))
 879          monkeypatch.setattr("sys.base_prefix", "/usr")
 880  
 881          result = gateway_cli._detect_venv_dir()
 882          assert result == venv_path
 883  
 884      def test_falls_back_to_dot_venv_directory(self, tmp_path, monkeypatch):
 885          # Not inside a virtualenv
 886          monkeypatch.setattr("sys.prefix", "/usr")
 887          monkeypatch.setattr("sys.base_prefix", "/usr")
 888          monkeypatch.delenv("VIRTUAL_ENV", raising=False)
 889          monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
 890  
 891          dot_venv = tmp_path / ".venv"
 892          dot_venv.mkdir()
 893  
 894          result = gateway_cli._detect_venv_dir()
 895          assert result == dot_venv
 896  
 897      def test_falls_back_to_venv_directory(self, tmp_path, monkeypatch):
 898          monkeypatch.setattr("sys.prefix", "/usr")
 899          monkeypatch.setattr("sys.base_prefix", "/usr")
 900          monkeypatch.delenv("VIRTUAL_ENV", raising=False)
 901          monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
 902  
 903          venv = tmp_path / "venv"
 904          venv.mkdir()
 905  
 906          result = gateway_cli._detect_venv_dir()
 907          assert result == venv
 908  
 909      def test_prefers_dot_venv_over_venv(self, tmp_path, monkeypatch):
 910          monkeypatch.setattr("sys.prefix", "/usr")
 911          monkeypatch.setattr("sys.base_prefix", "/usr")
 912          monkeypatch.delenv("VIRTUAL_ENV", raising=False)
 913          monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
 914  
 915          (tmp_path / ".venv").mkdir()
 916          (tmp_path / "venv").mkdir()
 917  
 918          result = gateway_cli._detect_venv_dir()
 919          assert result == tmp_path / ".venv"
 920  
 921      def test_returns_none_when_no_virtualenv(self, tmp_path, monkeypatch):
 922          monkeypatch.setattr("sys.prefix", "/usr")
 923          monkeypatch.setattr("sys.base_prefix", "/usr")
 924          monkeypatch.delenv("VIRTUAL_ENV", raising=False)
 925          monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
 926  
 927          result = gateway_cli._detect_venv_dir()
 928          assert result is None
 929  
 930  
 931  class TestSystemUnitHermesHome:
 932      """HERMES_HOME in system units must reference the target user, not root."""
 933  
 934      def test_system_unit_uses_target_user_home_not_calling_user(self, monkeypatch):
 935          # Simulate sudo: Path.home() returns /root, target user is alice
 936          monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
 937          monkeypatch.delenv("HERMES_HOME", raising=False)
 938          monkeypatch.setattr(
 939              gateway_cli, "_system_service_identity",
 940              lambda run_as_user=None: ("alice", "alice", "/home/alice"),
 941          )
 942          monkeypatch.setattr(
 943              gateway_cli, "_build_user_local_paths",
 944              lambda home, existing: [],
 945          )
 946  
 947          unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
 948  
 949          assert 'HERMES_HOME=/home/alice/.hermes' in unit
 950          assert '/root/.hermes' not in unit
 951  
 952      def test_system_unit_remaps_profile_to_target_user(self, monkeypatch):
 953          # Simulate sudo with a profile: HERMES_HOME was resolved under root
 954          monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
 955          monkeypatch.setenv("HERMES_HOME", "/root/.hermes/profiles/coder")
 956          monkeypatch.setattr(
 957              gateway_cli, "_system_service_identity",
 958              lambda run_as_user=None: ("alice", "alice", "/home/alice"),
 959          )
 960          monkeypatch.setattr(
 961              gateway_cli, "_build_user_local_paths",
 962              lambda home, existing: [],
 963          )
 964  
 965          unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
 966  
 967          assert 'HERMES_HOME=/home/alice/.hermes/profiles/coder' in unit
 968          assert '/root/' not in unit
 969  
 970      def test_system_unit_preserves_custom_hermes_home(self, monkeypatch):
 971          # Custom HERMES_HOME not under any user's home — keep as-is
 972          monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
 973          monkeypatch.setenv("HERMES_HOME", "/opt/hermes-shared")
 974          monkeypatch.setattr(
 975              gateway_cli, "_system_service_identity",
 976              lambda run_as_user=None: ("alice", "alice", "/home/alice"),
 977          )
 978          monkeypatch.setattr(
 979              gateway_cli, "_build_user_local_paths",
 980              lambda home, existing: [],
 981          )
 982  
 983          unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
 984  
 985          assert 'HERMES_HOME=/opt/hermes-shared' in unit
 986  
 987      def test_user_unit_unaffected_by_change(self):
 988          # User-scope units should still use the calling user's HERMES_HOME
 989          unit = gateway_cli.generate_systemd_unit(system=False)
 990  
 991          hermes_home = str(gateway_cli.get_hermes_home().resolve())
 992          assert f'HERMES_HOME={hermes_home}' in unit
 993  
 994  
 995  class TestHermesHomeForTargetUser:
 996      """Unit tests for _hermes_home_for_target_user()."""
 997  
 998      def test_remaps_default_home(self, monkeypatch):
 999          monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
1000          monkeypatch.delenv("HERMES_HOME", raising=False)
1001  
1002          result = gateway_cli._hermes_home_for_target_user("/home/alice")
1003          assert result == "/home/alice/.hermes"
1004  
1005      def test_remaps_profile_path(self, monkeypatch):
1006          monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
1007          monkeypatch.setenv("HERMES_HOME", "/root/.hermes/profiles/coder")
1008  
1009          result = gateway_cli._hermes_home_for_target_user("/home/alice")
1010          assert result == "/home/alice/.hermes/profiles/coder"
1011  
1012      def test_keeps_custom_path(self, monkeypatch):
1013          monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
1014          monkeypatch.setenv("HERMES_HOME", "/opt/hermes")
1015  
1016          result = gateway_cli._hermes_home_for_target_user("/home/alice")
1017          assert result == "/opt/hermes"
1018  
1019      def test_noop_when_same_user(self, monkeypatch):
1020          monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/home/alice")))
1021          monkeypatch.delenv("HERMES_HOME", raising=False)
1022  
1023          result = gateway_cli._hermes_home_for_target_user("/home/alice")
1024          assert result == "/home/alice/.hermes"
1025  
1026  
1027  class TestGeneratedUnitUsesDetectedVenv:
1028      def test_systemd_unit_uses_dot_venv_when_detected(self, tmp_path, monkeypatch):
1029          dot_venv = tmp_path / ".venv"
1030          dot_venv.mkdir()
1031          (dot_venv / "bin").mkdir()
1032  
1033          monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: dot_venv)
1034          monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(dot_venv / "bin" / "python"))
1035  
1036          unit = gateway_cli.generate_systemd_unit(system=False)
1037  
1038          assert f"VIRTUAL_ENV={dot_venv}" in unit
1039          assert f"{dot_venv}/bin" in unit
1040          # Must NOT contain a hardcoded /venv/ path
1041          assert "/venv/" not in unit or "/.venv/" in unit
1042  
1043  
1044  class TestGeneratedUnitIncludesLocalBin:
1045      """~/.local/bin must be in PATH so uvx/pipx tools are discoverable."""
1046  
1047      def test_user_unit_includes_local_bin_in_path(self, monkeypatch):
1048          home = Path.home()
1049          monkeypatch.setattr(
1050              gateway_cli,
1051              "_build_user_local_paths",
1052              lambda home_path, existing: [str(home / ".local" / "bin")],
1053          )
1054          unit = gateway_cli.generate_systemd_unit(system=False)
1055          assert f"{home}/.local/bin" in unit
1056  
1057      def test_system_unit_includes_local_bin_in_path(self, monkeypatch):
1058          monkeypatch.setattr(
1059              gateway_cli,
1060              "_build_user_local_paths",
1061              lambda home_path, existing: [str(home_path / ".local" / "bin")],
1062          )
1063          unit = gateway_cli.generate_systemd_unit(system=True)
1064          # System unit uses the resolved home dir from _system_service_identity
1065          assert "/.local/bin" in unit
1066  
1067  
1068  class TestSystemServiceIdentityRootHandling:
1069      """Root user handling in _system_service_identity()."""
1070  
1071      def test_auto_detected_root_is_rejected(self, monkeypatch):
1072          """When root is auto-detected (not explicitly requested), raise."""
1073          import pwd
1074          import grp
1075  
1076          monkeypatch.delenv("SUDO_USER", raising=False)
1077          monkeypatch.setenv("USER", "root")
1078          monkeypatch.setenv("LOGNAME", "root")
1079  
1080          import pytest
1081          with pytest.raises(ValueError, match="pass --run-as-user root to override"):
1082              gateway_cli._system_service_identity(run_as_user=None)
1083  
1084      def test_explicit_root_is_allowed(self, monkeypatch):
1085          """When root is explicitly passed via --run-as-user root, allow it."""
1086          import pwd
1087          import grp
1088  
1089          root_info = pwd.getpwnam("root")
1090          root_group = grp.getgrgid(root_info.pw_gid).gr_name
1091  
1092          username, group, home = gateway_cli._system_service_identity(run_as_user="root")
1093          assert username == "root"
1094          assert home == root_info.pw_dir
1095  
1096      def test_non_root_user_passes_through(self, monkeypatch):
1097          """Normal non-root user works as before."""
1098          import pwd
1099          import grp
1100  
1101          monkeypatch.delenv("SUDO_USER", raising=False)
1102          monkeypatch.setenv("USER", "nobody")
1103          monkeypatch.setenv("LOGNAME", "nobody")
1104  
1105          try:
1106              username, group, home = gateway_cli._system_service_identity(run_as_user=None)
1107              assert username == "nobody"
1108          except ValueError as e:
1109              # "nobody" might not exist on all systems
1110              assert "Unknown user" in str(e)
1111  
1112  
1113  class TestEnsureUserSystemdEnv:
1114      """Tests for _ensure_user_systemd_env() D-Bus session bus auto-detection."""
1115  
1116      def test_sets_xdg_runtime_dir_when_missing(self, tmp_path, monkeypatch):
1117          monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False)
1118          monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
1119          monkeypatch.setattr(os, "getuid", lambda: 42)
1120  
1121          # Patch Path.exists so /run/user/42 appears to exist.
1122          # Using a FakePath subclass breaks on Python 3.12+ where
1123          # PosixPath.__new__ ignores the redirected path argument.
1124          _orig_exists = gateway_cli.Path.exists
1125          monkeypatch.setattr(
1126              gateway_cli.Path, "exists",
1127              lambda self: True if str(self) == "/run/user/42" else _orig_exists(self),
1128          )
1129  
1130          gateway_cli._ensure_user_systemd_env()
1131  
1132          assert os.environ.get("XDG_RUNTIME_DIR") == "/run/user/42"
1133  
1134      def test_sets_dbus_address_when_bus_socket_exists(self, tmp_path, monkeypatch):
1135          runtime = tmp_path / "runtime"
1136          runtime.mkdir()
1137          bus_socket = runtime / "bus"
1138          bus_socket.touch()  # simulate the socket file
1139  
1140          monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime))
1141          monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
1142          monkeypatch.setattr(os, "getuid", lambda: 99)
1143  
1144          gateway_cli._ensure_user_systemd_env()
1145  
1146          assert os.environ["DBUS_SESSION_BUS_ADDRESS"] == f"unix:path={bus_socket}"
1147  
1148      def test_preserves_existing_env_vars(self, monkeypatch):
1149          monkeypatch.setenv("XDG_RUNTIME_DIR", "/custom/runtime")
1150          monkeypatch.setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/custom/bus")
1151  
1152          gateway_cli._ensure_user_systemd_env()
1153  
1154          assert os.environ["XDG_RUNTIME_DIR"] == "/custom/runtime"
1155          assert os.environ["DBUS_SESSION_BUS_ADDRESS"] == "unix:path=/custom/bus"
1156  
1157      def test_no_dbus_when_bus_socket_missing(self, tmp_path, monkeypatch):
1158          runtime = tmp_path / "runtime"
1159          runtime.mkdir()
1160          # no bus socket created
1161  
1162          monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime))
1163          monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
1164          monkeypatch.setattr(os, "getuid", lambda: 99)
1165  
1166          gateway_cli._ensure_user_systemd_env()
1167  
1168          assert "DBUS_SESSION_BUS_ADDRESS" not in os.environ
1169  
1170      def test_systemctl_cmd_calls_ensure_for_user_mode(self, monkeypatch):
1171          calls = []
1172          monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("called"))
1173  
1174          result = gateway_cli._systemctl_cmd(system=False)
1175          assert result == ["systemctl", "--user"]
1176          assert calls == ["called"]
1177  
1178      def test_systemctl_cmd_skips_ensure_for_system_mode(self, monkeypatch):
1179          calls = []
1180          monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("called"))
1181  
1182          result = gateway_cli._systemctl_cmd(system=True)
1183          assert result == ["systemctl"]
1184          assert calls == []
1185  
1186  
1187  class TestPreflightUserSystemd:
1188      """Tests for _preflight_user_systemd() — D-Bus reachability before systemctl --user.
1189  
1190      Covers issue #5130 / Rick's RHEL 9.6 SSH scenario: setup tries to start the
1191      gateway via ``systemctl --user start`` in a shell with no user D-Bus session,
1192      which previously failed with a raw ``CalledProcessError`` and no remediation.
1193      """
1194  
1195      def test_noop_when_bus_socket_exists(self, monkeypatch):
1196          """Socket already there (desktop / linger + prior login) → no-op."""
1197          monkeypatch.setattr(
1198              gateway_cli, "_user_dbus_socket_path",
1199              lambda: type("P", (), {"exists": lambda self: True})(),
1200          )
1201          monkeypatch.setattr(
1202              gateway_cli, "_user_systemd_private_socket_path",
1203              lambda: type("P", (), {"exists": lambda self: False})(),
1204          )
1205          # Should not raise, no subprocess calls needed.
1206          gateway_cli._preflight_user_systemd()
1207  
1208      def test_raises_when_linger_disabled_and_loginctl_denied(self, monkeypatch):
1209          """Rick's scenario: no D-Bus, no linger, non-root SSH → clear error."""
1210          monkeypatch.setattr(
1211              gateway_cli, "_user_dbus_socket_path",
1212              lambda: type("P", (), {"exists": lambda self: False})(),
1213          )
1214          monkeypatch.setattr(
1215              gateway_cli, "_user_systemd_private_socket_path",
1216              lambda: type("P", (), {"exists": lambda self: False})(),
1217          )
1218          monkeypatch.setattr(
1219              gateway_cli, "get_systemd_linger_status", lambda: (False, ""),
1220          )
1221          monkeypatch.setattr(gateway_cli.shutil, "which", lambda _: "/usr/bin/loginctl")
1222  
1223          class _Result:
1224              returncode = 1
1225              stdout = ""
1226              stderr = "Interactive authentication required."
1227  
1228          monkeypatch.setattr(
1229              gateway_cli.subprocess, "run", lambda *a, **kw: _Result(),
1230          )
1231  
1232          with pytest.raises(gateway_cli.UserSystemdUnavailableError) as exc_info:
1233              gateway_cli._preflight_user_systemd()
1234  
1235          msg = str(exc_info.value)
1236          assert "sudo loginctl enable-linger" in msg
1237          assert "hermes gateway run" in msg  # foreground fallback mentioned
1238          assert "Interactive authentication required" in msg
1239  
1240      def test_raises_when_loginctl_missing(self, monkeypatch):
1241          """No loginctl binary at all → suggest sudo install + manual fix."""
1242          monkeypatch.setattr(
1243              gateway_cli, "_user_dbus_socket_path",
1244              lambda: type("P", (), {"exists": lambda self: False})(),
1245          )
1246          monkeypatch.setattr(
1247              gateway_cli, "_user_systemd_private_socket_path",
1248              lambda: type("P", (), {"exists": lambda self: False})(),
1249          )
1250          monkeypatch.setattr(
1251              gateway_cli, "get_systemd_linger_status",
1252              lambda: (None, "loginctl not found"),
1253          )
1254          monkeypatch.setattr(gateway_cli.shutil, "which", lambda _: None)
1255  
1256          with pytest.raises(gateway_cli.UserSystemdUnavailableError) as exc_info:
1257              gateway_cli._preflight_user_systemd()
1258  
1259          assert "sudo loginctl enable-linger" in str(exc_info.value)
1260  
1261      def test_linger_enabled_but_socket_still_missing(self, monkeypatch):
1262          """Edge case: linger says yes but the bus socket never came up."""
1263          monkeypatch.setattr(
1264              gateway_cli, "_user_dbus_socket_path",
1265              lambda: type("P", (), {"exists": lambda self: False})(),
1266          )
1267          monkeypatch.setattr(
1268              gateway_cli, "_user_systemd_private_socket_path",
1269              lambda: type("P", (), {"exists": lambda self: False})(),
1270          )
1271          monkeypatch.setattr(
1272              gateway_cli, "get_systemd_linger_status", lambda: (True, ""),
1273          )
1274          monkeypatch.setattr(
1275              gateway_cli, "_wait_for_user_dbus_socket", lambda timeout=3.0: False,
1276          )
1277  
1278          with pytest.raises(gateway_cli.UserSystemdUnavailableError) as exc_info:
1279              gateway_cli._preflight_user_systemd()
1280  
1281          assert "linger is enabled" in str(exc_info.value)
1282  
1283      def test_enable_linger_succeeds_and_socket_appears(self, monkeypatch, capsys):
1284          """Happy remediation path: polkit allows enable-linger, socket spawns."""
1285          monkeypatch.setattr(
1286              gateway_cli, "_user_dbus_socket_path",
1287              lambda: type("P", (), {"exists": lambda self: False})(),
1288          )
1289          monkeypatch.setattr(
1290              gateway_cli, "_user_systemd_private_socket_path",
1291              lambda: type("P", (), {"exists": lambda self: False})(),
1292          )
1293          monkeypatch.setattr(
1294              gateway_cli, "get_systemd_linger_status", lambda: (False, ""),
1295          )
1296          monkeypatch.setattr(gateway_cli.shutil, "which", lambda _: "/usr/bin/loginctl")
1297  
1298          class _OkResult:
1299              returncode = 0
1300              stdout = ""
1301              stderr = ""
1302  
1303          monkeypatch.setattr(
1304              gateway_cli.subprocess, "run", lambda *a, **kw: _OkResult(),
1305          )
1306          monkeypatch.setattr(
1307              gateway_cli, "_wait_for_user_dbus_socket",
1308              lambda timeout=5.0: True,
1309          )
1310  
1311          # Should not raise.
1312          gateway_cli._preflight_user_systemd()
1313          out = capsys.readouterr().out
1314          assert "Enabled linger" in out
1315  
1316  
1317  class TestProfileArg:
1318      """Tests for _profile_arg — returns '--profile <name>' for named profiles."""
1319  
1320      def test_default_hermes_home_returns_empty(self, tmp_path, monkeypatch):
1321          """Default ~/.hermes should not produce a --profile flag."""
1322          hermes_home = tmp_path / ".hermes"
1323          hermes_home.mkdir()
1324          monkeypatch.setattr(Path, "home", lambda: tmp_path)
1325          monkeypatch.setenv("HERMES_HOME", str(hermes_home))
1326          result = gateway_cli._profile_arg(str(hermes_home))
1327          assert result == ""
1328  
1329      def test_named_profile_returns_flag(self, tmp_path, monkeypatch):
1330          """~/.hermes/profiles/mybot should return '--profile mybot'."""
1331          profile_dir = tmp_path / ".hermes" / "profiles" / "mybot"
1332          profile_dir.mkdir(parents=True)
1333          monkeypatch.setattr(Path, "home", lambda: tmp_path)
1334          monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
1335          result = gateway_cli._profile_arg(str(profile_dir))
1336          assert result == "--profile mybot"
1337  
1338      def test_hash_path_returns_empty(self, tmp_path, monkeypatch):
1339          """Arbitrary non-profile HERMES_HOME should return empty string."""
1340          custom_home = tmp_path / "custom" / "hermes"
1341          custom_home.mkdir(parents=True)
1342          monkeypatch.setattr(Path, "home", lambda: tmp_path)
1343          monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
1344          result = gateway_cli._profile_arg(str(custom_home))
1345          assert result == ""
1346  
1347      def test_nested_profile_path_returns_empty(self, tmp_path, monkeypatch):
1348          """~/.hermes/profiles/mybot/subdir should NOT match — too deep."""
1349          nested = tmp_path / ".hermes" / "profiles" / "mybot" / "subdir"
1350          nested.mkdir(parents=True)
1351          monkeypatch.setattr(Path, "home", lambda: tmp_path)
1352          monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
1353          result = gateway_cli._profile_arg(str(nested))
1354          assert result == ""
1355  
1356      def test_invalid_profile_name_returns_empty(self, tmp_path, monkeypatch):
1357          """Profile names with invalid chars should not match the regex."""
1358          bad_profile = tmp_path / ".hermes" / "profiles" / "My Bot!"
1359          bad_profile.mkdir(parents=True)
1360          monkeypatch.setattr(Path, "home", lambda: tmp_path)
1361          monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
1362          result = gateway_cli._profile_arg(str(bad_profile))
1363          assert result == ""
1364  
1365      def test_systemd_unit_includes_profile(self, tmp_path, monkeypatch):
1366          """generate_systemd_unit should include --profile in ExecStart for named profiles."""
1367          profile_dir = tmp_path / ".hermes" / "profiles" / "mybot"
1368          profile_dir.mkdir(parents=True)
1369          monkeypatch.setattr(Path, "home", lambda: tmp_path)
1370          monkeypatch.setenv("HERMES_HOME", str(profile_dir))
1371          monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: profile_dir)
1372          unit = gateway_cli.generate_systemd_unit(system=False)
1373          assert "--profile mybot" in unit
1374          assert "gateway run --replace" in unit
1375  
1376      def test_launchd_plist_includes_profile(self, tmp_path, monkeypatch):
1377          """generate_launchd_plist should include --profile in ProgramArguments for named profiles."""
1378          profile_dir = tmp_path / ".hermes" / "profiles" / "mybot"
1379          profile_dir.mkdir(parents=True)
1380          monkeypatch.setattr(Path, "home", lambda: tmp_path)
1381          monkeypatch.setenv("HERMES_HOME", str(profile_dir))
1382          monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: profile_dir)
1383          plist = gateway_cli.generate_launchd_plist()
1384          assert "<string>--profile</string>" in plist
1385          assert "<string>mybot</string>" in plist
1386  
1387      def test_launchd_plist_path_uses_real_user_home_not_profile_home(self, tmp_path, monkeypatch):
1388          profile_dir = tmp_path / ".hermes" / "profiles" / "orcha"
1389          profile_dir.mkdir(parents=True)
1390          machine_home = tmp_path / "machine-home"
1391          machine_home.mkdir()
1392          profile_home = profile_dir / "home"
1393          profile_home.mkdir()
1394  
1395          monkeypatch.setattr(Path, "home", lambda: profile_home)
1396          monkeypatch.setenv("HERMES_HOME", str(profile_dir))
1397          monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: profile_dir)
1398          monkeypatch.setattr(pwd, "getpwuid", lambda uid: SimpleNamespace(pw_dir=str(machine_home)))
1399  
1400          plist_path = gateway_cli.get_launchd_plist_path()
1401  
1402          assert plist_path == machine_home / "Library" / "LaunchAgents" / "ai.hermes.gateway-orcha.plist"
1403  
1404  
1405  class TestRemapPathForUser:
1406      """Unit tests for _remap_path_for_user()."""
1407  
1408      def test_remaps_path_under_current_home(self, monkeypatch, tmp_path):
1409          monkeypatch.setattr(Path, "home", lambda: tmp_path / "root")
1410          (tmp_path / "root").mkdir()
1411          result = gateway_cli._remap_path_for_user(
1412              str(tmp_path / "root" / ".hermes" / "hermes-agent"),
1413              str(tmp_path / "alice"),
1414          )
1415          assert result == str(tmp_path / "alice" / ".hermes" / "hermes-agent")
1416  
1417      def test_keeps_system_path_unchanged(self, monkeypatch, tmp_path):
1418          monkeypatch.setattr(Path, "home", lambda: tmp_path / "root")
1419          (tmp_path / "root").mkdir()
1420          result = gateway_cli._remap_path_for_user("/opt/hermes", str(tmp_path / "alice"))
1421          assert result == "/opt/hermes"
1422  
1423      def test_noop_when_same_user(self, monkeypatch, tmp_path):
1424          monkeypatch.setattr(Path, "home", lambda: tmp_path / "alice")
1425          (tmp_path / "alice").mkdir()
1426          original = str(tmp_path / "alice" / ".hermes" / "hermes-agent")
1427          result = gateway_cli._remap_path_for_user(original, str(tmp_path / "alice"))
1428          assert result == original
1429  
1430  
1431  class TestSystemUnitPathRemapping:
1432      """System units must remap ALL paths from the caller's home to the target user."""
1433  
1434      def test_system_unit_has_no_root_paths(self, monkeypatch, tmp_path):
1435          root_home = tmp_path / "root"
1436          root_home.mkdir()
1437          project = root_home / ".hermes" / "hermes-agent"
1438          project.mkdir(parents=True)
1439          venv_bin = project / "venv" / "bin"
1440          venv_bin.mkdir(parents=True)
1441          (venv_bin / "python").write_text("")
1442  
1443          target_home = "/home/alice"
1444  
1445          monkeypatch.setattr(Path, "home", lambda: root_home)
1446          monkeypatch.setenv("HERMES_HOME", str(root_home / ".hermes"))
1447          monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: root_home / ".hermes")
1448          monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", project)
1449          monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: project / "venv")
1450          monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(venv_bin / "python"))
1451          monkeypatch.setattr(
1452              gateway_cli, "_system_service_identity",
1453              lambda run_as_user=None: ("alice", "alice", target_home),
1454          )
1455  
1456          unit = gateway_cli.generate_systemd_unit(system=True)
1457  
1458          # No root paths should leak into the unit
1459          assert str(root_home) not in unit
1460          # Target user paths should be present
1461          assert "/home/alice" in unit
1462          assert "WorkingDirectory=/home/alice/.hermes/hermes-agent" in unit
1463  
1464  
1465  class TestDockerAwareGateway:
1466      """Tests for Docker container awareness in gateway commands."""
1467  
1468      def test_run_systemctl_raises_runtimeerror_when_missing(self, monkeypatch):
1469          """_run_systemctl raises RuntimeError with container guidance when systemctl is absent."""
1470          import pytest
1471  
1472          def fake_run(cmd, **kwargs):
1473              raise FileNotFoundError("systemctl")
1474  
1475          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
1476  
1477          with pytest.raises(RuntimeError, match="systemctl is not available"):
1478              gateway_cli._run_systemctl(["start", "hermes-gateway"])
1479  
1480      def test_run_systemctl_passes_through_on_success(self, monkeypatch):
1481          """_run_systemctl delegates to subprocess.run when systemctl exists."""
1482          calls = []
1483  
1484          def fake_run(cmd, **kwargs):
1485              calls.append(cmd)
1486              return SimpleNamespace(returncode=0, stdout="", stderr="")
1487  
1488          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
1489  
1490          result = gateway_cli._run_systemctl(["status", "hermes-gateway"])
1491          assert result.returncode == 0
1492          assert len(calls) == 1
1493          assert "status" in calls[0]
1494  
1495      def test_install_in_container_prints_docker_guidance(self, monkeypatch, capsys):
1496          """'hermes gateway install' inside Docker exits 0 with container guidance."""
1497          import pytest
1498  
1499          monkeypatch.setattr(gateway_cli, "is_managed", lambda: False)
1500          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
1501          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
1502          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
1503          monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
1504          monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
1505  
1506          args = SimpleNamespace(gateway_command="install", force=False, system=False, run_as_user=None)
1507          with pytest.raises(SystemExit) as exc_info:
1508              gateway_cli.gateway_command(args)
1509  
1510          assert exc_info.value.code == 0
1511          out = capsys.readouterr().out
1512          assert "Docker" in out or "docker" in out
1513          assert "restart" in out.lower()
1514  
1515      def test_uninstall_in_container_prints_docker_guidance(self, monkeypatch, capsys):
1516          """'hermes gateway uninstall' inside Docker exits 0 with container guidance."""
1517          import pytest
1518  
1519          monkeypatch.setattr(gateway_cli, "is_managed", lambda: False)
1520          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
1521          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
1522          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
1523          monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
1524  
1525          args = SimpleNamespace(gateway_command="uninstall", system=False)
1526          with pytest.raises(SystemExit) as exc_info:
1527              gateway_cli.gateway_command(args)
1528  
1529          assert exc_info.value.code == 0
1530          out = capsys.readouterr().out
1531          assert "docker" in out.lower()
1532  
1533      def test_start_in_container_prints_docker_guidance(self, monkeypatch, capsys):
1534          """'hermes gateway start' inside Docker exits 0 with container guidance."""
1535          import pytest
1536  
1537          monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
1538          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
1539          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
1540          monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
1541          monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
1542  
1543          args = SimpleNamespace(gateway_command="start", system=False)
1544          with pytest.raises(SystemExit) as exc_info:
1545              gateway_cli.gateway_command(args)
1546  
1547          assert exc_info.value.code == 0
1548          out = capsys.readouterr().out
1549          assert "docker" in out.lower()
1550          assert "hermes gateway run" in out
1551  
1552  
1553  class TestLegacyHermesUnitDetection:
1554      """Tests for _find_legacy_hermes_units / has_legacy_hermes_units.
1555  
1556      These guard against the scenario that tripped Luis in April 2026: an
1557      older install left a ``hermes.service`` unit behind when the service was
1558      renamed to ``hermes-gateway.service``. After PR #5646 (signal recovery
1559      via systemd), the two services began SIGTERM-flapping over the same
1560      Telegram bot token in a 30-second cycle.
1561  
1562      The detector must flag ``hermes.service`` ONLY when it actually runs our
1563      gateway, and must NEVER flag profile units
1564      (``hermes-gateway-<profile>.service``) or unrelated third-party services.
1565      """
1566  
1567      # Minimal ExecStart that looks like our gateway
1568      _OUR_UNIT_TEXT = (
1569          "[Unit]\nDescription=Hermes Gateway\n[Service]\n"
1570          "ExecStart=/usr/bin/python -m hermes_cli.main gateway run --replace\n"
1571      )
1572  
1573      @staticmethod
1574      def _setup_search_paths(tmp_path, monkeypatch):
1575          """Redirect the legacy search to user_dir + system_dir under tmp_path."""
1576          user_dir = tmp_path / "user"
1577          system_dir = tmp_path / "system"
1578          user_dir.mkdir()
1579          system_dir.mkdir()
1580          monkeypatch.setattr(
1581              gateway_cli,
1582              "_legacy_unit_search_paths",
1583              lambda: [(False, user_dir), (True, system_dir)],
1584          )
1585          return user_dir, system_dir
1586  
1587      def test_detects_legacy_hermes_service_in_user_scope(self, tmp_path, monkeypatch):
1588          user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
1589          legacy = user_dir / "hermes.service"
1590          legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1591  
1592          results = gateway_cli._find_legacy_hermes_units()
1593  
1594          assert len(results) == 1
1595          name, path, is_system = results[0]
1596          assert name == "hermes.service"
1597          assert path == legacy
1598          assert is_system is False
1599          assert gateway_cli.has_legacy_hermes_units() is True
1600  
1601      def test_detects_legacy_hermes_service_in_system_scope(self, tmp_path, monkeypatch):
1602          _, system_dir = self._setup_search_paths(tmp_path, monkeypatch)
1603          legacy = system_dir / "hermes.service"
1604          legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1605  
1606          results = gateway_cli._find_legacy_hermes_units()
1607  
1608          assert len(results) == 1
1609          name, path, is_system = results[0]
1610          assert name == "hermes.service"
1611          assert path == legacy
1612          assert is_system is True
1613  
1614      def test_ignores_profile_unit_hermes_gateway_coder(self, tmp_path, monkeypatch):
1615          """CRITICAL: profile units must NOT be flagged as legacy.
1616  
1617          Teknium's concern — ``hermes-gateway-coder.service`` is our standard
1618          naming for the ``coder`` profile. The legacy detector is an explicit
1619          allowlist, not a glob, so profile units are safe.
1620          """
1621          user_dir, system_dir = self._setup_search_paths(tmp_path, monkeypatch)
1622          # Drop profile units in BOTH scopes with our ExecStart
1623          for base in (user_dir, system_dir):
1624              (base / "hermes-gateway-coder.service").write_text(
1625                  self._OUR_UNIT_TEXT, encoding="utf-8"
1626              )
1627              (base / "hermes-gateway-orcha.service").write_text(
1628                  self._OUR_UNIT_TEXT, encoding="utf-8"
1629              )
1630              (base / "hermes-gateway.service").write_text(
1631                  self._OUR_UNIT_TEXT, encoding="utf-8"
1632              )
1633  
1634          results = gateway_cli._find_legacy_hermes_units()
1635  
1636          assert results == []
1637          assert gateway_cli.has_legacy_hermes_units() is False
1638  
1639      def test_ignores_unrelated_hermes_service(self, tmp_path, monkeypatch):
1640          """Third-party ``hermes.service`` that isn't ours stays untouched.
1641  
1642          If a user has some other package named ``hermes`` installed as a
1643          service, we must not flag it.
1644          """
1645          user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
1646          (user_dir / "hermes.service").write_text(
1647              "[Unit]\nDescription=Some Other Hermes\n[Service]\n"
1648              "ExecStart=/opt/other-hermes/bin/daemon --foreground\n",
1649              encoding="utf-8",
1650          )
1651  
1652          results = gateway_cli._find_legacy_hermes_units()
1653  
1654          assert results == []
1655          assert gateway_cli.has_legacy_hermes_units() is False
1656  
1657      def test_returns_empty_when_no_legacy_files_exist(self, tmp_path, monkeypatch):
1658          self._setup_search_paths(tmp_path, monkeypatch)
1659  
1660          assert gateway_cli._find_legacy_hermes_units() == []
1661          assert gateway_cli.has_legacy_hermes_units() is False
1662  
1663      def test_detects_both_scopes_simultaneously(self, tmp_path, monkeypatch):
1664          """When a user has BOTH user-scope and system-scope legacy units,
1665          both are reported so the migration step can remove them together."""
1666          user_dir, system_dir = self._setup_search_paths(tmp_path, monkeypatch)
1667          (user_dir / "hermes.service").write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1668          (system_dir / "hermes.service").write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1669  
1670          results = gateway_cli._find_legacy_hermes_units()
1671  
1672          scopes = sorted(is_system for _, _, is_system in results)
1673          assert scopes == [False, True]
1674  
1675      def test_accepts_alternate_execstart_formats(self, tmp_path, monkeypatch):
1676          """Older installs may have used different python invocations.
1677  
1678          ExecStart variants we've seen in the wild:
1679            - python -m hermes_cli.main gateway run
1680            - python path/to/hermes_cli/main.py gateway run
1681            - hermes gateway run   (direct binary)
1682            - python path/to/gateway/run.py
1683          """
1684          user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
1685          variants = [
1686              "ExecStart=/venv/bin/python -m hermes_cli.main gateway run --replace",
1687              "ExecStart=/venv/bin/python /opt/hermes/hermes_cli/main.py gateway run",
1688              "ExecStart=/usr/local/bin/hermes gateway run --replace",
1689              "ExecStart=/venv/bin/python /opt/hermes/gateway/run.py",
1690          ]
1691          for i, execstart in enumerate(variants):
1692              name = f"hermes.service" if i == 0 else f"hermes.service"  # same name
1693              # Test each variant fresh
1694              (user_dir / "hermes.service").write_text(
1695                  f"[Unit]\nDescription=Old Hermes\n[Service]\n{execstart}\n",
1696                  encoding="utf-8",
1697              )
1698              results = gateway_cli._find_legacy_hermes_units()
1699              assert len(results) == 1, f"Variant {i} not detected: {execstart!r}"
1700  
1701      def test_print_legacy_unit_warning_is_noop_when_empty(self, tmp_path, monkeypatch, capsys):
1702          self._setup_search_paths(tmp_path, monkeypatch)
1703  
1704          gateway_cli.print_legacy_unit_warning()
1705          out = capsys.readouterr().out
1706  
1707          assert out == ""
1708  
1709      def test_print_legacy_unit_warning_shows_migration_hint(self, tmp_path, monkeypatch, capsys):
1710          user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
1711          (user_dir / "hermes.service").write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1712  
1713          gateway_cli.print_legacy_unit_warning()
1714          out = capsys.readouterr().out
1715  
1716          assert "Legacy" in out
1717          assert "hermes.service" in out
1718          assert "hermes gateway migrate-legacy" in out
1719  
1720      def test_handles_unreadable_unit_file_gracefully(self, tmp_path, monkeypatch):
1721          """A permission error reading a unit file must not crash detection."""
1722          user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
1723          unreadable = user_dir / "hermes.service"
1724          unreadable.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1725          # Simulate a read failure — monkeypatch Path.read_text to raise
1726          original_read_text = gateway_cli.Path.read_text
1727  
1728          def raising_read_text(self, *args, **kwargs):
1729              if self == unreadable:
1730                  raise PermissionError("simulated")
1731              return original_read_text(self, *args, **kwargs)
1732  
1733          monkeypatch.setattr(gateway_cli.Path, "read_text", raising_read_text)
1734  
1735          # Should not raise
1736          results = gateway_cli._find_legacy_hermes_units()
1737          assert results == []
1738  
1739  
1740  class TestRemoveLegacyHermesUnits:
1741      """Tests for remove_legacy_hermes_units (the migration action)."""
1742  
1743      _OUR_UNIT_TEXT = (
1744          "[Unit]\nDescription=Hermes Gateway\n[Service]\n"
1745          "ExecStart=/usr/bin/python -m hermes_cli.main gateway run --replace\n"
1746      )
1747  
1748      @staticmethod
1749      def _setup(tmp_path, monkeypatch, as_root=False):
1750          user_dir = tmp_path / "user"
1751          system_dir = tmp_path / "system"
1752          user_dir.mkdir()
1753          system_dir.mkdir()
1754          monkeypatch.setattr(
1755              gateway_cli,
1756              "_legacy_unit_search_paths",
1757              lambda: [(False, user_dir), (True, system_dir)],
1758          )
1759          # Mock systemctl — return success for everything
1760          systemctl_calls: list[list[str]] = []
1761  
1762          def fake_run(cmd, **kwargs):
1763              systemctl_calls.append(cmd)
1764              return SimpleNamespace(returncode=0, stdout="", stderr="")
1765  
1766          monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
1767          monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 0 if as_root else 1000)
1768          return user_dir, system_dir, systemctl_calls
1769  
1770      def test_returns_zero_when_no_legacy_units(self, tmp_path, monkeypatch, capsys):
1771          self._setup(tmp_path, monkeypatch)
1772  
1773          removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
1774  
1775          assert removed == 0
1776          assert remaining == []
1777          assert "No legacy" in capsys.readouterr().out
1778  
1779      def test_dry_run_lists_without_removing(self, tmp_path, monkeypatch, capsys):
1780          user_dir, _, calls = self._setup(tmp_path, monkeypatch)
1781          legacy = user_dir / "hermes.service"
1782          legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1783  
1784          removed, remaining = gateway_cli.remove_legacy_hermes_units(
1785              interactive=False, dry_run=True
1786          )
1787  
1788          assert removed == 0
1789          assert remaining == [legacy]
1790          assert legacy.exists()  # Not removed
1791          assert calls == []  # No systemctl invocations
1792          out = capsys.readouterr().out
1793          assert "dry-run" in out
1794  
1795      def test_removes_user_scope_legacy_unit(self, tmp_path, monkeypatch, capsys):
1796          user_dir, _, calls = self._setup(tmp_path, monkeypatch)
1797          legacy = user_dir / "hermes.service"
1798          legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1799  
1800          removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
1801  
1802          assert removed == 1
1803          assert remaining == []
1804          assert not legacy.exists()
1805          # Must have invoked stop → disable → daemon-reload on user scope
1806          cmds_joined = [" ".join(c) for c in calls]
1807          assert any("--user stop hermes.service" in c for c in cmds_joined)
1808          assert any("--user disable hermes.service" in c for c in cmds_joined)
1809          assert any("--user daemon-reload" in c for c in cmds_joined)
1810  
1811      def test_system_scope_without_root_defers_removal(self, tmp_path, monkeypatch, capsys):
1812          _, system_dir, calls = self._setup(tmp_path, monkeypatch, as_root=False)
1813          legacy = system_dir / "hermes.service"
1814          legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1815  
1816          removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
1817  
1818          assert removed == 0
1819          assert remaining == [legacy]
1820          assert legacy.exists()  # Not removed — requires sudo
1821          out = capsys.readouterr().out
1822          assert "sudo hermes gateway migrate-legacy" in out
1823  
1824      def test_system_scope_with_root_removes(self, tmp_path, monkeypatch, capsys):
1825          _, system_dir, calls = self._setup(tmp_path, monkeypatch, as_root=True)
1826          legacy = system_dir / "hermes.service"
1827          legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1828  
1829          removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
1830  
1831          assert removed == 1
1832          assert remaining == []
1833          assert not legacy.exists()
1834          cmds_joined = [" ".join(c) for c in calls]
1835          # System-scope uses plain "systemctl" (no --user)
1836          assert any(
1837              c.startswith("systemctl stop hermes.service") for c in cmds_joined
1838          )
1839          assert any(
1840              c.startswith("systemctl disable hermes.service") for c in cmds_joined
1841          )
1842  
1843      def test_removes_both_scopes_with_root(self, tmp_path, monkeypatch, capsys):
1844          user_dir, system_dir, _ = self._setup(tmp_path, monkeypatch, as_root=True)
1845          user_legacy = user_dir / "hermes.service"
1846          system_legacy = system_dir / "hermes.service"
1847          user_legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1848          system_legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1849  
1850          removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
1851  
1852          assert removed == 2
1853          assert remaining == []
1854          assert not user_legacy.exists()
1855          assert not system_legacy.exists()
1856  
1857      def test_does_not_touch_profile_units_during_migration(
1858          self, tmp_path, monkeypatch, capsys
1859      ):
1860          """Teknium's constraint: profile units (hermes-gateway-coder.service)
1861          must survive a migration call, even if we somehow include them in the
1862          search dir."""
1863          user_dir, _, _ = self._setup(tmp_path, monkeypatch, as_root=True)
1864          profile_unit = user_dir / "hermes-gateway-coder.service"
1865          profile_unit.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1866          default_unit = user_dir / "hermes-gateway.service"
1867          default_unit.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1868  
1869          removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
1870  
1871          assert removed == 0
1872          assert remaining == []
1873          # Both the profile unit and the current default unit must survive
1874          assert profile_unit.exists()
1875          assert default_unit.exists()
1876  
1877      def test_interactive_prompt_no_skips_removal(self, tmp_path, monkeypatch, capsys):
1878          """When interactive=True and user answers no, no removal happens."""
1879          user_dir, _, _ = self._setup(tmp_path, monkeypatch)
1880          legacy = user_dir / "hermes.service"
1881          legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
1882  
1883          monkeypatch.setattr(gateway_cli, "prompt_yes_no", lambda *a, **k: False)
1884  
1885          removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=True)
1886  
1887          assert removed == 0
1888          assert remaining == [legacy]
1889          assert legacy.exists()
1890  
1891  
1892  class TestMigrateLegacyCommand:
1893      """Tests for the `hermes gateway migrate-legacy` subcommand dispatch."""
1894  
1895      def test_migrate_legacy_subparser_accepts_dry_run_and_yes(self):
1896          """Verify the argparse subparser is registered and parses flags."""
1897          import hermes_cli.main as cli_main
1898  
1899          parser = cli_main.build_parser() if hasattr(cli_main, "build_parser") else None
1900          # Fall back to calling main's setup helper if direct access isn't exposed
1901          # The key thing: the subparser must exist. We verify by constructing
1902          # a namespace through argparse directly — but if build_parser isn't
1903          # public, just confirm that `hermes gateway --help` shows it.
1904          import subprocess
1905          import sys
1906  
1907          project_root = cli_main.PROJECT_ROOT if hasattr(cli_main, "PROJECT_ROOT") else None
1908          if project_root is None:
1909              import hermes_cli.gateway as gw
1910              project_root = gw.PROJECT_ROOT
1911  
1912          result = subprocess.run(
1913              [sys.executable, "-m", "hermes_cli.main", "gateway", "--help"],
1914              cwd=str(project_root),
1915              capture_output=True,
1916              text=True,
1917              timeout=15,
1918          )
1919          assert result.returncode == 0
1920          assert "migrate-legacy" in result.stdout
1921  
1922      def test_gateway_command_migrate_legacy_dispatches(
1923          self, tmp_path, monkeypatch, capsys
1924      ):
1925          """gateway_command(args) with subcmd='migrate-legacy' calls the helper."""
1926          called = {}
1927  
1928          def fake_remove(interactive=True, dry_run=False):
1929              called["interactive"] = interactive
1930              called["dry_run"] = dry_run
1931              return 0, []
1932  
1933          monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
1934          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
1935          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
1936  
1937          args = SimpleNamespace(
1938              gateway_command="migrate-legacy", dry_run=False, yes=True
1939          )
1940          gateway_cli.gateway_command(args)
1941  
1942          assert called == {"interactive": False, "dry_run": False}
1943  
1944  
1945  class TestGatewayStatusParser:
1946      def test_gateway_status_subparser_accepts_full_flag(self):
1947          import subprocess
1948          import sys
1949  
1950          result = subprocess.run(
1951              [sys.executable, "-m", "hermes_cli.main", "gateway", "status", "-l", "--help"],
1952              cwd=str(gateway_cli.PROJECT_ROOT),
1953              capture_output=True,
1954              text=True,
1955              timeout=15,
1956          )
1957  
1958          assert result.returncode == 0
1959          assert "unrecognized arguments" not in result.stderr
1960  
1961      def test_gateway_command_migrate_legacy_dry_run_passes_through(
1962          self, monkeypatch
1963      ):
1964          called = {}
1965  
1966          def fake_remove(interactive=True, dry_run=False):
1967              called["interactive"] = interactive
1968              called["dry_run"] = dry_run
1969              return 0, []
1970  
1971          monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
1972          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
1973          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
1974  
1975          args = SimpleNamespace(
1976              gateway_command="migrate-legacy", dry_run=True, yes=False
1977          )
1978          gateway_cli.gateway_command(args)
1979  
1980          assert called == {"interactive": True, "dry_run": True}
1981  
1982      def test_migrate_legacy_on_unsupported_platform_prints_message(
1983          self, monkeypatch, capsys
1984      ):
1985          monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
1986          monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
1987  
1988          args = SimpleNamespace(
1989              gateway_command="migrate-legacy", dry_run=False, yes=True
1990          )
1991          gateway_cli.gateway_command(args)
1992  
1993          out = capsys.readouterr().out
1994          assert "only applies to systemd" in out
1995  
1996  
1997  class TestSystemdInstallOffersLegacyRemoval:
1998      """Verify that systemd_install prompts to remove legacy units first."""
1999  
2000      def test_install_offers_removal_when_legacy_detected(
2001          self, tmp_path, monkeypatch, capsys
2002      ):
2003          """When legacy units exist, install flow should call the removal
2004          helper before writing the new unit."""
2005          remove_called = {}
2006  
2007          def fake_remove(interactive=True, dry_run=False):
2008              remove_called["invoked"] = True
2009              remove_called["interactive"] = interactive
2010              return 1, []
2011  
2012          # has_legacy_hermes_units must return True
2013          monkeypatch.setattr(gateway_cli, "has_legacy_hermes_units", lambda: True)
2014          monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
2015          monkeypatch.setattr(gateway_cli, "print_legacy_unit_warning", lambda: None)
2016          # Answer "yes" to the legacy-removal prompt
2017          monkeypatch.setattr(gateway_cli, "prompt_yes_no", lambda *a, **k: True)
2018  
2019          # Mock the rest of the install flow
2020          unit_path = tmp_path / "hermes-gateway.service"
2021          monkeypatch.setattr(
2022              gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path
2023          )
2024          monkeypatch.setattr(
2025              gateway_cli,
2026              "generate_systemd_unit",
2027              lambda system=False, run_as_user=None: "unit text\n",
2028          )
2029          monkeypatch.setattr(
2030              gateway_cli.subprocess,
2031              "run",
2032              lambda cmd, **kw: SimpleNamespace(returncode=0, stdout="", stderr=""),
2033          )
2034          monkeypatch.setattr(gateway_cli, "_ensure_linger_enabled", lambda: None)
2035  
2036          gateway_cli.systemd_install()
2037  
2038          assert remove_called.get("invoked") is True
2039          assert remove_called.get("interactive") is False  # prompted elsewhere
2040  
2041      def test_install_declines_legacy_removal_when_user_says_no(
2042          self, tmp_path, monkeypatch
2043      ):
2044          """When legacy units exist and user declines, install still proceeds
2045          but doesn't touch them."""
2046          remove_called = {"invoked": False}
2047  
2048          def fake_remove(interactive=True, dry_run=False):
2049              remove_called["invoked"] = True
2050              return 0, []
2051  
2052          monkeypatch.setattr(gateway_cli, "has_legacy_hermes_units", lambda: True)
2053          monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
2054          monkeypatch.setattr(gateway_cli, "print_legacy_unit_warning", lambda: None)
2055          monkeypatch.setattr(gateway_cli, "prompt_yes_no", lambda *a, **k: False)
2056  
2057          unit_path = tmp_path / "hermes-gateway.service"
2058          monkeypatch.setattr(
2059              gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path
2060          )
2061          monkeypatch.setattr(
2062              gateway_cli,
2063              "generate_systemd_unit",
2064              lambda system=False, run_as_user=None: "unit text\n",
2065          )
2066          monkeypatch.setattr(
2067              gateway_cli.subprocess,
2068              "run",
2069              lambda cmd, **kw: SimpleNamespace(returncode=0, stdout="", stderr=""),
2070          )
2071          monkeypatch.setattr(gateway_cli, "_ensure_linger_enabled", lambda: None)
2072  
2073          gateway_cli.systemd_install()
2074  
2075          # Helper must NOT have been called
2076          assert remove_called["invoked"] is False
2077          # New unit should still have been written
2078          assert unit_path.exists()
2079          assert unit_path.read_text() == "unit text\n"
2080  
2081      def test_install_skips_legacy_check_when_none_present(
2082          self, tmp_path, monkeypatch
2083      ):
2084          """No legacy → no prompt, no helper call."""
2085          prompt_called = {"count": 0}
2086  
2087          def counting_prompt(*a, **k):
2088              prompt_called["count"] += 1
2089              return True
2090  
2091          remove_called = {"invoked": False}
2092  
2093          def fake_remove(interactive=True, dry_run=False):
2094              remove_called["invoked"] = True
2095              return 0, []
2096  
2097          monkeypatch.setattr(gateway_cli, "has_legacy_hermes_units", lambda: False)
2098          monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
2099          monkeypatch.setattr(gateway_cli, "prompt_yes_no", counting_prompt)
2100  
2101          unit_path = tmp_path / "hermes-gateway.service"
2102          monkeypatch.setattr(
2103              gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path
2104          )
2105          monkeypatch.setattr(
2106              gateway_cli,
2107              "generate_systemd_unit",
2108              lambda system=False, run_as_user=None: "unit text\n",
2109          )
2110          monkeypatch.setattr(
2111              gateway_cli.subprocess,
2112              "run",
2113              lambda cmd, **kw: SimpleNamespace(returncode=0, stdout="", stderr=""),
2114          )
2115          monkeypatch.setattr(gateway_cli, "_ensure_linger_enabled", lambda: None)
2116  
2117          gateway_cli.systemd_install()
2118  
2119          assert prompt_called["count"] == 0
2120          assert remove_called["invoked"] is False