mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-27 01:11:40 +00:00
2eab04fd84
Mock _preflight_user_systemd and _select_systemd_scope in test_systemd_start_refreshes_outdated_unit and test_systemd_restart_refreshes_outdated_unit. These tests target unit-file refresh logic, not D-Bus reachability, so the preflight check was causing spurious UserSystemdUnavailableError on macOS, WSL, and Docker where systemd is unavailable.
2015 lines
83 KiB
Python
2015 lines
83 KiB
Python
"""Tests for gateway service management helpers."""
|
|
|
|
import os
|
|
import pwd
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import hermes_cli.gateway as gateway_cli
|
|
from gateway.restart import (
|
|
DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
|
|
GATEWAY_SERVICE_RESTART_EXIT_CODE,
|
|
)
|
|
|
|
|
|
class TestSystemdServiceRefresh:
|
|
def test_systemd_install_repairs_outdated_unit_without_force(self, tmp_path, monkeypatch):
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("old unit\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
|
|
monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=True, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.systemd_install()
|
|
|
|
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
|
|
assert calls[:2] == [
|
|
["systemctl", "--user", "daemon-reload"],
|
|
["systemctl", "--user", "enable", gateway_cli.get_service_name()],
|
|
]
|
|
|
|
def test_systemd_start_refreshes_outdated_unit(self, tmp_path, monkeypatch):
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("old unit\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
|
|
monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
|
|
# Bypass systemd availability checks — this test targets unit-file
|
|
# refresh logic, not D-Bus reachability (fails on macOS/WSL/Docker).
|
|
monkeypatch.setattr(gateway_cli, "_preflight_user_systemd", lambda **kw: None)
|
|
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=True, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.systemd_start()
|
|
|
|
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
|
|
assert calls[:2] == [
|
|
["systemctl", "--user", "daemon-reload"],
|
|
["systemctl", "--user", "start", gateway_cli.get_service_name()],
|
|
]
|
|
|
|
def test_systemd_restart_refreshes_outdated_unit(self, tmp_path, monkeypatch):
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("old unit\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
|
|
monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
|
|
# Bypass systemd availability checks — this test targets unit-file
|
|
# refresh logic, not D-Bus reachability (fails on macOS/WSL/Docker).
|
|
monkeypatch.setattr(gateway_cli, "_preflight_user_systemd", lambda **kw: None)
|
|
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=True, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.systemd_restart()
|
|
|
|
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
|
|
assert calls[:4] == [
|
|
["systemctl", "--user", "daemon-reload"],
|
|
["systemctl", "--user", "show", gateway_cli.get_service_name(), "--no-pager", "--property", "ActiveState,SubState,Result,ExecMainStatus"],
|
|
["systemctl", "--user", "reset-failed", gateway_cli.get_service_name()],
|
|
["systemctl", "--user", "reload-or-restart", gateway_cli.get_service_name()],
|
|
]
|
|
|
|
|
|
class TestGeneratedSystemdUnits:
|
|
def test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self):
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
|
|
assert "ExecStart=" in unit
|
|
assert "ExecStop=" not in unit
|
|
assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
|
|
assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
|
|
# TimeoutStopSec must exceed the default drain_timeout (60s) so
|
|
# systemd doesn't SIGKILL the cgroup before post-interrupt cleanup
|
|
# (tool subprocess kill, adapter disconnect) runs — issue #8202.
|
|
assert "TimeoutStopSec=90" in unit
|
|
|
|
def test_user_unit_includes_resolved_node_directory_in_path(self, monkeypatch):
|
|
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: "/home/test/.nvm/versions/node/v24.14.0/bin/node" if cmd == "node" else None)
|
|
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
|
|
assert "/home/test/.nvm/versions/node/v24.14.0/bin" in unit
|
|
|
|
def test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self):
|
|
unit = gateway_cli.generate_systemd_unit(system=True)
|
|
|
|
assert "ExecStart=" in unit
|
|
assert "ExecStop=" not in unit
|
|
assert "ExecReload=/bin/kill -USR1 $MAINPID" in unit
|
|
assert f"RestartForceExitStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}" in unit
|
|
# TimeoutStopSec must exceed the default drain_timeout (60s) so
|
|
# systemd doesn't SIGKILL the cgroup before post-interrupt cleanup
|
|
# (tool subprocess kill, adapter disconnect) runs — issue #8202.
|
|
assert "TimeoutStopSec=90" in unit
|
|
assert "WantedBy=multi-user.target" in unit
|
|
|
|
|
|
class TestGatewayStopCleanup:
|
|
def test_stop_only_kills_current_profile_by_default(self, tmp_path, monkeypatch):
|
|
"""Without --all, stop uses systemd (if available) and does NOT call
|
|
the global kill_gateway_processes()."""
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("unit\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
|
|
|
|
service_calls = []
|
|
kill_calls = []
|
|
|
|
monkeypatch.setattr(gateway_cli, "systemd_stop", lambda system=False: service_calls.append("stop"))
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"kill_gateway_processes",
|
|
lambda force=False, all_profiles=False: kill_calls.append(force) or 2,
|
|
)
|
|
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="stop"))
|
|
|
|
assert service_calls == ["stop"]
|
|
# Global kill should NOT be called without --all
|
|
assert kill_calls == []
|
|
|
|
def test_stop_all_sweeps_all_gateway_processes(self, tmp_path, monkeypatch):
|
|
"""With --all, stop uses systemd AND calls the global kill_gateway_processes()."""
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("unit\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
|
|
|
|
service_calls = []
|
|
kill_calls = []
|
|
|
|
monkeypatch.setattr(gateway_cli, "systemd_stop", lambda system=False: service_calls.append("stop"))
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"kill_gateway_processes",
|
|
lambda force=False, all_profiles=False: kill_calls.append(force) or 2,
|
|
)
|
|
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="stop", **{"all": True}))
|
|
|
|
assert service_calls == ["stop"]
|
|
assert kill_calls == [False]
|
|
|
|
|
|
class TestLaunchdServiceRecovery:
|
|
def test_get_restart_drain_timeout_prefers_env_then_config_then_default(self, monkeypatch):
|
|
monkeypatch.delenv("HERMES_RESTART_DRAIN_TIMEOUT", raising=False)
|
|
monkeypatch.setattr(gateway_cli, "read_raw_config", lambda: {})
|
|
|
|
assert (
|
|
gateway_cli._get_restart_drain_timeout()
|
|
== DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"read_raw_config",
|
|
lambda: {"agent": {"restart_drain_timeout": 14}},
|
|
)
|
|
assert gateway_cli._get_restart_drain_timeout() == 14.0
|
|
|
|
monkeypatch.setenv("HERMES_RESTART_DRAIN_TIMEOUT", "9")
|
|
assert gateway_cli._get_restart_drain_timeout() == 9.0
|
|
|
|
monkeypatch.setenv("HERMES_RESTART_DRAIN_TIMEOUT", "invalid")
|
|
assert (
|
|
gateway_cli._get_restart_drain_timeout()
|
|
== DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
|
|
)
|
|
|
|
def test_launchd_install_repairs_outdated_plist_without_force(self, tmp_path, monkeypatch):
|
|
plist_path = tmp_path / "ai.hermes.gateway.plist"
|
|
plist_path.write_text("<plist>old content</plist>", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.launchd_install()
|
|
|
|
label = gateway_cli.get_launchd_label()
|
|
domain = gateway_cli._launchd_domain()
|
|
assert "--replace" in plist_path.read_text(encoding="utf-8")
|
|
assert calls[:2] == [
|
|
["launchctl", "bootout", f"{domain}/{label}"],
|
|
["launchctl", "bootstrap", domain, str(plist_path)],
|
|
]
|
|
|
|
def test_launchd_start_reloads_unloaded_job_and_retries(self, tmp_path, monkeypatch):
|
|
plist_path = tmp_path / "ai.hermes.gateway.plist"
|
|
plist_path.write_text(gateway_cli.generate_launchd_plist(), encoding="utf-8")
|
|
label = gateway_cli.get_launchd_label()
|
|
|
|
calls = []
|
|
domain = gateway_cli._launchd_domain()
|
|
target = f"{domain}/{label}"
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
calls.append(cmd)
|
|
if cmd == ["launchctl", "kickstart", target] and calls.count(cmd) == 1:
|
|
raise gateway_cli.subprocess.CalledProcessError(3, cmd, stderr="Could not find service")
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.launchd_start()
|
|
|
|
assert calls == [
|
|
["launchctl", "kickstart", target],
|
|
["launchctl", "bootstrap", domain, str(plist_path)],
|
|
["launchctl", "kickstart", target],
|
|
]
|
|
|
|
def test_launchd_start_reloads_on_kickstart_exit_code_113(self, tmp_path, monkeypatch):
|
|
"""Exit code 113 (\"Could not find service\") should also trigger bootstrap recovery."""
|
|
plist_path = tmp_path / "ai.hermes.gateway.plist"
|
|
plist_path.write_text(gateway_cli.generate_launchd_plist(), encoding="utf-8")
|
|
label = gateway_cli.get_launchd_label()
|
|
|
|
calls = []
|
|
domain = gateway_cli._launchd_domain()
|
|
target = f"{domain}/{label}"
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
calls.append(cmd)
|
|
if cmd == ["launchctl", "kickstart", target] and calls.count(cmd) == 1:
|
|
raise gateway_cli.subprocess.CalledProcessError(113, cmd, stderr="Could not find service")
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.launchd_start()
|
|
|
|
assert calls == [
|
|
["launchctl", "kickstart", target],
|
|
["launchctl", "bootstrap", domain, str(plist_path)],
|
|
["launchctl", "kickstart", target],
|
|
]
|
|
|
|
def test_launchd_restart_drains_running_gateway_before_kickstart(self, monkeypatch):
|
|
calls = []
|
|
target = f"{gateway_cli._launchd_domain()}/{gateway_cli.get_launchd_label()}"
|
|
|
|
monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 12.0)
|
|
monkeypatch.setattr(gateway_cli, "_request_gateway_self_restart", lambda pid: False)
|
|
monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", lambda timeout, force_after=None: True)
|
|
monkeypatch.setattr(gateway_cli, "terminate_pid", lambda pid, force=False: calls.append(("term", pid, force)))
|
|
monkeypatch.setattr(
|
|
"gateway.status.get_running_pid",
|
|
lambda: 321,
|
|
)
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.launchd_restart()
|
|
|
|
assert calls == [
|
|
("term", 321, False),
|
|
["launchctl", "kickstart", "-k", target],
|
|
]
|
|
|
|
def test_launchd_restart_self_requests_graceful_restart_without_kickstart(self, monkeypatch, capsys):
|
|
calls = []
|
|
|
|
monkeypatch.setattr(
|
|
"gateway.status.get_running_pid",
|
|
lambda: 321,
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"_request_gateway_self_restart",
|
|
lambda pid: calls.append(("self", pid)) or True,
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli.subprocess,
|
|
"run",
|
|
lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("launchctl should not run")),
|
|
)
|
|
|
|
gateway_cli.launchd_restart()
|
|
|
|
assert calls == [("self", 321)]
|
|
assert "restart requested" in capsys.readouterr().out.lower()
|
|
|
|
def test_launchd_stop_uses_bootout_not_kill(self, monkeypatch):
|
|
"""launchd_stop must bootout the service so KeepAlive doesn't respawn it."""
|
|
label = gateway_cli.get_launchd_label()
|
|
domain = gateway_cli._launchd_domain()
|
|
target = f"{domain}/{label}"
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", lambda **kw: None)
|
|
|
|
gateway_cli.launchd_stop()
|
|
|
|
assert calls == [["launchctl", "bootout", target]]
|
|
|
|
def test_launchd_stop_tolerates_already_unloaded(self, monkeypatch, capsys):
|
|
"""launchd_stop silently handles exit codes 3/113 (job not loaded)."""
|
|
label = gateway_cli.get_launchd_label()
|
|
domain = gateway_cli._launchd_domain()
|
|
target = f"{domain}/{label}"
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
if "bootout" in cmd:
|
|
raise gateway_cli.subprocess.CalledProcessError(3, cmd, stderr="Could not find service")
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", lambda **kw: None)
|
|
|
|
# Should not raise — exit code 3 means already unloaded
|
|
gateway_cli.launchd_stop()
|
|
|
|
output = capsys.readouterr().out
|
|
assert "stopped" in output.lower()
|
|
|
|
def test_launchd_stop_waits_for_process_exit(self, monkeypatch):
|
|
"""launchd_stop calls _wait_for_gateway_exit after bootout."""
|
|
wait_called = []
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
def fake_wait(**kwargs):
|
|
wait_called.append(kwargs)
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", fake_wait)
|
|
|
|
gateway_cli.launchd_stop()
|
|
|
|
assert len(wait_called) == 1
|
|
assert wait_called[0] == {"timeout": 10.0, "force_after": 5.0}
|
|
|
|
def test_launchd_status_reports_local_stale_plist_when_unloaded(self, tmp_path, monkeypatch, capsys):
|
|
plist_path = tmp_path / "ai.hermes.gateway.plist"
|
|
plist_path.write_text("<plist>old content</plist>", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
monkeypatch.setattr(
|
|
gateway_cli.subprocess,
|
|
"run",
|
|
lambda *args, **kwargs: SimpleNamespace(returncode=113, stdout="", stderr="Could not find service"),
|
|
)
|
|
|
|
gateway_cli.launchd_status()
|
|
|
|
output = capsys.readouterr().out
|
|
assert str(plist_path) in output
|
|
assert "stale" in output.lower()
|
|
assert "not loaded" in output.lower()
|
|
|
|
|
|
class TestGatewayServiceDetection:
|
|
def test_supports_systemd_services_requires_systemctl_binary(self, monkeypatch):
|
|
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli.shutil, "which", lambda name: None)
|
|
|
|
assert gateway_cli.supports_systemd_services() is False
|
|
|
|
def test_supports_systemd_services_returns_true_when_systemctl_present(self, monkeypatch):
|
|
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
|
|
monkeypatch.setattr(gateway_cli.shutil, "which", lambda name: "/usr/bin/systemctl")
|
|
|
|
assert gateway_cli.supports_systemd_services() is True
|
|
|
|
def test_is_service_running_checks_system_scope_when_user_scope_is_inactive(self, monkeypatch):
|
|
user_unit = SimpleNamespace(exists=lambda: True)
|
|
system_unit = SimpleNamespace(exists=lambda: True)
|
|
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_systemd_unit_path",
|
|
lambda system=False: system_unit if system else user_unit,
|
|
)
|
|
|
|
def fake_run(cmd, capture_output=True, text=True, **kwargs):
|
|
if cmd == ["systemctl", "--user", "is-active", gateway_cli.get_service_name()]:
|
|
return SimpleNamespace(returncode=0, stdout="inactive\n", stderr="")
|
|
if cmd == ["systemctl", "is-active", gateway_cli.get_service_name()]:
|
|
return SimpleNamespace(returncode=0, stdout="active\n", stderr="")
|
|
raise AssertionError(f"Unexpected command: {cmd}")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
assert gateway_cli._is_service_running() is True
|
|
|
|
def test_is_service_running_returns_false_when_systemctl_missing(self, monkeypatch):
|
|
unit = SimpleNamespace(exists=lambda: True)
|
|
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_systemd_unit_path",
|
|
lambda system=False: unit,
|
|
)
|
|
|
|
def fake_run(*args, **kwargs):
|
|
raise FileNotFoundError("systemctl")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
assert gateway_cli._is_service_running() is False
|
|
|
|
class TestGatewaySystemServiceRouting:
|
|
def test_systemd_restart_self_requests_graceful_restart_and_waits(self, monkeypatch, capsys):
|
|
calls = []
|
|
|
|
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
|
|
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: calls.append(("refresh", system)))
|
|
monkeypatch.setattr(
|
|
"gateway.status.get_running_pid",
|
|
lambda: 654,
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"_request_gateway_self_restart",
|
|
lambda pid: calls.append(("self", pid)) or True,
|
|
)
|
|
|
|
# Simulate: old process dies immediately, new process becomes active
|
|
kill_call_count = [0]
|
|
def fake_kill(pid, sig):
|
|
kill_call_count[0] += 1
|
|
if kill_call_count[0] >= 2: # first call checks, second = dead
|
|
raise ProcessLookupError()
|
|
monkeypatch.setattr(os, "kill", fake_kill)
|
|
|
|
# Simulate systemctl reset-failed/start followed by an active unit
|
|
new_pid = [None]
|
|
def fake_subprocess_run(cmd, **kwargs):
|
|
if "reset-failed" in cmd:
|
|
calls.append(("reset-failed", cmd))
|
|
return SimpleNamespace(stdout="", returncode=0)
|
|
if "start" in cmd:
|
|
calls.append(("start", cmd))
|
|
return SimpleNamespace(stdout="", returncode=0)
|
|
if "show" in cmd:
|
|
new_pid[0] = 999
|
|
return SimpleNamespace(
|
|
stdout="ActiveState=active\nSubState=running\nResult=success\nExecMainStatus=0\n",
|
|
returncode=0,
|
|
)
|
|
raise AssertionError(f"Unexpected systemctl call: {cmd}")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_subprocess_run)
|
|
# get_running_pid returns new PID after restart
|
|
pid_calls = [0]
|
|
def fake_get_pid():
|
|
pid_calls[0] += 1
|
|
return 999 if pid_calls[0] > 1 else 654
|
|
monkeypatch.setattr("gateway.status.get_running_pid", fake_get_pid)
|
|
|
|
gateway_cli.systemd_restart()
|
|
|
|
assert ("self", 654) in calls
|
|
assert any(call[0] == "reset-failed" for call in calls)
|
|
assert any(call[0] == "start" for call in calls)
|
|
out = capsys.readouterr().out.lower()
|
|
assert "restarted" in out
|
|
|
|
def test_systemd_restart_recovers_failed_planned_restart(self, monkeypatch, capsys):
|
|
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
|
|
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
|
|
monkeypatch.setattr(
|
|
"gateway.status.read_runtime_status",
|
|
lambda: {"restart_requested": True, "gateway_state": "stopped"},
|
|
)
|
|
monkeypatch.setattr(gateway_cli, "_request_gateway_self_restart", lambda pid: False)
|
|
|
|
calls = []
|
|
started = {"value": False}
|
|
|
|
def fake_subprocess_run(cmd, **kwargs):
|
|
if "show" in cmd:
|
|
if not started["value"]:
|
|
return SimpleNamespace(
|
|
stdout=(
|
|
"ActiveState=failed\n"
|
|
"SubState=failed\n"
|
|
"Result=exit-code\n"
|
|
f"ExecMainStatus={GATEWAY_SERVICE_RESTART_EXIT_CODE}\n"
|
|
),
|
|
returncode=0,
|
|
)
|
|
return SimpleNamespace(
|
|
stdout="ActiveState=active\nSubState=running\nResult=success\nExecMainStatus=0\n",
|
|
returncode=0,
|
|
)
|
|
if "reset-failed" in cmd:
|
|
calls.append(("reset-failed", cmd))
|
|
return SimpleNamespace(stdout="", returncode=0)
|
|
if "start" in cmd:
|
|
started["value"] = True
|
|
calls.append(("start", cmd))
|
|
return SimpleNamespace(stdout="", returncode=0)
|
|
raise AssertionError(f"Unexpected command: {cmd}")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_subprocess_run)
|
|
monkeypatch.setattr(
|
|
"gateway.status.get_running_pid",
|
|
lambda: 999 if started["value"] else None,
|
|
)
|
|
|
|
gateway_cli.systemd_restart()
|
|
|
|
assert any(call[0] == "reset-failed" for call in calls)
|
|
assert any(call[0] == "start" for call in calls)
|
|
out = capsys.readouterr().out.lower()
|
|
assert "restarted" in out
|
|
|
|
def test_systemd_status_surfaces_planned_restart_failure(self, monkeypatch, capsys):
|
|
unit = SimpleNamespace(exists=lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
|
|
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit)
|
|
monkeypatch.setattr(gateway_cli, "has_conflicting_systemd_units", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "has_legacy_hermes_units", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "systemd_unit_is_current", lambda system=False: True)
|
|
monkeypatch.setattr(gateway_cli, "_runtime_health_lines", lambda: ["⚠ Last shutdown reason: Gateway restart requested"])
|
|
monkeypatch.setattr(gateway_cli, "get_systemd_linger_status", lambda: (True, ""))
|
|
monkeypatch.setattr(gateway_cli, "_read_systemd_unit_properties", lambda system=False: {
|
|
"ActiveState": "failed",
|
|
"SubState": "failed",
|
|
"Result": "exit-code",
|
|
"ExecMainStatus": str(GATEWAY_SERVICE_RESTART_EXIT_CODE),
|
|
})
|
|
|
|
calls = []
|
|
|
|
def fake_run_systemctl(args, **kwargs):
|
|
calls.append(args)
|
|
if args[:2] == ["status", gateway_cli.get_service_name()]:
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
if args[:2] == ["is-active", gateway_cli.get_service_name()]:
|
|
return SimpleNamespace(returncode=3, stdout="failed\n", stderr="")
|
|
raise AssertionError(f"Unexpected args: {args}")
|
|
|
|
monkeypatch.setattr(gateway_cli, "_run_systemctl", fake_run_systemctl)
|
|
|
|
gateway_cli.systemd_status()
|
|
|
|
out = capsys.readouterr().out
|
|
assert "Planned restart is stuck in systemd failed state" in out
|
|
|
|
def test_gateway_status_dispatches_full_flag(self, monkeypatch):
|
|
user_unit = SimpleNamespace(exists=lambda: True)
|
|
system_unit = SimpleNamespace(exists=lambda: False)
|
|
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_systemd_unit_path",
|
|
lambda system=False: system_unit if system else user_unit,
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_gateway_runtime_snapshot",
|
|
lambda system=False: gateway_cli.GatewayRuntimeSnapshot(
|
|
manager="systemd (user)",
|
|
service_installed=True,
|
|
service_running=False,
|
|
gateway_pids=(),
|
|
service_scope="user",
|
|
),
|
|
)
|
|
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"systemd_status",
|
|
lambda deep=False, system=False, full=False: calls.append((deep, system, full)),
|
|
)
|
|
|
|
gateway_cli.gateway_command(
|
|
SimpleNamespace(gateway_command="status", deep=False, system=False, full=True)
|
|
)
|
|
|
|
assert calls == [(False, False, True)]
|
|
|
|
def test_gateway_install_passes_system_flags(self, monkeypatch):
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"systemd_install",
|
|
lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)),
|
|
)
|
|
|
|
gateway_cli.gateway_command(
|
|
SimpleNamespace(gateway_command="install", force=True, system=True, run_as_user="alice")
|
|
)
|
|
|
|
assert calls == [(True, True, "alice")]
|
|
|
|
def test_gateway_install_reports_termux_manual_mode(self, monkeypatch, capsys):
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
|
|
try:
|
|
gateway_cli.gateway_command(
|
|
SimpleNamespace(gateway_command="install", force=False, system=False, run_as_user=None)
|
|
)
|
|
except SystemExit as exc:
|
|
assert exc.code == 1
|
|
else:
|
|
raise AssertionError("Expected gateway_command to exit on unsupported Termux service install")
|
|
|
|
out = capsys.readouterr().out
|
|
assert "not supported on Termux" in out
|
|
assert "Run manually: hermes gateway" in out
|
|
|
|
def test_gateway_status_prefers_system_service_when_only_system_unit_exists(self, monkeypatch):
|
|
user_unit = SimpleNamespace(exists=lambda: False)
|
|
system_unit = SimpleNamespace(exists=lambda: True)
|
|
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_systemd_unit_path",
|
|
lambda system=False: system_unit if system else user_unit,
|
|
)
|
|
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"systemd_status",
|
|
lambda deep=False, system=False, full=False: calls.append((deep, system, full)),
|
|
)
|
|
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="status", deep=False, system=False))
|
|
|
|
assert calls == [(False, False, False)]
|
|
|
|
def test_gateway_status_reports_manual_process_when_service_is_stopped(self, monkeypatch, capsys):
|
|
user_unit = SimpleNamespace(exists=lambda: True)
|
|
system_unit = SimpleNamespace(exists=lambda: False)
|
|
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_systemd_unit_path",
|
|
lambda system=False: system_unit if system else user_unit,
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"systemd_status",
|
|
lambda deep=False, system=False, full=False: print("service stopped"),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_gateway_runtime_snapshot",
|
|
lambda system=False: gateway_cli.GatewayRuntimeSnapshot(
|
|
manager="systemd (user)",
|
|
service_installed=True,
|
|
service_running=False,
|
|
gateway_pids=(4321,),
|
|
service_scope="user",
|
|
),
|
|
)
|
|
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="status", deep=False, system=False))
|
|
|
|
out = capsys.readouterr().out
|
|
assert "service stopped" in out
|
|
assert "Gateway process is running for this profile" in out
|
|
assert "PID(s): 4321" in out
|
|
|
|
def test_gateway_status_on_termux_shows_manual_guidance(self, monkeypatch, capsys):
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "find_gateway_pids", lambda exclude_pids=None: [])
|
|
monkeypatch.setattr(gateway_cli, "_runtime_health_lines", lambda: [])
|
|
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="status", deep=False, system=False))
|
|
|
|
out = capsys.readouterr().out
|
|
assert "Gateway is not running" in out
|
|
assert "nohup hermes gateway" in out
|
|
assert "install as user service" not in out
|
|
|
|
def test_gateway_restart_does_not_fallback_to_foreground_when_launchd_restart_fails(self, tmp_path, monkeypatch):
|
|
plist_path = tmp_path / "ai.hermes.gateway.plist"
|
|
plist_path.write_text("plist\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "is_linux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"launchd_restart",
|
|
lambda: (_ for _ in ()).throw(
|
|
gateway_cli.subprocess.CalledProcessError(5, ["launchctl", "kickstart", "-k", "gui/501/ai.hermes.gateway"])
|
|
),
|
|
)
|
|
|
|
run_calls = []
|
|
monkeypatch.setattr(gateway_cli, "run_gateway", lambda verbose=0, quiet=False, replace=False: run_calls.append((verbose, quiet, replace)))
|
|
monkeypatch.setattr(gateway_cli, "kill_gateway_processes", lambda force=False: 0)
|
|
|
|
try:
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="restart", system=False))
|
|
except SystemExit as exc:
|
|
assert exc.code == 1
|
|
else:
|
|
raise AssertionError("Expected gateway_command to exit when service restart fails")
|
|
|
|
assert run_calls == []
|
|
|
|
|
|
class TestDetectVenvDir:
|
|
"""Tests for _detect_venv_dir() virtualenv detection."""
|
|
|
|
def test_detects_active_virtualenv_via_sys_prefix(self, tmp_path, monkeypatch):
|
|
venv_path = tmp_path / "my-custom-venv"
|
|
venv_path.mkdir()
|
|
monkeypatch.setattr("sys.prefix", str(venv_path))
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result == venv_path
|
|
|
|
def test_falls_back_to_dot_venv_directory(self, tmp_path, monkeypatch):
|
|
# Not inside a virtualenv
|
|
monkeypatch.setattr("sys.prefix", "/usr")
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
|
|
|
|
dot_venv = tmp_path / ".venv"
|
|
dot_venv.mkdir()
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result == dot_venv
|
|
|
|
def test_falls_back_to_venv_directory(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("sys.prefix", "/usr")
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
|
|
|
|
venv = tmp_path / "venv"
|
|
venv.mkdir()
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result == venv
|
|
|
|
def test_prefers_dot_venv_over_venv(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("sys.prefix", "/usr")
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
|
|
|
|
(tmp_path / ".venv").mkdir()
|
|
(tmp_path / "venv").mkdir()
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result == tmp_path / ".venv"
|
|
|
|
def test_returns_none_when_no_virtualenv(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("sys.prefix", "/usr")
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result is None
|
|
|
|
|
|
class TestSystemUnitHermesHome:
|
|
"""HERMES_HOME in system units must reference the target user, not root."""
|
|
|
|
def test_system_unit_uses_target_user_home_not_calling_user(self, monkeypatch):
|
|
# Simulate sudo: Path.home() returns /root, target user is alice
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
|
|
monkeypatch.delenv("HERMES_HOME", raising=False)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_system_service_identity",
|
|
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_build_user_local_paths",
|
|
lambda home, existing: [],
|
|
)
|
|
|
|
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
|
|
|
|
assert 'HERMES_HOME=/home/alice/.hermes' in unit
|
|
assert '/root/.hermes' not in unit
|
|
|
|
def test_system_unit_remaps_profile_to_target_user(self, monkeypatch):
|
|
# Simulate sudo with a profile: HERMES_HOME was resolved under root
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
|
|
monkeypatch.setenv("HERMES_HOME", "/root/.hermes/profiles/coder")
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_system_service_identity",
|
|
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_build_user_local_paths",
|
|
lambda home, existing: [],
|
|
)
|
|
|
|
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
|
|
|
|
assert 'HERMES_HOME=/home/alice/.hermes/profiles/coder' in unit
|
|
assert '/root/' not in unit
|
|
|
|
def test_system_unit_preserves_custom_hermes_home(self, monkeypatch):
|
|
# Custom HERMES_HOME not under any user's home — keep as-is
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
|
|
monkeypatch.setenv("HERMES_HOME", "/opt/hermes-shared")
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_system_service_identity",
|
|
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_build_user_local_paths",
|
|
lambda home, existing: [],
|
|
)
|
|
|
|
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
|
|
|
|
assert 'HERMES_HOME=/opt/hermes-shared' in unit
|
|
|
|
def test_user_unit_unaffected_by_change(self):
|
|
# User-scope units should still use the calling user's HERMES_HOME
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
|
|
hermes_home = str(gateway_cli.get_hermes_home().resolve())
|
|
assert f'HERMES_HOME={hermes_home}' in unit
|
|
|
|
|
|
class TestHermesHomeForTargetUser:
|
|
"""Unit tests for _hermes_home_for_target_user()."""
|
|
|
|
def test_remaps_default_home(self, monkeypatch):
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
|
|
monkeypatch.delenv("HERMES_HOME", raising=False)
|
|
|
|
result = gateway_cli._hermes_home_for_target_user("/home/alice")
|
|
assert result == "/home/alice/.hermes"
|
|
|
|
def test_remaps_profile_path(self, monkeypatch):
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
|
|
monkeypatch.setenv("HERMES_HOME", "/root/.hermes/profiles/coder")
|
|
|
|
result = gateway_cli._hermes_home_for_target_user("/home/alice")
|
|
assert result == "/home/alice/.hermes/profiles/coder"
|
|
|
|
def test_keeps_custom_path(self, monkeypatch):
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
|
|
monkeypatch.setenv("HERMES_HOME", "/opt/hermes")
|
|
|
|
result = gateway_cli._hermes_home_for_target_user("/home/alice")
|
|
assert result == "/opt/hermes"
|
|
|
|
def test_noop_when_same_user(self, monkeypatch):
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/home/alice")))
|
|
monkeypatch.delenv("HERMES_HOME", raising=False)
|
|
|
|
result = gateway_cli._hermes_home_for_target_user("/home/alice")
|
|
assert result == "/home/alice/.hermes"
|
|
|
|
|
|
class TestGeneratedUnitUsesDetectedVenv:
|
|
def test_systemd_unit_uses_dot_venv_when_detected(self, tmp_path, monkeypatch):
|
|
dot_venv = tmp_path / ".venv"
|
|
dot_venv.mkdir()
|
|
(dot_venv / "bin").mkdir()
|
|
|
|
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: dot_venv)
|
|
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(dot_venv / "bin" / "python"))
|
|
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
|
|
assert f"VIRTUAL_ENV={dot_venv}" in unit
|
|
assert f"{dot_venv}/bin" in unit
|
|
# Must NOT contain a hardcoded /venv/ path
|
|
assert "/venv/" not in unit or "/.venv/" in unit
|
|
|
|
|
|
class TestGeneratedUnitIncludesLocalBin:
|
|
"""~/.local/bin must be in PATH so uvx/pipx tools are discoverable."""
|
|
|
|
def test_user_unit_includes_local_bin_in_path(self, monkeypatch):
|
|
home = Path.home()
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"_build_user_local_paths",
|
|
lambda home_path, existing: [str(home / ".local" / "bin")],
|
|
)
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
assert f"{home}/.local/bin" in unit
|
|
|
|
def test_system_unit_includes_local_bin_in_path(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"_build_user_local_paths",
|
|
lambda home_path, existing: [str(home_path / ".local" / "bin")],
|
|
)
|
|
unit = gateway_cli.generate_systemd_unit(system=True)
|
|
# System unit uses the resolved home dir from _system_service_identity
|
|
assert "/.local/bin" in unit
|
|
|
|
|
|
class TestSystemServiceIdentityRootHandling:
|
|
"""Root user handling in _system_service_identity()."""
|
|
|
|
def test_auto_detected_root_is_rejected(self, monkeypatch):
|
|
"""When root is auto-detected (not explicitly requested), raise."""
|
|
import pwd
|
|
import grp
|
|
|
|
monkeypatch.delenv("SUDO_USER", raising=False)
|
|
monkeypatch.setenv("USER", "root")
|
|
monkeypatch.setenv("LOGNAME", "root")
|
|
|
|
import pytest
|
|
with pytest.raises(ValueError, match="pass --run-as-user root to override"):
|
|
gateway_cli._system_service_identity(run_as_user=None)
|
|
|
|
def test_explicit_root_is_allowed(self, monkeypatch):
|
|
"""When root is explicitly passed via --run-as-user root, allow it."""
|
|
import pwd
|
|
import grp
|
|
|
|
root_info = pwd.getpwnam("root")
|
|
root_group = grp.getgrgid(root_info.pw_gid).gr_name
|
|
|
|
username, group, home = gateway_cli._system_service_identity(run_as_user="root")
|
|
assert username == "root"
|
|
assert home == root_info.pw_dir
|
|
|
|
def test_non_root_user_passes_through(self, monkeypatch):
|
|
"""Normal non-root user works as before."""
|
|
import pwd
|
|
import grp
|
|
|
|
monkeypatch.delenv("SUDO_USER", raising=False)
|
|
monkeypatch.setenv("USER", "nobody")
|
|
monkeypatch.setenv("LOGNAME", "nobody")
|
|
|
|
try:
|
|
username, group, home = gateway_cli._system_service_identity(run_as_user=None)
|
|
assert username == "nobody"
|
|
except ValueError as e:
|
|
# "nobody" might not exist on all systems
|
|
assert "Unknown user" in str(e)
|
|
|
|
|
|
class TestEnsureUserSystemdEnv:
|
|
"""Tests for _ensure_user_systemd_env() D-Bus session bus auto-detection."""
|
|
|
|
def test_sets_xdg_runtime_dir_when_missing(self, tmp_path, monkeypatch):
|
|
monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False)
|
|
monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
|
|
monkeypatch.setattr(os, "getuid", lambda: 42)
|
|
|
|
# Patch Path.exists so /run/user/42 appears to exist.
|
|
# Using a FakePath subclass breaks on Python 3.12+ where
|
|
# PosixPath.__new__ ignores the redirected path argument.
|
|
_orig_exists = gateway_cli.Path.exists
|
|
monkeypatch.setattr(
|
|
gateway_cli.Path, "exists",
|
|
lambda self: True if str(self) == "/run/user/42" else _orig_exists(self),
|
|
)
|
|
|
|
gateway_cli._ensure_user_systemd_env()
|
|
|
|
assert os.environ.get("XDG_RUNTIME_DIR") == "/run/user/42"
|
|
|
|
def test_sets_dbus_address_when_bus_socket_exists(self, tmp_path, monkeypatch):
|
|
runtime = tmp_path / "runtime"
|
|
runtime.mkdir()
|
|
bus_socket = runtime / "bus"
|
|
bus_socket.touch() # simulate the socket file
|
|
|
|
monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime))
|
|
monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
|
|
monkeypatch.setattr(os, "getuid", lambda: 99)
|
|
|
|
gateway_cli._ensure_user_systemd_env()
|
|
|
|
assert os.environ["DBUS_SESSION_BUS_ADDRESS"] == f"unix:path={bus_socket}"
|
|
|
|
def test_preserves_existing_env_vars(self, monkeypatch):
|
|
monkeypatch.setenv("XDG_RUNTIME_DIR", "/custom/runtime")
|
|
monkeypatch.setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/custom/bus")
|
|
|
|
gateway_cli._ensure_user_systemd_env()
|
|
|
|
assert os.environ["XDG_RUNTIME_DIR"] == "/custom/runtime"
|
|
assert os.environ["DBUS_SESSION_BUS_ADDRESS"] == "unix:path=/custom/bus"
|
|
|
|
def test_no_dbus_when_bus_socket_missing(self, tmp_path, monkeypatch):
|
|
runtime = tmp_path / "runtime"
|
|
runtime.mkdir()
|
|
# no bus socket created
|
|
|
|
monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime))
|
|
monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
|
|
monkeypatch.setattr(os, "getuid", lambda: 99)
|
|
|
|
gateway_cli._ensure_user_systemd_env()
|
|
|
|
assert "DBUS_SESSION_BUS_ADDRESS" not in os.environ
|
|
|
|
def test_systemctl_cmd_calls_ensure_for_user_mode(self, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("called"))
|
|
|
|
result = gateway_cli._systemctl_cmd(system=False)
|
|
assert result == ["systemctl", "--user"]
|
|
assert calls == ["called"]
|
|
|
|
def test_systemctl_cmd_skips_ensure_for_system_mode(self, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("called"))
|
|
|
|
result = gateway_cli._systemctl_cmd(system=True)
|
|
assert result == ["systemctl"]
|
|
assert calls == []
|
|
|
|
|
|
class TestPreflightUserSystemd:
|
|
"""Tests for _preflight_user_systemd() — D-Bus reachability before systemctl --user.
|
|
|
|
Covers issue #5130 / Rick's RHEL 9.6 SSH scenario: setup tries to start the
|
|
gateway via ``systemctl --user start`` in a shell with no user D-Bus session,
|
|
which previously failed with a raw ``CalledProcessError`` and no remediation.
|
|
"""
|
|
|
|
def test_noop_when_bus_socket_exists(self, monkeypatch):
|
|
"""Socket already there (desktop / linger + prior login) → no-op."""
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_user_dbus_socket_path",
|
|
lambda: type("P", (), {"exists": lambda self: True})(),
|
|
)
|
|
# Should not raise, no subprocess calls needed.
|
|
gateway_cli._preflight_user_systemd()
|
|
|
|
def test_raises_when_linger_disabled_and_loginctl_denied(self, monkeypatch):
|
|
"""Rick's scenario: no D-Bus, no linger, non-root SSH → clear error."""
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_user_dbus_socket_path",
|
|
lambda: type("P", (), {"exists": lambda self: False})(),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "get_systemd_linger_status", lambda: (False, ""),
|
|
)
|
|
monkeypatch.setattr(gateway_cli.shutil, "which", lambda _: "/usr/bin/loginctl")
|
|
|
|
class _Result:
|
|
returncode = 1
|
|
stdout = ""
|
|
stderr = "Interactive authentication required."
|
|
|
|
monkeypatch.setattr(
|
|
gateway_cli.subprocess, "run", lambda *a, **kw: _Result(),
|
|
)
|
|
|
|
with pytest.raises(gateway_cli.UserSystemdUnavailableError) as exc_info:
|
|
gateway_cli._preflight_user_systemd()
|
|
|
|
msg = str(exc_info.value)
|
|
assert "sudo loginctl enable-linger" in msg
|
|
assert "hermes gateway run" in msg # foreground fallback mentioned
|
|
assert "Interactive authentication required" in msg
|
|
|
|
def test_raises_when_loginctl_missing(self, monkeypatch):
|
|
"""No loginctl binary at all → suggest sudo install + manual fix."""
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_user_dbus_socket_path",
|
|
lambda: type("P", (), {"exists": lambda self: False})(),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "get_systemd_linger_status",
|
|
lambda: (None, "loginctl not found"),
|
|
)
|
|
monkeypatch.setattr(gateway_cli.shutil, "which", lambda _: None)
|
|
|
|
with pytest.raises(gateway_cli.UserSystemdUnavailableError) as exc_info:
|
|
gateway_cli._preflight_user_systemd()
|
|
|
|
assert "sudo loginctl enable-linger" in str(exc_info.value)
|
|
|
|
def test_linger_enabled_but_socket_still_missing(self, monkeypatch):
|
|
"""Edge case: linger says yes but the bus socket never came up."""
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_user_dbus_socket_path",
|
|
lambda: type("P", (), {"exists": lambda self: False})(),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "get_systemd_linger_status", lambda: (True, ""),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_wait_for_user_dbus_socket", lambda timeout=3.0: False,
|
|
)
|
|
|
|
with pytest.raises(gateway_cli.UserSystemdUnavailableError) as exc_info:
|
|
gateway_cli._preflight_user_systemd()
|
|
|
|
assert "linger is enabled" in str(exc_info.value)
|
|
|
|
def test_enable_linger_succeeds_and_socket_appears(self, monkeypatch, capsys):
|
|
"""Happy remediation path: polkit allows enable-linger, socket spawns."""
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_user_dbus_socket_path",
|
|
lambda: type("P", (), {"exists": lambda self: False})(),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "get_systemd_linger_status", lambda: (False, ""),
|
|
)
|
|
monkeypatch.setattr(gateway_cli.shutil, "which", lambda _: "/usr/bin/loginctl")
|
|
|
|
class _OkResult:
|
|
returncode = 0
|
|
stdout = ""
|
|
stderr = ""
|
|
|
|
monkeypatch.setattr(
|
|
gateway_cli.subprocess, "run", lambda *a, **kw: _OkResult(),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_wait_for_user_dbus_socket",
|
|
lambda timeout=5.0: True,
|
|
)
|
|
|
|
# Should not raise.
|
|
gateway_cli._preflight_user_systemd()
|
|
out = capsys.readouterr().out
|
|
assert "Enabled linger" in out
|
|
|
|
|
|
class TestProfileArg:
|
|
"""Tests for _profile_arg — returns '--profile <name>' for named profiles."""
|
|
|
|
def test_default_hermes_home_returns_empty(self, tmp_path, monkeypatch):
|
|
"""Default ~/.hermes should not produce a --profile flag."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
result = gateway_cli._profile_arg(str(hermes_home))
|
|
assert result == ""
|
|
|
|
def test_named_profile_returns_flag(self, tmp_path, monkeypatch):
|
|
"""~/.hermes/profiles/mybot should return '--profile mybot'."""
|
|
profile_dir = tmp_path / ".hermes" / "profiles" / "mybot"
|
|
profile_dir.mkdir(parents=True)
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
result = gateway_cli._profile_arg(str(profile_dir))
|
|
assert result == "--profile mybot"
|
|
|
|
def test_hash_path_returns_empty(self, tmp_path, monkeypatch):
|
|
"""Arbitrary non-profile HERMES_HOME should return empty string."""
|
|
custom_home = tmp_path / "custom" / "hermes"
|
|
custom_home.mkdir(parents=True)
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
result = gateway_cli._profile_arg(str(custom_home))
|
|
assert result == ""
|
|
|
|
def test_nested_profile_path_returns_empty(self, tmp_path, monkeypatch):
|
|
"""~/.hermes/profiles/mybot/subdir should NOT match — too deep."""
|
|
nested = tmp_path / ".hermes" / "profiles" / "mybot" / "subdir"
|
|
nested.mkdir(parents=True)
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
result = gateway_cli._profile_arg(str(nested))
|
|
assert result == ""
|
|
|
|
def test_invalid_profile_name_returns_empty(self, tmp_path, monkeypatch):
|
|
"""Profile names with invalid chars should not match the regex."""
|
|
bad_profile = tmp_path / ".hermes" / "profiles" / "My Bot!"
|
|
bad_profile.mkdir(parents=True)
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
result = gateway_cli._profile_arg(str(bad_profile))
|
|
assert result == ""
|
|
|
|
def test_systemd_unit_includes_profile(self, tmp_path, monkeypatch):
|
|
"""generate_systemd_unit should include --profile in ExecStart for named profiles."""
|
|
profile_dir = tmp_path / ".hermes" / "profiles" / "mybot"
|
|
profile_dir.mkdir(parents=True)
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
monkeypatch.setenv("HERMES_HOME", str(profile_dir))
|
|
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: profile_dir)
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
assert "--profile mybot" in unit
|
|
assert "gateway run --replace" in unit
|
|
|
|
def test_launchd_plist_includes_profile(self, tmp_path, monkeypatch):
|
|
"""generate_launchd_plist should include --profile in ProgramArguments for named profiles."""
|
|
profile_dir = tmp_path / ".hermes" / "profiles" / "mybot"
|
|
profile_dir.mkdir(parents=True)
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
monkeypatch.setenv("HERMES_HOME", str(profile_dir))
|
|
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: profile_dir)
|
|
plist = gateway_cli.generate_launchd_plist()
|
|
assert "<string>--profile</string>" in plist
|
|
assert "<string>mybot</string>" in plist
|
|
|
|
def test_launchd_plist_path_uses_real_user_home_not_profile_home(self, tmp_path, monkeypatch):
|
|
profile_dir = tmp_path / ".hermes" / "profiles" / "orcha"
|
|
profile_dir.mkdir(parents=True)
|
|
machine_home = tmp_path / "machine-home"
|
|
machine_home.mkdir()
|
|
profile_home = profile_dir / "home"
|
|
profile_home.mkdir()
|
|
|
|
monkeypatch.setattr(Path, "home", lambda: profile_home)
|
|
monkeypatch.setenv("HERMES_HOME", str(profile_dir))
|
|
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: profile_dir)
|
|
monkeypatch.setattr(pwd, "getpwuid", lambda uid: SimpleNamespace(pw_dir=str(machine_home)))
|
|
|
|
plist_path = gateway_cli.get_launchd_plist_path()
|
|
|
|
assert plist_path == machine_home / "Library" / "LaunchAgents" / "ai.hermes.gateway-orcha.plist"
|
|
|
|
|
|
class TestRemapPathForUser:
|
|
"""Unit tests for _remap_path_for_user()."""
|
|
|
|
def test_remaps_path_under_current_home(self, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path / "root")
|
|
(tmp_path / "root").mkdir()
|
|
result = gateway_cli._remap_path_for_user(
|
|
str(tmp_path / "root" / ".hermes" / "hermes-agent"),
|
|
str(tmp_path / "alice"),
|
|
)
|
|
assert result == str(tmp_path / "alice" / ".hermes" / "hermes-agent")
|
|
|
|
def test_keeps_system_path_unchanged(self, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path / "root")
|
|
(tmp_path / "root").mkdir()
|
|
result = gateway_cli._remap_path_for_user("/opt/hermes", str(tmp_path / "alice"))
|
|
assert result == "/opt/hermes"
|
|
|
|
def test_noop_when_same_user(self, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path / "alice")
|
|
(tmp_path / "alice").mkdir()
|
|
original = str(tmp_path / "alice" / ".hermes" / "hermes-agent")
|
|
result = gateway_cli._remap_path_for_user(original, str(tmp_path / "alice"))
|
|
assert result == original
|
|
|
|
|
|
class TestSystemUnitPathRemapping:
|
|
"""System units must remap ALL paths from the caller's home to the target user."""
|
|
|
|
def test_system_unit_has_no_root_paths(self, monkeypatch, tmp_path):
|
|
root_home = tmp_path / "root"
|
|
root_home.mkdir()
|
|
project = root_home / ".hermes" / "hermes-agent"
|
|
project.mkdir(parents=True)
|
|
venv_bin = project / "venv" / "bin"
|
|
venv_bin.mkdir(parents=True)
|
|
(venv_bin / "python").write_text("")
|
|
|
|
target_home = "/home/alice"
|
|
|
|
monkeypatch.setattr(Path, "home", lambda: root_home)
|
|
monkeypatch.setenv("HERMES_HOME", str(root_home / ".hermes"))
|
|
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: root_home / ".hermes")
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", project)
|
|
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: project / "venv")
|
|
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(venv_bin / "python"))
|
|
monkeypatch.setattr(
|
|
gateway_cli, "_system_service_identity",
|
|
lambda run_as_user=None: ("alice", "alice", target_home),
|
|
)
|
|
|
|
unit = gateway_cli.generate_systemd_unit(system=True)
|
|
|
|
# No root paths should leak into the unit
|
|
assert str(root_home) not in unit
|
|
# Target user paths should be present
|
|
assert "/home/alice" in unit
|
|
assert "WorkingDirectory=/home/alice/.hermes/hermes-agent" in unit
|
|
|
|
|
|
class TestDockerAwareGateway:
|
|
"""Tests for Docker container awareness in gateway commands."""
|
|
|
|
def test_run_systemctl_raises_runtimeerror_when_missing(self, monkeypatch):
|
|
"""_run_systemctl raises RuntimeError with container guidance when systemctl is absent."""
|
|
import pytest
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
raise FileNotFoundError("systemctl")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
with pytest.raises(RuntimeError, match="systemctl is not available"):
|
|
gateway_cli._run_systemctl(["start", "hermes-gateway"])
|
|
|
|
def test_run_systemctl_passes_through_on_success(self, monkeypatch):
|
|
"""_run_systemctl delegates to subprocess.run when systemctl exists."""
|
|
calls = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
result = gateway_cli._run_systemctl(["status", "hermes-gateway"])
|
|
assert result.returncode == 0
|
|
assert len(calls) == 1
|
|
assert "status" in calls[0]
|
|
|
|
def test_install_in_container_prints_docker_guidance(self, monkeypatch, capsys):
|
|
"""'hermes gateway install' inside Docker exits 0 with container guidance."""
|
|
import pytest
|
|
|
|
monkeypatch.setattr(gateway_cli, "is_managed", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
|
|
|
|
args = SimpleNamespace(gateway_command="install", force=False, system=False, run_as_user=None)
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
gateway_cli.gateway_command(args)
|
|
|
|
assert exc_info.value.code == 0
|
|
out = capsys.readouterr().out
|
|
assert "Docker" in out or "docker" in out
|
|
assert "restart" in out.lower()
|
|
|
|
def test_uninstall_in_container_prints_docker_guidance(self, monkeypatch, capsys):
|
|
"""'hermes gateway uninstall' inside Docker exits 0 with container guidance."""
|
|
import pytest
|
|
|
|
monkeypatch.setattr(gateway_cli, "is_managed", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
|
|
|
|
args = SimpleNamespace(gateway_command="uninstall", system=False)
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
gateway_cli.gateway_command(args)
|
|
|
|
assert exc_info.value.code == 0
|
|
out = capsys.readouterr().out
|
|
assert "docker" in out.lower()
|
|
|
|
def test_start_in_container_prints_docker_guidance(self, monkeypatch, capsys):
|
|
"""'hermes gateway start' inside Docker exits 0 with container guidance."""
|
|
import pytest
|
|
|
|
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
|
|
|
|
args = SimpleNamespace(gateway_command="start", system=False)
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
gateway_cli.gateway_command(args)
|
|
|
|
assert exc_info.value.code == 0
|
|
out = capsys.readouterr().out
|
|
assert "docker" in out.lower()
|
|
assert "hermes gateway run" in out
|
|
|
|
|
|
class TestLegacyHermesUnitDetection:
|
|
"""Tests for _find_legacy_hermes_units / has_legacy_hermes_units.
|
|
|
|
These guard against the scenario that tripped Luis in April 2026: an
|
|
older install left a ``hermes.service`` unit behind when the service was
|
|
renamed to ``hermes-gateway.service``. After PR #5646 (signal recovery
|
|
via systemd), the two services began SIGTERM-flapping over the same
|
|
Telegram bot token in a 30-second cycle.
|
|
|
|
The detector must flag ``hermes.service`` ONLY when it actually runs our
|
|
gateway, and must NEVER flag profile units
|
|
(``hermes-gateway-<profile>.service``) or unrelated third-party services.
|
|
"""
|
|
|
|
# Minimal ExecStart that looks like our gateway
|
|
_OUR_UNIT_TEXT = (
|
|
"[Unit]\nDescription=Hermes Gateway\n[Service]\n"
|
|
"ExecStart=/usr/bin/python -m hermes_cli.main gateway run --replace\n"
|
|
)
|
|
|
|
@staticmethod
|
|
def _setup_search_paths(tmp_path, monkeypatch):
|
|
"""Redirect the legacy search to user_dir + system_dir under tmp_path."""
|
|
user_dir = tmp_path / "user"
|
|
system_dir = tmp_path / "system"
|
|
user_dir.mkdir()
|
|
system_dir.mkdir()
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"_legacy_unit_search_paths",
|
|
lambda: [(False, user_dir), (True, system_dir)],
|
|
)
|
|
return user_dir, system_dir
|
|
|
|
def test_detects_legacy_hermes_service_in_user_scope(self, tmp_path, monkeypatch):
|
|
user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
|
|
legacy = user_dir / "hermes.service"
|
|
legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
results = gateway_cli._find_legacy_hermes_units()
|
|
|
|
assert len(results) == 1
|
|
name, path, is_system = results[0]
|
|
assert name == "hermes.service"
|
|
assert path == legacy
|
|
assert is_system is False
|
|
assert gateway_cli.has_legacy_hermes_units() is True
|
|
|
|
def test_detects_legacy_hermes_service_in_system_scope(self, tmp_path, monkeypatch):
|
|
_, system_dir = self._setup_search_paths(tmp_path, monkeypatch)
|
|
legacy = system_dir / "hermes.service"
|
|
legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
results = gateway_cli._find_legacy_hermes_units()
|
|
|
|
assert len(results) == 1
|
|
name, path, is_system = results[0]
|
|
assert name == "hermes.service"
|
|
assert path == legacy
|
|
assert is_system is True
|
|
|
|
def test_ignores_profile_unit_hermes_gateway_coder(self, tmp_path, monkeypatch):
|
|
"""CRITICAL: profile units must NOT be flagged as legacy.
|
|
|
|
Teknium's concern — ``hermes-gateway-coder.service`` is our standard
|
|
naming for the ``coder`` profile. The legacy detector is an explicit
|
|
allowlist, not a glob, so profile units are safe.
|
|
"""
|
|
user_dir, system_dir = self._setup_search_paths(tmp_path, monkeypatch)
|
|
# Drop profile units in BOTH scopes with our ExecStart
|
|
for base in (user_dir, system_dir):
|
|
(base / "hermes-gateway-coder.service").write_text(
|
|
self._OUR_UNIT_TEXT, encoding="utf-8"
|
|
)
|
|
(base / "hermes-gateway-orcha.service").write_text(
|
|
self._OUR_UNIT_TEXT, encoding="utf-8"
|
|
)
|
|
(base / "hermes-gateway.service").write_text(
|
|
self._OUR_UNIT_TEXT, encoding="utf-8"
|
|
)
|
|
|
|
results = gateway_cli._find_legacy_hermes_units()
|
|
|
|
assert results == []
|
|
assert gateway_cli.has_legacy_hermes_units() is False
|
|
|
|
def test_ignores_unrelated_hermes_service(self, tmp_path, monkeypatch):
|
|
"""Third-party ``hermes.service`` that isn't ours stays untouched.
|
|
|
|
If a user has some other package named ``hermes`` installed as a
|
|
service, we must not flag it.
|
|
"""
|
|
user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
|
|
(user_dir / "hermes.service").write_text(
|
|
"[Unit]\nDescription=Some Other Hermes\n[Service]\n"
|
|
"ExecStart=/opt/other-hermes/bin/daemon --foreground\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
results = gateway_cli._find_legacy_hermes_units()
|
|
|
|
assert results == []
|
|
assert gateway_cli.has_legacy_hermes_units() is False
|
|
|
|
def test_returns_empty_when_no_legacy_files_exist(self, tmp_path, monkeypatch):
|
|
self._setup_search_paths(tmp_path, monkeypatch)
|
|
|
|
assert gateway_cli._find_legacy_hermes_units() == []
|
|
assert gateway_cli.has_legacy_hermes_units() is False
|
|
|
|
def test_detects_both_scopes_simultaneously(self, tmp_path, monkeypatch):
|
|
"""When a user has BOTH user-scope and system-scope legacy units,
|
|
both are reported so the migration step can remove them together."""
|
|
user_dir, system_dir = self._setup_search_paths(tmp_path, monkeypatch)
|
|
(user_dir / "hermes.service").write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
(system_dir / "hermes.service").write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
results = gateway_cli._find_legacy_hermes_units()
|
|
|
|
scopes = sorted(is_system for _, _, is_system in results)
|
|
assert scopes == [False, True]
|
|
|
|
def test_accepts_alternate_execstart_formats(self, tmp_path, monkeypatch):
|
|
"""Older installs may have used different python invocations.
|
|
|
|
ExecStart variants we've seen in the wild:
|
|
- python -m hermes_cli.main gateway run
|
|
- python path/to/hermes_cli/main.py gateway run
|
|
- hermes gateway run (direct binary)
|
|
- python path/to/gateway/run.py
|
|
"""
|
|
user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
|
|
variants = [
|
|
"ExecStart=/venv/bin/python -m hermes_cli.main gateway run --replace",
|
|
"ExecStart=/venv/bin/python /opt/hermes/hermes_cli/main.py gateway run",
|
|
"ExecStart=/usr/local/bin/hermes gateway run --replace",
|
|
"ExecStart=/venv/bin/python /opt/hermes/gateway/run.py",
|
|
]
|
|
for i, execstart in enumerate(variants):
|
|
name = f"hermes.service" if i == 0 else f"hermes.service" # same name
|
|
# Test each variant fresh
|
|
(user_dir / "hermes.service").write_text(
|
|
f"[Unit]\nDescription=Old Hermes\n[Service]\n{execstart}\n",
|
|
encoding="utf-8",
|
|
)
|
|
results = gateway_cli._find_legacy_hermes_units()
|
|
assert len(results) == 1, f"Variant {i} not detected: {execstart!r}"
|
|
|
|
def test_print_legacy_unit_warning_is_noop_when_empty(self, tmp_path, monkeypatch, capsys):
|
|
self._setup_search_paths(tmp_path, monkeypatch)
|
|
|
|
gateway_cli.print_legacy_unit_warning()
|
|
out = capsys.readouterr().out
|
|
|
|
assert out == ""
|
|
|
|
def test_print_legacy_unit_warning_shows_migration_hint(self, tmp_path, monkeypatch, capsys):
|
|
user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
|
|
(user_dir / "hermes.service").write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
gateway_cli.print_legacy_unit_warning()
|
|
out = capsys.readouterr().out
|
|
|
|
assert "Legacy" in out
|
|
assert "hermes.service" in out
|
|
assert "hermes gateway migrate-legacy" in out
|
|
|
|
def test_handles_unreadable_unit_file_gracefully(self, tmp_path, monkeypatch):
|
|
"""A permission error reading a unit file must not crash detection."""
|
|
user_dir, _ = self._setup_search_paths(tmp_path, monkeypatch)
|
|
unreadable = user_dir / "hermes.service"
|
|
unreadable.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
# Simulate a read failure — monkeypatch Path.read_text to raise
|
|
original_read_text = gateway_cli.Path.read_text
|
|
|
|
def raising_read_text(self, *args, **kwargs):
|
|
if self == unreadable:
|
|
raise PermissionError("simulated")
|
|
return original_read_text(self, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(gateway_cli.Path, "read_text", raising_read_text)
|
|
|
|
# Should not raise
|
|
results = gateway_cli._find_legacy_hermes_units()
|
|
assert results == []
|
|
|
|
|
|
class TestRemoveLegacyHermesUnits:
|
|
"""Tests for remove_legacy_hermes_units (the migration action)."""
|
|
|
|
_OUR_UNIT_TEXT = (
|
|
"[Unit]\nDescription=Hermes Gateway\n[Service]\n"
|
|
"ExecStart=/usr/bin/python -m hermes_cli.main gateway run --replace\n"
|
|
)
|
|
|
|
@staticmethod
|
|
def _setup(tmp_path, monkeypatch, as_root=False):
|
|
user_dir = tmp_path / "user"
|
|
system_dir = tmp_path / "system"
|
|
user_dir.mkdir()
|
|
system_dir.mkdir()
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"_legacy_unit_search_paths",
|
|
lambda: [(False, user_dir), (True, system_dir)],
|
|
)
|
|
# Mock systemctl — return success for everything
|
|
systemctl_calls: list[list[str]] = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
systemctl_calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 0 if as_root else 1000)
|
|
return user_dir, system_dir, systemctl_calls
|
|
|
|
def test_returns_zero_when_no_legacy_units(self, tmp_path, monkeypatch, capsys):
|
|
self._setup(tmp_path, monkeypatch)
|
|
|
|
removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
|
|
|
|
assert removed == 0
|
|
assert remaining == []
|
|
assert "No legacy" in capsys.readouterr().out
|
|
|
|
def test_dry_run_lists_without_removing(self, tmp_path, monkeypatch, capsys):
|
|
user_dir, _, calls = self._setup(tmp_path, monkeypatch)
|
|
legacy = user_dir / "hermes.service"
|
|
legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
removed, remaining = gateway_cli.remove_legacy_hermes_units(
|
|
interactive=False, dry_run=True
|
|
)
|
|
|
|
assert removed == 0
|
|
assert remaining == [legacy]
|
|
assert legacy.exists() # Not removed
|
|
assert calls == [] # No systemctl invocations
|
|
out = capsys.readouterr().out
|
|
assert "dry-run" in out
|
|
|
|
def test_removes_user_scope_legacy_unit(self, tmp_path, monkeypatch, capsys):
|
|
user_dir, _, calls = self._setup(tmp_path, monkeypatch)
|
|
legacy = user_dir / "hermes.service"
|
|
legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
|
|
|
|
assert removed == 1
|
|
assert remaining == []
|
|
assert not legacy.exists()
|
|
# Must have invoked stop → disable → daemon-reload on user scope
|
|
cmds_joined = [" ".join(c) for c in calls]
|
|
assert any("--user stop hermes.service" in c for c in cmds_joined)
|
|
assert any("--user disable hermes.service" in c for c in cmds_joined)
|
|
assert any("--user daemon-reload" in c for c in cmds_joined)
|
|
|
|
def test_system_scope_without_root_defers_removal(self, tmp_path, monkeypatch, capsys):
|
|
_, system_dir, calls = self._setup(tmp_path, monkeypatch, as_root=False)
|
|
legacy = system_dir / "hermes.service"
|
|
legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
|
|
|
|
assert removed == 0
|
|
assert remaining == [legacy]
|
|
assert legacy.exists() # Not removed — requires sudo
|
|
out = capsys.readouterr().out
|
|
assert "sudo hermes gateway migrate-legacy" in out
|
|
|
|
def test_system_scope_with_root_removes(self, tmp_path, monkeypatch, capsys):
|
|
_, system_dir, calls = self._setup(tmp_path, monkeypatch, as_root=True)
|
|
legacy = system_dir / "hermes.service"
|
|
legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
|
|
|
|
assert removed == 1
|
|
assert remaining == []
|
|
assert not legacy.exists()
|
|
cmds_joined = [" ".join(c) for c in calls]
|
|
# System-scope uses plain "systemctl" (no --user)
|
|
assert any(
|
|
c.startswith("systemctl stop hermes.service") for c in cmds_joined
|
|
)
|
|
assert any(
|
|
c.startswith("systemctl disable hermes.service") for c in cmds_joined
|
|
)
|
|
|
|
def test_removes_both_scopes_with_root(self, tmp_path, monkeypatch, capsys):
|
|
user_dir, system_dir, _ = self._setup(tmp_path, monkeypatch, as_root=True)
|
|
user_legacy = user_dir / "hermes.service"
|
|
system_legacy = system_dir / "hermes.service"
|
|
user_legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
system_legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
|
|
|
|
assert removed == 2
|
|
assert remaining == []
|
|
assert not user_legacy.exists()
|
|
assert not system_legacy.exists()
|
|
|
|
def test_does_not_touch_profile_units_during_migration(
|
|
self, tmp_path, monkeypatch, capsys
|
|
):
|
|
"""Teknium's constraint: profile units (hermes-gateway-coder.service)
|
|
must survive a migration call, even if we somehow include them in the
|
|
search dir."""
|
|
user_dir, _, _ = self._setup(tmp_path, monkeypatch, as_root=True)
|
|
profile_unit = user_dir / "hermes-gateway-coder.service"
|
|
profile_unit.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
default_unit = user_dir / "hermes-gateway.service"
|
|
default_unit.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=False)
|
|
|
|
assert removed == 0
|
|
assert remaining == []
|
|
# Both the profile unit and the current default unit must survive
|
|
assert profile_unit.exists()
|
|
assert default_unit.exists()
|
|
|
|
def test_interactive_prompt_no_skips_removal(self, tmp_path, monkeypatch, capsys):
|
|
"""When interactive=True and user answers no, no removal happens."""
|
|
user_dir, _, _ = self._setup(tmp_path, monkeypatch)
|
|
legacy = user_dir / "hermes.service"
|
|
legacy.write_text(self._OUR_UNIT_TEXT, encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "prompt_yes_no", lambda *a, **k: False)
|
|
|
|
removed, remaining = gateway_cli.remove_legacy_hermes_units(interactive=True)
|
|
|
|
assert removed == 0
|
|
assert remaining == [legacy]
|
|
assert legacy.exists()
|
|
|
|
|
|
class TestMigrateLegacyCommand:
|
|
"""Tests for the `hermes gateway migrate-legacy` subcommand dispatch."""
|
|
|
|
def test_migrate_legacy_subparser_accepts_dry_run_and_yes(self):
|
|
"""Verify the argparse subparser is registered and parses flags."""
|
|
import hermes_cli.main as cli_main
|
|
|
|
parser = cli_main.build_parser() if hasattr(cli_main, "build_parser") else None
|
|
# Fall back to calling main's setup helper if direct access isn't exposed
|
|
# The key thing: the subparser must exist. We verify by constructing
|
|
# a namespace through argparse directly — but if build_parser isn't
|
|
# public, just confirm that `hermes gateway --help` shows it.
|
|
import subprocess
|
|
import sys
|
|
|
|
project_root = cli_main.PROJECT_ROOT if hasattr(cli_main, "PROJECT_ROOT") else None
|
|
if project_root is None:
|
|
import hermes_cli.gateway as gw
|
|
project_root = gw.PROJECT_ROOT
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "hermes_cli.main", "gateway", "--help"],
|
|
cwd=str(project_root),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
)
|
|
assert result.returncode == 0
|
|
assert "migrate-legacy" in result.stdout
|
|
|
|
def test_gateway_command_migrate_legacy_dispatches(
|
|
self, tmp_path, monkeypatch, capsys
|
|
):
|
|
"""gateway_command(args) with subcmd='migrate-legacy' calls the helper."""
|
|
called = {}
|
|
|
|
def fake_remove(interactive=True, dry_run=False):
|
|
called["interactive"] = interactive
|
|
called["dry_run"] = dry_run
|
|
return 0, []
|
|
|
|
monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
|
|
args = SimpleNamespace(
|
|
gateway_command="migrate-legacy", dry_run=False, yes=True
|
|
)
|
|
gateway_cli.gateway_command(args)
|
|
|
|
assert called == {"interactive": False, "dry_run": False}
|
|
|
|
|
|
class TestGatewayStatusParser:
|
|
def test_gateway_status_subparser_accepts_full_flag(self):
|
|
import subprocess
|
|
import sys
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "hermes_cli.main", "gateway", "status", "-l", "--help"],
|
|
cwd=str(gateway_cli.PROJECT_ROOT),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert "unrecognized arguments" not in result.stderr
|
|
|
|
def test_gateway_command_migrate_legacy_dry_run_passes_through(
|
|
self, monkeypatch
|
|
):
|
|
called = {}
|
|
|
|
def fake_remove(interactive=True, dry_run=False):
|
|
called["interactive"] = interactive
|
|
called["dry_run"] = dry_run
|
|
return 0, []
|
|
|
|
monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
|
|
args = SimpleNamespace(
|
|
gateway_command="migrate-legacy", dry_run=True, yes=False
|
|
)
|
|
gateway_cli.gateway_command(args)
|
|
|
|
assert called == {"interactive": True, "dry_run": True}
|
|
|
|
def test_migrate_legacy_on_unsupported_platform_prints_message(
|
|
self, monkeypatch, capsys
|
|
):
|
|
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
|
|
args = SimpleNamespace(
|
|
gateway_command="migrate-legacy", dry_run=False, yes=True
|
|
)
|
|
gateway_cli.gateway_command(args)
|
|
|
|
out = capsys.readouterr().out
|
|
assert "only applies to systemd" in out
|
|
|
|
|
|
class TestSystemdInstallOffersLegacyRemoval:
|
|
"""Verify that systemd_install prompts to remove legacy units first."""
|
|
|
|
def test_install_offers_removal_when_legacy_detected(
|
|
self, tmp_path, monkeypatch, capsys
|
|
):
|
|
"""When legacy units exist, install flow should call the removal
|
|
helper before writing the new unit."""
|
|
remove_called = {}
|
|
|
|
def fake_remove(interactive=True, dry_run=False):
|
|
remove_called["invoked"] = True
|
|
remove_called["interactive"] = interactive
|
|
return 1, []
|
|
|
|
# has_legacy_hermes_units must return True
|
|
monkeypatch.setattr(gateway_cli, "has_legacy_hermes_units", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
|
|
monkeypatch.setattr(gateway_cli, "print_legacy_unit_warning", lambda: None)
|
|
# Answer "yes" to the legacy-removal prompt
|
|
monkeypatch.setattr(gateway_cli, "prompt_yes_no", lambda *a, **k: True)
|
|
|
|
# Mock the rest of the install flow
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
monkeypatch.setattr(
|
|
gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"generate_systemd_unit",
|
|
lambda system=False, run_as_user=None: "unit text\n",
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli.subprocess,
|
|
"run",
|
|
lambda cmd, **kw: SimpleNamespace(returncode=0, stdout="", stderr=""),
|
|
)
|
|
monkeypatch.setattr(gateway_cli, "_ensure_linger_enabled", lambda: None)
|
|
|
|
gateway_cli.systemd_install()
|
|
|
|
assert remove_called.get("invoked") is True
|
|
assert remove_called.get("interactive") is False # prompted elsewhere
|
|
|
|
def test_install_declines_legacy_removal_when_user_says_no(
|
|
self, tmp_path, monkeypatch
|
|
):
|
|
"""When legacy units exist and user declines, install still proceeds
|
|
but doesn't touch them."""
|
|
remove_called = {"invoked": False}
|
|
|
|
def fake_remove(interactive=True, dry_run=False):
|
|
remove_called["invoked"] = True
|
|
return 0, []
|
|
|
|
monkeypatch.setattr(gateway_cli, "has_legacy_hermes_units", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
|
|
monkeypatch.setattr(gateway_cli, "print_legacy_unit_warning", lambda: None)
|
|
monkeypatch.setattr(gateway_cli, "prompt_yes_no", lambda *a, **k: False)
|
|
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
monkeypatch.setattr(
|
|
gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"generate_systemd_unit",
|
|
lambda system=False, run_as_user=None: "unit text\n",
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli.subprocess,
|
|
"run",
|
|
lambda cmd, **kw: SimpleNamespace(returncode=0, stdout="", stderr=""),
|
|
)
|
|
monkeypatch.setattr(gateway_cli, "_ensure_linger_enabled", lambda: None)
|
|
|
|
gateway_cli.systemd_install()
|
|
|
|
# Helper must NOT have been called
|
|
assert remove_called["invoked"] is False
|
|
# New unit should still have been written
|
|
assert unit_path.exists()
|
|
assert unit_path.read_text() == "unit text\n"
|
|
|
|
def test_install_skips_legacy_check_when_none_present(
|
|
self, tmp_path, monkeypatch
|
|
):
|
|
"""No legacy → no prompt, no helper call."""
|
|
prompt_called = {"count": 0}
|
|
|
|
def counting_prompt(*a, **k):
|
|
prompt_called["count"] += 1
|
|
return True
|
|
|
|
remove_called = {"invoked": False}
|
|
|
|
def fake_remove(interactive=True, dry_run=False):
|
|
remove_called["invoked"] = True
|
|
return 0, []
|
|
|
|
monkeypatch.setattr(gateway_cli, "has_legacy_hermes_units", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "remove_legacy_hermes_units", fake_remove)
|
|
monkeypatch.setattr(gateway_cli, "prompt_yes_no", counting_prompt)
|
|
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
monkeypatch.setattr(
|
|
gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"generate_systemd_unit",
|
|
lambda system=False, run_as_user=None: "unit text\n",
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway_cli.subprocess,
|
|
"run",
|
|
lambda cmd, **kw: SimpleNamespace(returncode=0, stdout="", stderr=""),
|
|
)
|
|
monkeypatch.setattr(gateway_cli, "_ensure_linger_enabled", lambda: None)
|
|
|
|
gateway_cli.systemd_install()
|
|
|
|
assert prompt_called["count"] == 0
|
|
assert remove_called["invoked"] is False
|