Compare commits

...

1 Commits

Author SHA1 Message Date
teknium1
6f46f84b26 feat: add system gateway service mode 2026-03-14 20:50:26 -07:00
5 changed files with 314 additions and 97 deletions

View File

@@ -123,10 +123,61 @@ SERVICE_NAME = "hermes-gateway"
SERVICE_DESCRIPTION = "Hermes Agent Gateway - Messaging Platform Integration" SERVICE_DESCRIPTION = "Hermes Agent Gateway - Messaging Platform Integration"
def get_systemd_unit_path() -> Path: def get_systemd_unit_path(system: bool = False) -> Path:
if system:
return Path("/etc/systemd/system") / f"{SERVICE_NAME}.service"
return Path.home() / ".config" / "systemd" / "user" / f"{SERVICE_NAME}.service" return Path.home() / ".config" / "systemd" / "user" / f"{SERVICE_NAME}.service"
def _systemctl_cmd(system: bool = False) -> list[str]:
return ["systemctl"] if system else ["systemctl", "--user"]
def _journalctl_cmd(system: bool = False) -> list[str]:
return ["journalctl"] if system else ["journalctl", "--user"]
def _service_scope_label(system: bool = False) -> str:
return "system" if system else "user"
def _require_root_for_system_service(action: str) -> None:
if os.geteuid() != 0:
print(f"System gateway {action} requires root. Re-run with sudo.")
sys.exit(1)
def _system_service_identity(run_as_user: str | None = None) -> tuple[str, str, str]:
import getpass
import grp
import pwd
username = (run_as_user or os.getenv("SUDO_USER") or os.getenv("USER") or os.getenv("LOGNAME") or getpass.getuser()).strip()
if not username:
raise ValueError("Could not determine which user the gateway service should run as")
if username == "root":
raise ValueError("Refusing to install the gateway system service as root; pass --run-as USER")
try:
user_info = pwd.getpwnam(username)
except KeyError as e:
raise ValueError(f"Unknown user: {username}") from e
group_name = grp.getgrgid(user_info.pw_gid).gr_name
return username, group_name, user_info.pw_dir
def _read_systemd_user_from_unit(unit_path: Path) -> str | None:
if not unit_path.exists():
return None
for line in unit_path.read_text(encoding="utf-8").splitlines():
if line.startswith("User="):
value = line.split("=", 1)[1].strip()
return value or None
return None
def get_systemd_linger_status() -> tuple[bool | None, str]: def get_systemd_linger_status() -> tuple[bool | None, str]:
"""Return whether systemd user lingering is enabled for the current user. """Return whether systemd user lingering is enabled for the current user.
@@ -216,8 +267,9 @@ def get_hermes_cli_path() -> str:
# Systemd (Linux) # Systemd (Linux)
# ============================================================================= # =============================================================================
def generate_systemd_unit() -> str: def generate_systemd_unit(system: bool = False, run_as_user: str | None = None) -> str:
import shutil import shutil
python_path = get_python_path() python_path = get_python_path()
working_dir = str(PROJECT_ROOT) working_dir = str(PROJECT_ROOT)
venv_dir = str(PROJECT_ROOT / "venv") venv_dir = str(PROJECT_ROOT / "venv")
@@ -226,8 +278,38 @@ def generate_systemd_unit() -> str:
# Build a PATH that includes the venv, node_modules, and standard system dirs # Build a PATH that includes the venv, node_modules, and standard system dirs
sane_path = f"{venv_bin}:{node_bin}:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" sane_path = f"{venv_bin}:{node_bin}:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
hermes_cli = shutil.which("hermes") or f"{python_path} -m hermes_cli.main" hermes_cli = shutil.which("hermes") or f"{python_path} -m hermes_cli.main"
if system:
username, group_name, home_dir = _system_service_identity(run_as_user)
return f"""[Unit]
Description={SERVICE_DESCRIPTION}
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User={username}
Group={group_name}
ExecStart={python_path} -m hermes_cli.main gateway run --replace
WorkingDirectory={working_dir}
Environment="HOME={home_dir}"
Environment="USER={username}"
Environment="LOGNAME={username}"
Environment="PATH={sane_path}"
Environment="VIRTUAL_ENV={venv_dir}"
Restart=on-failure
RestartSec=10
KillMode=mixed
KillSignal=SIGTERM
TimeoutStopSec=15
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
"""
return f"""[Unit] return f"""[Unit]
Description={SERVICE_DESCRIPTION} Description={SERVICE_DESCRIPTION}
After=network.target After=network.target
@@ -255,26 +337,28 @@ def _normalize_service_definition(text: str) -> str:
return "\n".join(line.rstrip() for line in text.strip().splitlines()) return "\n".join(line.rstrip() for line in text.strip().splitlines())
def systemd_unit_is_current() -> bool: def systemd_unit_is_current(system: bool = False) -> bool:
unit_path = get_systemd_unit_path() unit_path = get_systemd_unit_path(system=system)
if not unit_path.exists(): if not unit_path.exists():
return False return False
installed = unit_path.read_text(encoding="utf-8") installed = unit_path.read_text(encoding="utf-8")
expected = generate_systemd_unit() expected_user = _read_systemd_user_from_unit(unit_path) if system else None
expected = generate_systemd_unit(system=system, run_as_user=expected_user)
return _normalize_service_definition(installed) == _normalize_service_definition(expected) return _normalize_service_definition(installed) == _normalize_service_definition(expected)
def refresh_systemd_unit_if_needed() -> bool: def refresh_systemd_unit_if_needed(system: bool = False) -> bool:
"""Rewrite the installed user unit when the generated definition has changed.""" """Rewrite the installed systemd unit when the generated definition has changed."""
unit_path = get_systemd_unit_path() unit_path = get_systemd_unit_path(system=system)
if not unit_path.exists() or systemd_unit_is_current(): if not unit_path.exists() or systemd_unit_is_current(system=system):
return False return False
unit_path.write_text(generate_systemd_unit(), encoding="utf-8") expected_user = _read_systemd_user_from_unit(unit_path) if system else None
subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) unit_path.write_text(generate_systemd_unit(system=system, run_as_user=expected_user), encoding="utf-8")
print("↻ Updated gateway service definition to match the current Hermes install") subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True)
print(f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install")
return True return True
@@ -337,93 +421,131 @@ def _ensure_linger_enabled() -> None:
_print_linger_enable_warning(username, detail or linger_detail) _print_linger_enable_warning(username, detail or linger_detail)
def systemd_install(force: bool = False): def _select_systemd_scope(system: bool = False) -> bool:
unit_path = get_systemd_unit_path() if system:
return True
return get_systemd_unit_path(system=True).exists() and not get_systemd_unit_path(system=False).exists()
def systemd_install(force: bool = False, system: bool = False, run_as_user: str | None = None):
if system:
_require_root_for_system_service("install")
unit_path = get_systemd_unit_path(system=system)
scope_flag = " --system" if system else ""
if unit_path.exists() and not force: if unit_path.exists() and not force:
print(f"Service already installed at: {unit_path}") print(f"Service already installed at: {unit_path}")
print("Use --force to reinstall") print("Use --force to reinstall")
return return
unit_path.parent.mkdir(parents=True, exist_ok=True) unit_path.parent.mkdir(parents=True, exist_ok=True)
print(f"Installing systemd service to: {unit_path}") print(f"Installing {_service_scope_label(system)} systemd service to: {unit_path}")
unit_path.write_text(generate_systemd_unit()) unit_path.write_text(generate_systemd_unit(system=system, run_as_user=run_as_user), encoding="utf-8")
subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True)
subprocess.run(["systemctl", "--user", "enable", SERVICE_NAME], check=True) subprocess.run(_systemctl_cmd(system) + ["enable", SERVICE_NAME], check=True)
print() print()
print("Service installed and enabled!") print(f"{_service_scope_label(system).capitalize()} service installed and enabled!")
print() print()
print("Next steps:") print("Next steps:")
print(f" hermes gateway start # Start the service") print(f" {'sudo ' if system else ''}hermes gateway start{scope_flag} # Start the service")
print(f" hermes gateway status # Check status") print(f" {'sudo ' if system else ''}hermes gateway status{scope_flag} # Check status")
print(f" journalctl --user -u {SERVICE_NAME} -f # View logs") print(f" {'journalctl' if system else 'journalctl --user'} -u {SERVICE_NAME} -f # View logs")
print() print()
_ensure_linger_enabled()
def systemd_uninstall(): if system:
subprocess.run(["systemctl", "--user", "stop", SERVICE_NAME], check=False) configured_user = _read_systemd_user_from_unit(unit_path)
subprocess.run(["systemctl", "--user", "disable", SERVICE_NAME], check=False) if configured_user:
print(f"Configured to run as: {configured_user}")
unit_path = get_systemd_unit_path() else:
_ensure_linger_enabled()
def systemd_uninstall(system: bool = False):
system = _select_systemd_scope(system)
if system:
_require_root_for_system_service("uninstall")
subprocess.run(_systemctl_cmd(system) + ["stop", SERVICE_NAME], check=False)
subprocess.run(_systemctl_cmd(system) + ["disable", SERVICE_NAME], check=False)
unit_path = get_systemd_unit_path(system=system)
if unit_path.exists(): if unit_path.exists():
unit_path.unlink() unit_path.unlink()
print(f"✓ Removed {unit_path}") print(f"✓ Removed {unit_path}")
subprocess.run(["systemctl", "--user", "daemon-reload"], check=True)
print("✓ Service uninstalled")
def systemd_start(): subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True)
refresh_systemd_unit_if_needed() print(f"{_service_scope_label(system).capitalize()} service uninstalled")
subprocess.run(["systemctl", "--user", "start", SERVICE_NAME], check=True)
print("✓ Service started")
def systemd_stop(): def systemd_start(system: bool = False):
subprocess.run(["systemctl", "--user", "stop", SERVICE_NAME], check=True) system = _select_systemd_scope(system)
print("✓ Service stopped") if system:
_require_root_for_system_service("start")
refresh_systemd_unit_if_needed(system=system)
subprocess.run(_systemctl_cmd(system) + ["start", SERVICE_NAME], check=True)
print(f"{_service_scope_label(system).capitalize()} service started")
def systemd_restart():
refresh_systemd_unit_if_needed() def systemd_stop(system: bool = False):
subprocess.run(["systemctl", "--user", "restart", SERVICE_NAME], check=True) system = _select_systemd_scope(system)
print("✓ Service restarted") if system:
_require_root_for_system_service("stop")
subprocess.run(_systemctl_cmd(system) + ["stop", SERVICE_NAME], check=True)
print(f"{_service_scope_label(system).capitalize()} service stopped")
def systemd_status(deep: bool = False):
# Check if service unit file exists def systemd_restart(system: bool = False):
unit_path = get_systemd_unit_path() system = _select_systemd_scope(system)
if system:
_require_root_for_system_service("restart")
refresh_systemd_unit_if_needed(system=system)
subprocess.run(_systemctl_cmd(system) + ["restart", SERVICE_NAME], check=True)
print(f"{_service_scope_label(system).capitalize()} service restarted")
def systemd_status(deep: bool = False, system: bool = False):
system = _select_systemd_scope(system)
unit_path = get_systemd_unit_path(system=system)
scope_flag = " --system" if system else ""
if not unit_path.exists(): if not unit_path.exists():
print("✗ Gateway service is not installed") print("✗ Gateway service is not installed")
print(" Run: hermes gateway install") print(f" Run: {'sudo ' if system else ''}hermes gateway install{scope_flag}")
return return
if not systemd_unit_is_current(): if not systemd_unit_is_current(system=system):
print("⚠ Installed gateway service definition is outdated") print("⚠ Installed gateway service definition is outdated")
print(" Run: hermes gateway restart # auto-refreshes the unit") print(f" Run: {'sudo ' if system else ''}hermes gateway restart{scope_flag} # auto-refreshes the unit")
print() print()
# Show detailed status first
subprocess.run( subprocess.run(
["systemctl", "--user", "status", SERVICE_NAME, "--no-pager"], _systemctl_cmd(system) + ["status", SERVICE_NAME, "--no-pager"],
capture_output=False capture_output=False,
) )
# Check if service is active
result = subprocess.run( result = subprocess.run(
["systemctl", "--user", "is-active", SERVICE_NAME], _systemctl_cmd(system) + ["is-active", SERVICE_NAME],
capture_output=True, capture_output=True,
text=True text=True,
) )
status = result.stdout.strip() status = result.stdout.strip()
if status == "active": if status == "active":
print("Gateway service is running") print(f"{_service_scope_label(system).capitalize()} gateway service is running")
else: else:
print("Gateway service is stopped") print(f"{_service_scope_label(system).capitalize()} gateway service is stopped")
print(" Run: hermes gateway start") print(f" Run: {'sudo ' if system else ''}hermes gateway start{scope_flag}")
configured_user = _read_systemd_user_from_unit(unit_path) if system else None
if configured_user:
print(f"Configured to run as: {configured_user}")
runtime_lines = _runtime_health_lines() runtime_lines = _runtime_health_lines()
if runtime_lines: if runtime_lines:
@@ -432,7 +554,9 @@ def systemd_status(deep: bool = False):
for line in runtime_lines: for line in runtime_lines:
print(f" {line}") print(f" {line}")
if deep: if system:
print("✓ System service starts at boot without requiring systemd linger")
elif deep:
print_systemd_linger_guidance() print_systemd_linger_guidance()
else: else:
linger_enabled, _ = get_systemd_linger_status() linger_enabled, _ = get_systemd_linger_status()
@@ -445,10 +569,7 @@ def systemd_status(deep: bool = False):
if deep: if deep:
print() print()
print("Recent logs:") print("Recent logs:")
subprocess.run([ subprocess.run(_journalctl_cmd(system) + ["-u", SERVICE_NAME, "-n", "20", "--no-pager"])
"journalctl", "--user", "-u", SERVICE_NAME,
"-n", "20", "--no-pager"
])
# ============================================================================= # =============================================================================
@@ -895,7 +1016,7 @@ def _setup_whatsapp():
def _is_service_installed() -> bool: def _is_service_installed() -> bool:
"""Check if the gateway is installed as a system service.""" """Check if the gateway is installed as a system service."""
if is_linux(): if is_linux():
return get_systemd_unit_path().exists() return get_systemd_unit_path(system=False).exists() or get_systemd_unit_path(system=True).exists()
elif is_macos(): elif is_macos():
return get_launchd_plist_path().exists() return get_launchd_plist_path().exists()
return False return False
@@ -903,12 +1024,19 @@ def _is_service_installed() -> bool:
def _is_service_running() -> bool: def _is_service_running() -> bool:
"""Check if the gateway service is currently running.""" """Check if the gateway service is currently running."""
if is_linux() and get_systemd_unit_path().exists(): if is_linux():
result = subprocess.run( if get_systemd_unit_path(system=False).exists():
["systemctl", "--user", "is-active", SERVICE_NAME], result = subprocess.run(
capture_output=True, text=True _systemctl_cmd(False) + ["is-active", SERVICE_NAME],
) capture_output=True, text=True
return result.stdout.strip() == "active" )
return result.stdout.strip() == "active"
if get_systemd_unit_path(system=True).exists():
result = subprocess.run(
_systemctl_cmd(True) + ["is-active", SERVICE_NAME],
capture_output=True, text=True
)
return result.stdout.strip() == "active"
elif is_macos() and get_launchd_plist_path().exists(): elif is_macos() and get_launchd_plist_path().exists():
result = subprocess.run( result = subprocess.run(
["launchctl", "list", "ai.hermes.gateway"], ["launchctl", "list", "ai.hermes.gateway"],
@@ -1183,8 +1311,10 @@ def gateway_command(args):
# Service management commands # Service management commands
if subcmd == "install": if subcmd == "install":
force = getattr(args, 'force', False) force = getattr(args, 'force', False)
system = getattr(args, 'system', False)
run_as_user = getattr(args, 'run_as_user', None)
if is_linux(): if is_linux():
systemd_install(force) systemd_install(force=force, system=system, run_as_user=run_as_user)
elif is_macos(): elif is_macos():
launchd_install(force) launchd_install(force)
else: else:
@@ -1193,8 +1323,9 @@ def gateway_command(args):
sys.exit(1) sys.exit(1)
elif subcmd == "uninstall": elif subcmd == "uninstall":
system = getattr(args, 'system', False)
if is_linux(): if is_linux():
systemd_uninstall() systemd_uninstall(system=system)
elif is_macos(): elif is_macos():
launchd_uninstall() launchd_uninstall()
else: else:
@@ -1202,8 +1333,9 @@ def gateway_command(args):
sys.exit(1) sys.exit(1)
elif subcmd == "start": elif subcmd == "start":
system = getattr(args, 'system', False)
if is_linux(): if is_linux():
systemd_start() systemd_start(system=system)
elif is_macos(): elif is_macos():
launchd_start() launchd_start()
else: else:
@@ -1213,10 +1345,11 @@ def gateway_command(args):
elif subcmd == "stop": elif subcmd == "stop":
# Try service first, then sweep any stray/manual gateway processes. # Try service first, then sweep any stray/manual gateway processes.
service_available = False service_available = False
system = getattr(args, 'system', False)
if is_linux() and get_systemd_unit_path().exists(): if is_linux() and (get_systemd_unit_path(system=False).exists() or get_systemd_unit_path(system=True).exists()):
try: try:
systemd_stop() systemd_stop(system=system)
service_available = True service_available = True
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
pass # Fall through to process kill pass # Fall through to process kill
@@ -1239,10 +1372,11 @@ def gateway_command(args):
elif subcmd == "restart": elif subcmd == "restart":
# Try service first, fall back to killing and restarting # Try service first, fall back to killing and restarting
service_available = False service_available = False
system = getattr(args, 'system', False)
if is_linux() and get_systemd_unit_path().exists(): if is_linux() and (get_systemd_unit_path(system=False).exists() or get_systemd_unit_path(system=True).exists()):
try: try:
systemd_restart() systemd_restart(system=system)
service_available = True service_available = True
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
pass pass
@@ -1268,10 +1402,11 @@ def gateway_command(args):
elif subcmd == "status": elif subcmd == "status":
deep = getattr(args, 'deep', False) deep = getattr(args, 'deep', False)
system = getattr(args, 'system', False)
# Check for service first # Check for service first
if is_linux() and get_systemd_unit_path().exists(): if is_linux() and (get_systemd_unit_path(system=False).exists() or get_systemd_unit_path(system=True).exists()):
systemd_status(deep) systemd_status(deep, system=system)
elif is_macos() and get_launchd_plist_path().exists(): elif is_macos() and get_launchd_plist_path().exists():
launchd_status(deep) launchd_status(deep)
else: else:
@@ -1289,6 +1424,7 @@ def gateway_command(args):
print() print()
print("To install as a service:") print("To install as a service:")
print(" hermes gateway install") print(" hermes gateway install")
print(" sudo hermes gateway install --system")
else: else:
print("✗ Gateway is not running") print("✗ Gateway is not running")
runtime_lines = _runtime_health_lines() runtime_lines = _runtime_health_lines()
@@ -1300,4 +1436,5 @@ def gateway_command(args):
print() print()
print("To start:") print("To start:")
print(" hermes gateway # Run in foreground") print(" hermes gateway # Run in foreground")
print(" hermes gateway install # Install as service") print(" hermes gateway install # Install as user service")
print(" sudo hermes gateway install --system # Install as boot-time system service")

View File

@@ -2433,23 +2433,30 @@ For more help on a command:
# gateway start # gateway start
gateway_start = gateway_subparsers.add_parser("start", help="Start gateway service") gateway_start = gateway_subparsers.add_parser("start", help="Start gateway service")
gateway_start.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service")
# gateway stop # gateway stop
gateway_stop = gateway_subparsers.add_parser("stop", help="Stop gateway service") gateway_stop = gateway_subparsers.add_parser("stop", help="Stop gateway service")
gateway_stop.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service")
# gateway restart # gateway restart
gateway_restart = gateway_subparsers.add_parser("restart", help="Restart gateway service") gateway_restart = gateway_subparsers.add_parser("restart", help="Restart gateway service")
gateway_restart.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service")
# gateway status # gateway status
gateway_status = gateway_subparsers.add_parser("status", help="Show gateway status") gateway_status = gateway_subparsers.add_parser("status", help="Show gateway status")
gateway_status.add_argument("--deep", action="store_true", help="Deep status check") gateway_status.add_argument("--deep", action="store_true", help="Deep status check")
gateway_status.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service")
# gateway install # gateway install
gateway_install = gateway_subparsers.add_parser("install", help="Install gateway as service") gateway_install = gateway_subparsers.add_parser("install", help="Install gateway as service")
gateway_install.add_argument("--force", action="store_true", help="Force reinstall") gateway_install.add_argument("--force", action="store_true", help="Force reinstall")
gateway_install.add_argument("--system", action="store_true", help="Install as a Linux system-level service (starts at boot)")
gateway_install.add_argument("--run-as-user", dest="run_as_user", help="User account the Linux system service should run as")
# gateway uninstall # gateway uninstall
gateway_uninstall = gateway_subparsers.add_parser("uninstall", help="Uninstall gateway service") gateway_uninstall = gateway_subparsers.add_parser("uninstall", help="Uninstall gateway service")
gateway_uninstall.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service")
# gateway setup # gateway setup
gateway_setup = gateway_subparsers.add_parser("setup", help="Configure messaging platforms") gateway_setup = gateway_subparsers.add_parser("setup", help="Configure messaging platforms")

View File

@@ -35,7 +35,7 @@ def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys
unit_path = tmp_path / "hermes-gateway.service" unit_path = tmp_path / "hermes-gateway.service"
unit_path.write_text("[Unit]\n") unit_path.write_text("[Unit]\n")
monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda: unit_path) monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (False, "")) monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (False, ""))
def fake_run(cmd, capture_output=False, text=False, check=False): def fake_run(cmd, capture_output=False, text=False, check=False):
@@ -50,7 +50,7 @@ def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys
gateway.systemd_status(deep=False) gateway.systemd_status(deep=False)
out = capsys.readouterr().out out = capsys.readouterr().out
assert "Gateway service is running" in out assert "gateway service is running" in out
assert "Systemd linger is disabled" in out assert "Systemd linger is disabled" in out
assert "loginctl enable-linger" in out assert "loginctl enable-linger" in out
@@ -58,7 +58,7 @@ def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys
def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys): def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys):
unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service" unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service"
monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda: unit_path) monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path)
calls = [] calls = []
helper_calls = [] helper_calls = []
@@ -79,4 +79,39 @@ def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys):
["systemctl", "--user", "enable", gateway.SERVICE_NAME], ["systemctl", "--user", "enable", gateway.SERVICE_NAME],
] ]
assert helper_calls == [True] assert helper_calls == [True]
assert "Service installed and enabled" in out assert "User service installed and enabled" in out
def test_systemd_install_system_scope_skips_linger_and_uses_systemctl(monkeypatch, tmp_path, capsys):
unit_path = tmp_path / "etc" / "systemd" / "system" / "hermes-gateway.service"
monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(
gateway,
"generate_systemd_unit",
lambda system=False, run_as_user=None: f"scope={system} user={run_as_user}\n",
)
monkeypatch.setattr(gateway, "_require_root_for_system_service", lambda action: None)
calls = []
helper_calls = []
def fake_run(cmd, check=False, **kwargs):
calls.append((cmd, check))
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway.subprocess, "run", fake_run)
monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True))
gateway.systemd_install(force=False, system=True, run_as_user="alice")
out = capsys.readouterr().out
assert unit_path.exists()
assert unit_path.read_text(encoding="utf-8") == "scope=True user=alice\n"
assert [cmd for cmd, _ in calls] == [
["systemctl", "daemon-reload"],
["systemctl", "enable", gateway.SERVICE_NAME],
]
assert helper_calls == []
assert "Configured to run as: alice" not in out # generated test unit has no User= line
assert "System service installed and enabled" in out

