From 937720b628556196ac9831b0ed180cebcf838ba6 Mon Sep 17 00:00:00 2001 From: Alberto Leal Date: Wed, 1 Apr 2026 03:46:41 -0400 Subject: [PATCH] fix: gracefully handle missing systemctl in container environments hermes status and gateway status crash with FileNotFoundError when running in K8s containers that don't have systemd/systemctl. Detect container environments via KUBERNETES_SERVICE_HOST env var or missing systemctl binary, and show informative N/A output instead of crashing. Tests: 10 tests (4 unit for _is_container_env, 2 for systemd_status, 2 for _is_service_running, 2 e2e for full status output). --- hermes_cli/gateway.py | 25 +++ hermes_cli/status.py | 49 +++-- tests/hermes_cli/test_container_detection.py | 188 +++++++++++++++++++ 3 files changed, 245 insertions(+), 17 deletions(-) create mode 100644 tests/hermes_cli/test_container_detection.py diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 4a12a34bb0e1..a2442b792b00 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -231,6 +231,10 @@ def is_macos() -> bool: def is_windows() -> bool: return sys.platform == 'win32' +def _is_container_env() -> bool: + """Detect container/K8s environment where systemd is unavailable.""" + return bool(os.environ.get("KUBERNETES_SERVICE_HOST") or not shutil.which("systemctl")) + # ============================================================================= # Service Configuration @@ -875,6 +879,24 @@ def systemd_restart(system: bool = False): def systemd_status(deep: bool = False, system: bool = False): + if _is_container_env(): + try: + from gateway.status import is_gateway_running, read_runtime_status + _running = is_gateway_running() + print(f"Gateway service: {'running' if _running else 'stopped'} (container/K8s)") + # Show runtime health details if available + _state = read_runtime_status() + if _state: + gw_state = _state.get("gateway_state", "unknown") + print(f" Runtime state: {gw_state}") + platforms = _state.get("platforms", {}) + for pname, pstate in platforms.items(): + status = pstate.get("status", "unknown") + print(f" Platform {pname}: {status}") + except Exception: + print("Gateway service: unknown (container/K8s, status check failed)") + return + system = _select_systemd_scope(system) unit_path = get_systemd_unit_path(system=system) scope_flag = " --system" if system else "" @@ -1741,6 +1763,9 @@ def _is_service_installed() -> bool: def _is_service_running() -> bool: """Check if the gateway service is currently running.""" if is_linux(): + if _is_container_env(): + return False + user_unit_exists = get_systemd_unit_path(system=False).exists() system_unit_exists = get_systemd_unit_path(system=True).exists() diff --git a/hermes_cli/status.py b/hermes_cli/status.py index 67b15bab7892..b9e8cf9980e6 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.py @@ -5,6 +5,7 @@ """ import os +import shutil import sys import subprocess from pathlib import Path @@ -307,23 +308,37 @@ def show_status(args): print(color("◆ Gateway Service", Colors.CYAN, Colors.BOLD)) if sys.platform.startswith('linux'): - try: - from hermes_cli.gateway import get_service_name - _gw_svc = get_service_name() - except Exception: - _gw_svc = "hermes-gateway" - try: - result = subprocess.run( - ["systemctl", "--user", "is-active", _gw_svc], - capture_output=True, - text=True, - timeout=5 - ) - is_active = result.stdout.strip() == "active" - except subprocess.TimeoutExpired: - is_active = False - print(f" Status: {check_mark(is_active)} {'running' if is_active else 'stopped'}") - print(" Manager: systemd (user)") + # Detect container environment (no systemd) + _in_container = ( + os.environ.get("KUBERNETES_SERVICE_HOST") + or not shutil.which("systemctl") + ) + if _in_container: + try: + from gateway.status import is_gateway_running + _gw_running = is_gateway_running() + except Exception: + _gw_running = False + print(f" Status: {check_mark(_gw_running)} {'running' if _gw_running else 'stopped'}") + print(" Manager: container/K8s") + else: + try: + from hermes_cli.gateway import get_service_name + _gw_svc = get_service_name() + except Exception: + _gw_svc = "hermes-gateway" + try: + result = subprocess.run( + ["systemctl", "--user", "is-active", _gw_svc], + capture_output=True, + text=True, + timeout=5 + ) + is_active = result.stdout.strip() == "active" + except subprocess.TimeoutExpired: + is_active = False + print(f" Status: {check_mark(is_active)} {'running' if is_active else 'stopped'}") + print(" Manager: systemd (user)") elif sys.platform == 'darwin': from hermes_cli.gateway import get_launchd_label diff --git a/tests/hermes_cli/test_container_detection.py b/tests/hermes_cli/test_container_detection.py new file mode 100644 index 000000000000..ed21b2daf8dc --- /dev/null +++ b/tests/hermes_cli/test_container_detection.py @@ -0,0 +1,188 @@ +"""Tests for container/K8s environment detection across gateway and status modules.""" + +from types import SimpleNamespace + +import hermes_cli.gateway as gateway_mod +import hermes_cli.status as status_mod + + +# ============================================================================= +# Unit Tests: _is_container_env() +# ============================================================================= + + +class TestIsContainerEnv: + def test_returns_true_when_kubernetes_service_host_set(self, monkeypatch): + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.43.0.1") + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl") + + assert gateway_mod._is_container_env() is True + + def test_returns_true_when_systemctl_missing(self, monkeypatch): + monkeypatch.delenv("KUBERNETES_SERVICE_HOST", raising=False) + monkeypatch.setattr( + "shutil.which", + lambda name: None if name == "systemctl" else f"/usr/bin/{name}", + ) + + assert gateway_mod._is_container_env() is True + + def test_returns_false_on_normal_linux(self, monkeypatch): + monkeypatch.delenv("KUBERNETES_SERVICE_HOST", raising=False) + monkeypatch.setattr( + "shutil.which", + lambda name: "/usr/bin/systemctl" if name == "systemctl" else None, + ) + + assert gateway_mod._is_container_env() is False + + def test_returns_true_when_both_k8s_and_no_systemctl(self, monkeypatch): + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.43.0.1") + monkeypatch.setattr("shutil.which", lambda name: None) + + assert gateway_mod._is_container_env() is True + + +# ============================================================================= +# Unit Tests: systemd_status() container early-return +# ============================================================================= + + +class TestSystemdStatusContainer: + def test_prints_running_in_container(self, monkeypatch, capsys): + monkeypatch.setattr(gateway_mod, "_is_container_env", lambda: True) + monkeypatch.setattr("gateway.status.is_gateway_running", lambda: True) + + gateway_mod.systemd_status() + + output = capsys.readouterr().out + assert "running" in output + assert "container/K8s" in output + + def test_prints_stopped_in_container(self, monkeypatch, capsys): + monkeypatch.setattr(gateway_mod, "_is_container_env", lambda: True) + monkeypatch.setattr("gateway.status.is_gateway_running", lambda: False) + + gateway_mod.systemd_status() + + output = capsys.readouterr().out + assert "stopped" in output + assert "container/K8s" in output + + def test_does_not_call_systemctl_in_container(self, monkeypatch, capsys): + monkeypatch.setattr(gateway_mod, "_is_container_env", lambda: True) + calls = [] + monkeypatch.setattr( + gateway_mod.subprocess, + "run", + lambda cmd, **kw: calls.append(cmd), + ) + + gateway_mod.systemd_status() + + assert calls == [], "systemctl should not be called in container env" + + +# ============================================================================= +# Unit Tests: _is_service_running() container path +# ============================================================================= + + +class TestIsServiceRunningContainer: + def test_returns_false_in_container_on_linux(self, monkeypatch): + monkeypatch.setattr(gateway_mod, "is_linux", lambda: True) + monkeypatch.setattr(gateway_mod, "_is_container_env", lambda: True) + + assert gateway_mod._is_service_running() is False + + def test_calls_systemctl_when_not_container_on_linux(self, monkeypatch, tmp_path): + monkeypatch.setattr(gateway_mod, "is_linux", lambda: True) + monkeypatch.setattr(gateway_mod, "_is_container_env", lambda: False) + + # Both unit paths must exist for the systemctl call to happen + user_unit = tmp_path / "user" / "hermes-gateway.service" + user_unit.parent.mkdir(parents=True) + user_unit.write_text("[Unit]\n") + monkeypatch.setattr( + gateway_mod, + "get_systemd_unit_path", + lambda system=False: user_unit if not system else tmp_path / "nope", + ) + + calls = [] + monkeypatch.setattr( + gateway_mod.subprocess, + "run", + lambda cmd, **kw: ( + calls.append(cmd), + SimpleNamespace(returncode=0, stdout="active\n", stderr=""), + )[1], + ) + + result = gateway_mod._is_service_running() + + assert len(calls) > 0, "should have called systemctl" + assert result is True + + +# ============================================================================= +# E2E Tests: status.py container detection output +# ============================================================================= + + +class TestStatusOutputContainer: + def test_status_shows_running_in_container_on_linux(self, monkeypatch, capsys): + """hermes status on Linux in a container shows running gateway.""" + monkeypatch.setattr("sys.platform", "linux") + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.43.0.1") + monkeypatch.setattr( + "shutil.which", + lambda name: None if name == "systemctl" else f"/usr/bin/{name}", + ) + monkeypatch.setattr("gateway.status.is_gateway_running", lambda: True) + + status_mod.show_status(SimpleNamespace(all=False, deep=False)) + + output = capsys.readouterr().out + assert "container/K8s" in output + assert "running" in output + + def test_status_shows_stopped_in_container_on_linux(self, monkeypatch, capsys): + """hermes status on Linux in a container shows stopped gateway.""" + monkeypatch.setattr("sys.platform", "linux") + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.43.0.1") + monkeypatch.setattr( + "shutil.which", + lambda name: None if name == "systemctl" else f"/usr/bin/{name}", + ) + monkeypatch.setattr("gateway.status.is_gateway_running", lambda: False) + + status_mod.show_status(SimpleNamespace(all=False, deep=False)) + + output = capsys.readouterr().out + assert "container/K8s" in output + assert "stopped" in output + + def test_status_shows_systemd_on_normal_linux(self, monkeypatch, capsys): + """hermes status on normal Linux shows systemd manager.""" + monkeypatch.setattr("sys.platform", "linux") + monkeypatch.delenv("KUBERNETES_SERVICE_HOST", raising=False) + monkeypatch.setattr( + "shutil.which", + lambda name: "/usr/bin/systemctl" if name == "systemctl" else None, + ) + + # Mock the systemctl call that checks gateway status + monkeypatch.setattr( + status_mod.subprocess, + "run", + lambda cmd, **kw: SimpleNamespace( + returncode=0, stdout="active\n", stderr="" + ), + ) + + status_mod.show_status(SimpleNamespace(all=False, deep=False)) + + output = capsys.readouterr().out + assert "systemd (user)" in output + assert "container/K8s" not in output