mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
2a98098035
Previously, systemd_restart() sent SIGUSR1 to the gateway, printed 'restart requested', and returned immediately. The gateway still needed to drain active agents, exit with code 75, wait for systemd's RestartSec=30, and start the new process. The user saw 'success' but the gateway was actually down for 30-60 seconds. Now the SIGUSR1 path blocks with progress feedback: Phase 1 — wait for old process to die: ⏳ User service draining active work... Polls os.kill(pid, 0) until ProcessLookupError (up to 90s) Phase 2 — wait for new process to become active: ⏳ Waiting for hermes-gateway to restart... Polls systemctl is-active + verifies new PID (up to 60s) Success: ✓ User service restarted (PID 12345) Timeout: ⚠ User service did not become active within 60s. Check status: hermes gateway status Check logs: journalctl --user -u hermes-gateway --since '2 min ago' The reload-or-restart fallback path (line 1189) already blocks because systemctl reload-or-restart is synchronous. Test plan: - Updated test to verify wait-for-restart behavior - All 118 gateway CLI tests pass
1144 lines
47 KiB
Python
1144 lines
47 KiB
Python
"""Tests for gateway service management helpers."""
|
|
|
|
import os
|
|
import pwd
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
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")
|
|
|
|
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")
|
|
|
|
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[:2] == [
|
|
["systemctl", "--user", "daemon-reload"],
|
|
["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
|
|
assert "TimeoutStopSec=60" 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
|
|
assert "TimeoutStopSec=60" 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 is-active returning "active" with a new PID
|
|
new_pid = [None]
|
|
def fake_subprocess_run(cmd, **kwargs):
|
|
if "is-active" in cmd:
|
|
result = SimpleNamespace(stdout="active\n", returncode=0)
|
|
new_pid[0] = 999 # new PID
|
|
return result
|
|
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
|
|
out = capsys.readouterr().out.lower()
|
|
assert "restarted" in out
|
|
|
|
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: calls.append((deep, system)))
|
|
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="status", deep=False, system=False))
|
|
|
|
assert calls == [(False, False)]
|
|
|
|
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.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.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.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.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 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
|