View File

@@ -96,7 +96,7 @@ class TestEnsureLingerEnabled:
def test_systemd_install_calls_linger_helper(monkeypatch, tmp_path, capsys): def test_systemd_install_calls_linger_helper(monkeypatch, tmp_path, capsys):
unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service" unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service"
monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda: unit_path) monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path)
calls = [] calls = []
@@ -117,4 +117,4 @@ def test_systemd_install_calls_linger_helper(monkeypatch, tmp_path, capsys):
["systemctl", "--user", "enable", gateway.SERVICE_NAME], ["systemctl", "--user", "enable", gateway.SERVICE_NAME],
] ]
assert helper_calls == [True] assert helper_calls == [True]
assert "Service installed and enabled" in out assert "User service installed and enabled" in out

View File

@@ -10,8 +10,8 @@ class TestSystemdServiceRefresh:
unit_path = tmp_path / "hermes-gateway.service" unit_path = tmp_path / "hermes-gateway.service"
unit_path.write_text("old unit\n", encoding="utf-8") unit_path.write_text("old unit\n", encoding="utf-8")
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda: unit_path) monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda: "new unit\n") monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
calls = [] calls = []
@@ -33,8 +33,8 @@ class TestSystemdServiceRefresh:
unit_path = tmp_path / "hermes-gateway.service" unit_path = tmp_path / "hermes-gateway.service"
unit_path.write_text("old unit\n", encoding="utf-8") unit_path.write_text("old unit\n", encoding="utf-8")
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda: unit_path) monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda: "new unit\n") monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
calls = [] calls = []
@@ -60,12 +60,12 @@ class TestGatewayStopCleanup:
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True) monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False) monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda: unit_path) monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
service_calls = [] service_calls = []
kill_calls = [] kill_calls = []
monkeypatch.setattr(gateway_cli, "systemd_stop", lambda: service_calls.append("stop")) monkeypatch.setattr(gateway_cli, "systemd_stop", lambda system=False: service_calls.append("stop"))
monkeypatch.setattr( monkeypatch.setattr(
gateway_cli, gateway_cli,
"kill_gateway_processes", "kill_gateway_processes",
@@ -76,3 +76,41 @@ class TestGatewayStopCleanup:
assert service_calls == ["stop"] assert service_calls == ["stop"]
assert kill_calls == [False] assert kill_calls == [False]
class TestGatewaySystemServiceRouting:
def test_gateway_install_passes_system_flags(self, monkeypatch):
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
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_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, "is_linux", lambda: True)
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)]