diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index c69265b19e..08f9999e66 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -487,25 +487,44 @@ def _wsl_systemd_operational() -> bool: WSL2 with ``systemd=true`` in wsl.conf has working systemd. WSL2 without it (or WSL1) does not — systemctl commands fail. """ + return _systemd_operational(system=True) + + +def _systemd_operational(system: bool = False) -> bool: + """Return True when the requested systemd scope is usable.""" try: - result = subprocess.run( - ["systemctl", "is-system-running"], - capture_output=True, text=True, timeout=5, + result = _run_systemctl( + ["is-system-running"], + system=system, + capture_output=True, + text=True, + timeout=5, ) # "running", "degraded", "starting" all mean systemd is PID 1 status = result.stdout.strip().lower() return status in ("running", "degraded", "starting", "initializing") - except (FileNotFoundError, subprocess.TimeoutExpired, OSError): + except (RuntimeError, subprocess.TimeoutExpired, OSError): return False +def _container_systemd_operational() -> bool: + """Return True when a container exposes working user or system systemd.""" + if _systemd_operational(system=False): + return True + if _systemd_operational(system=True): + return True + return False + + def supports_systemd_services() -> bool: - if not is_linux() or is_termux() or is_container(): + if not is_linux() or is_termux(): return False if shutil.which("systemctl") is None: return False if is_wsl(): return _wsl_systemd_operational() + if is_container(): + return _container_systemd_operational() return True diff --git a/tests/hermes_cli/test_gateway.py b/tests/hermes_cli/test_gateway.py index 9e9a14bce7..07265b2c3a 100644 --- a/tests/hermes_cli/test_gateway.py +++ b/tests/hermes_cli/test_gateway.py @@ -39,6 +39,76 @@ class TestSystemdLingerStatus: assert gateway.get_systemd_linger_status() == (None, "not supported in Termux") +class TestContainerSystemdSupport: + def test_supports_systemd_services_in_container_with_user_manager(self, monkeypatch): + monkeypatch.setattr(gateway, "is_linux", lambda: True) + monkeypatch.setattr(gateway, "is_termux", lambda: False) + monkeypatch.setattr(gateway, "is_wsl", lambda: False) + monkeypatch.setattr(gateway, "is_container", lambda: True) + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl") + monkeypatch.setattr(gateway, "_systemd_operational", lambda system=False: not system) + + assert gateway.supports_systemd_services() is True + + def test_supports_systemd_services_in_container_with_system_manager(self, monkeypatch): + monkeypatch.setattr(gateway, "is_linux", lambda: True) + monkeypatch.setattr(gateway, "is_termux", lambda: False) + monkeypatch.setattr(gateway, "is_wsl", lambda: False) + monkeypatch.setattr(gateway, "is_container", lambda: True) + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl") + monkeypatch.setattr(gateway, "_systemd_operational", lambda system=False: system) + + assert gateway.supports_systemd_services() is True + + def test_supports_systemd_services_in_container_without_systemd(self, monkeypatch): + monkeypatch.setattr(gateway, "is_linux", lambda: True) + monkeypatch.setattr(gateway, "is_termux", lambda: False) + monkeypatch.setattr(gateway, "is_wsl", lambda: False) + monkeypatch.setattr(gateway, "is_container", lambda: True) + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl") + monkeypatch.setattr(gateway, "_systemd_operational", lambda system=False: False) + + assert gateway.supports_systemd_services() is False + + +def test_gateway_install_in_container_with_operational_systemd_uses_systemd(monkeypatch): + monkeypatch.setattr(gateway, "supports_systemd_services", lambda: True) + monkeypatch.setattr(gateway, "is_wsl", lambda: False) + monkeypatch.setattr(gateway, "is_macos", lambda: False) + monkeypatch.setattr(gateway, "is_managed", lambda: False) + + calls = [] + monkeypatch.setattr( + gateway, + "systemd_install", + lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)), + ) + + args = SimpleNamespace( + gateway_command="install", + force=False, + system=False, + run_as_user=None, + ) + gateway.gateway_command(args) + + assert calls == [(False, False, None)] + + +def test_gateway_start_in_container_with_operational_systemd_uses_systemd(monkeypatch): + monkeypatch.setattr(gateway, "supports_systemd_services", lambda: True) + monkeypatch.setattr(gateway, "is_wsl", lambda: False) + monkeypatch.setattr(gateway, "is_macos", lambda: False) + + calls = [] + monkeypatch.setattr(gateway, "systemd_start", lambda system=False: calls.append(system)) + + args = SimpleNamespace(gateway_command="start", system=False, all=False) + gateway.gateway_command(args) + + assert calls == [False] + + def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys): unit_path = tmp_path / "hermes-gateway.service" unit_path.write_text("[Unit]\n")