Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions hermes_cli/gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 ""
Expand Down Expand Up @@ -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()

Expand Down
49 changes: 32 additions & 17 deletions hermes_cli/status.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
"""

import os
import shutil
import sys
import subprocess
from pathlib import Path
Expand Down Expand Up @@ -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
Expand Down
188 changes: 188 additions & 0 deletions tests/hermes_cli/test_container_detection.py
Original file line number Diff line number Diff line change
@@ -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
Loading