fix(gateway): allow systemd-backed distrobox services

This commit is contained in:
helix4u 2026-04-17 17:03:26 -06:00 committed by Teknium
parent 213e39463b
commit 47010e0757
2 changed files with 94 additions and 5 deletions

View file

@ -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

View file

@ -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")