9dc9a6998f
Some checks failed
Nix / nix (macos-latest) (pull_request) Waiting to run
Contributor Attribution Check / check-attribution (pull_request) Failing after 1m36s
Supply Chain Audit / Scan PR for critical supply chain risks (pull_request) Successful in 1m37s
Tests / e2e (pull_request) Successful in 1m59s
Tests / test (pull_request) Failing after 18m17s
Nix / nix (ubuntu-latest) (pull_request) Failing after 22m16s
Sub-shape A (TimeoutStopSec literal drift): - generate_systemd_unit() formula: max(60, drain_timeout) + 30 - DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT bumped 60→180 in config.py, so emitted TimeoutStopSec went 90→210; tests pinned the literal 90. - Replace literal with TestGeneratedSystemdUnits._expected_timeout_stop_sec() helper that mirrors the production formula via _get_restart_drain_timeout(), so future config-default bumps don't silently regress the test. Sub-shape B (production preflight not stubbed): - systemd_start() / systemd_restart() now call _preflight_user_systemd() before the systemctl call sequence (PR #14531: "preflight user D-Bus before systemctl --user start"). The preflight invokes loginctl enable-linger and waits for the D-Bus socket — neither of which the unit tests' fake subprocess runner answers. - Unit-tests under TestSystemdServiceRefresh and TestGatewaySystemServiceRouting assert the systemctl call sequence, not the preflight; preflight has dedicated coverage in TestUserSystemdPrivateSocketPreflight. Stub _preflight_user_systemd as a no-op in the four affected tests. Sub-shape B (supports_systemd_services container branch): - supports_systemd_services() now branches on is_container() to decide whether to probe `systemctl is-system-running`. Tests that assert the native-Linux True path didn't stub is_container, so a containerized CI runner inherited a real probe of the runner image's systemd: - test_supports_systemd_services_returns_true_when_systemctl_present - TestSupportsSystemdServicesWSL.test_native_linux - Stub is_container() False in both, plus shutil.which() in the WSL test so it also passes on macOS dev boxes (was implicitly Linux-only via systemctl-on-PATH). Tests fixed: test_systemd_start_refreshes_outdated_unit test_systemd_restart_refreshes_outdated_unit test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout test_supports_systemd_services_returns_true_when_systemctl_present test_systemd_restart_self_requests_graceful_restart_and_waits test_systemd_restart_recovers_failed_planned_restart TestSupportsSystemdServicesWSL.test_native_linux Verified locally on darwin py3.13: all 8 target tests pass; one unrelated macOS-only failure (test_wsl_with_systemd) remains because its body relies on the host having systemctl on PATH — not in this PR's scope (not in the issue's failing-list). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
289 lines
12 KiB
Python
289 lines
12 KiB
Python
"""Tests for WSL detection and WSL-aware gateway behavior."""
|
|
|
|
import io
|
|
import subprocess
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch, MagicMock, mock_open
|
|
|
|
import pytest
|
|
|
|
import hermes_cli.gateway as gateway
|
|
import hermes_constants
|
|
|
|
|
|
# =============================================================================
|
|
# is_wsl() in hermes_constants
|
|
# =============================================================================
|
|
|
|
class TestIsWsl:
|
|
"""Test the shared is_wsl() utility."""
|
|
|
|
def setup_method(self):
|
|
# Reset cached value between tests
|
|
hermes_constants._wsl_detected = None
|
|
|
|
def test_detects_wsl2(self):
|
|
fake_content = (
|
|
"Linux version 5.15.146.1-microsoft-standard-WSL2 "
|
|
"(gcc (GCC) 11.2.0) #1 SMP Thu Jan 11 04:09:03 UTC 2024\n"
|
|
)
|
|
with patch("builtins.open", mock_open(read_data=fake_content)):
|
|
assert hermes_constants.is_wsl() is True
|
|
|
|
def test_detects_wsl1(self):
|
|
fake_content = (
|
|
"Linux version 4.4.0-19041-Microsoft "
|
|
"(Microsoft@Microsoft.com) (gcc version 5.4.0) #1\n"
|
|
)
|
|
with patch("builtins.open", mock_open(read_data=fake_content)):
|
|
assert hermes_constants.is_wsl() is True
|
|
|
|
def test_native_linux(self):
|
|
fake_content = (
|
|
"Linux version 6.5.0-44-generic (buildd@lcy02-amd64-015) "
|
|
"(x86_64-linux-gnu-gcc-12 (Ubuntu 12.3.0-1ubuntu1~22.04) 12.3.0) #44\n"
|
|
)
|
|
with patch("builtins.open", mock_open(read_data=fake_content)):
|
|
assert hermes_constants.is_wsl() is False
|
|
|
|
def test_no_proc_version(self):
|
|
with patch("builtins.open", side_effect=FileNotFoundError):
|
|
assert hermes_constants.is_wsl() is False
|
|
|
|
def test_result_is_cached(self):
|
|
"""After first detection, subsequent calls return the cached value."""
|
|
hermes_constants._wsl_detected = True
|
|
# Even with open raising, cached value is returned
|
|
with patch("builtins.open", side_effect=FileNotFoundError):
|
|
assert hermes_constants.is_wsl() is True
|
|
|
|
|
|
# =============================================================================
|
|
# _wsl_systemd_operational() in gateway
|
|
# =============================================================================
|
|
|
|
class TestWslSystemdOperational:
|
|
"""Test the WSL systemd check."""
|
|
|
|
def test_running(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
gateway.subprocess, "run",
|
|
lambda *a, **kw: SimpleNamespace(
|
|
returncode=0, stdout="running\n", stderr=""
|
|
),
|
|
)
|
|
assert gateway._wsl_systemd_operational() is True
|
|
|
|
def test_degraded(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
gateway.subprocess, "run",
|
|
lambda *a, **kw: SimpleNamespace(
|
|
returncode=1, stdout="degraded\n", stderr=""
|
|
),
|
|
)
|
|
assert gateway._wsl_systemd_operational() is True
|
|
|
|
def test_starting(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
gateway.subprocess, "run",
|
|
lambda *a, **kw: SimpleNamespace(
|
|
returncode=1, stdout="starting\n", stderr=""
|
|
),
|
|
)
|
|
assert gateway._wsl_systemd_operational() is True
|
|
|
|
def test_offline_no_systemd(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
gateway.subprocess, "run",
|
|
lambda *a, **kw: SimpleNamespace(
|
|
returncode=1, stdout="offline\n", stderr=""
|
|
),
|
|
)
|
|
assert gateway._wsl_systemd_operational() is False
|
|
|
|
def test_systemctl_not_found(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
gateway.subprocess, "run",
|
|
MagicMock(side_effect=FileNotFoundError),
|
|
)
|
|
assert gateway._wsl_systemd_operational() is False
|
|
|
|
def test_timeout(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
gateway.subprocess, "run",
|
|
MagicMock(side_effect=subprocess.TimeoutExpired("systemctl", 5)),
|
|
)
|
|
assert gateway._wsl_systemd_operational() is False
|
|
|
|
|
|
# =============================================================================
|
|
# supports_systemd_services() WSL integration
|
|
# =============================================================================
|
|
|
|
class TestSupportsSystemdServicesWSL:
|
|
"""Test that supports_systemd_services() handles WSL correctly."""
|
|
|
|
def test_wsl_with_systemd(self, monkeypatch):
|
|
"""WSL + working systemd → True."""
|
|
monkeypatch.setattr(gateway, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_wsl", lambda: True)
|
|
monkeypatch.setattr(gateway, "_wsl_systemd_operational", lambda: True)
|
|
assert gateway.supports_systemd_services() is True
|
|
|
|
def test_wsl_without_systemd(self, monkeypatch):
|
|
"""WSL + no systemd → False."""
|
|
monkeypatch.setattr(gateway, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_wsl", lambda: True)
|
|
monkeypatch.setattr(gateway, "_wsl_systemd_operational", lambda: False)
|
|
assert gateway.supports_systemd_services() is False
|
|
|
|
def test_native_linux(self, monkeypatch):
|
|
"""Native Linux (not WSL, not container) → True without further probing."""
|
|
monkeypatch.setattr(gateway, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_wsl", lambda: False)
|
|
# supports_systemd_services() now also branches on is_container() to
|
|
# decide whether to probe `systemctl is-system-running` — explicitly
|
|
# opt this case out of the container path so a containerized CI
|
|
# runner doesn't inherit the probe of the runner image's systemd.
|
|
monkeypatch.setattr(gateway, "is_container", lambda: False)
|
|
# On macOS dev boxes shutil.which("systemctl") returns None; stub it
|
|
# so the test exercises the native-Linux branch independently of the
|
|
# host's $PATH.
|
|
monkeypatch.setattr(gateway.shutil, "which", lambda name: "/usr/bin/systemctl")
|
|
assert gateway.supports_systemd_services() is True
|
|
|
|
def test_termux_still_excluded(self, monkeypatch):
|
|
"""Termux → False regardless of WSL status."""
|
|
monkeypatch.setattr(gateway, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: True)
|
|
assert gateway.supports_systemd_services() is False
|
|
|
|
|
|
# =============================================================================
|
|
# WSL messaging in gateway commands
|
|
# =============================================================================
|
|
|
|
class TestGatewayCommandWSLMessages:
|
|
"""Test that WSL users see appropriate guidance."""
|
|
|
|
def test_install_wsl_no_systemd(self, monkeypatch, capsys):
|
|
"""hermes gateway install on WSL without systemd shows guidance."""
|
|
monkeypatch.setattr(gateway, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_wsl", lambda: True)
|
|
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_managed", lambda: False)
|
|
|
|
args = SimpleNamespace(
|
|
gateway_command="install", force=False, system=False,
|
|
run_as_user=None,
|
|
)
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
gateway.gateway_command(args)
|
|
assert exc_info.value.code == 1
|
|
|
|
out = capsys.readouterr().out
|
|
assert "WSL detected" in out
|
|
assert "systemd is not running" in out
|
|
assert "hermes gateway run" in out
|
|
assert "tmux" in out
|
|
|
|
def test_start_wsl_no_systemd(self, monkeypatch, capsys):
|
|
"""hermes gateway start on WSL without systemd shows guidance."""
|
|
monkeypatch.setattr(gateway, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_wsl", lambda: True)
|
|
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_macos", lambda: False)
|
|
|
|
args = SimpleNamespace(gateway_command="start", system=False)
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
gateway.gateway_command(args)
|
|
assert exc_info.value.code == 1
|
|
|
|
out = capsys.readouterr().out
|
|
assert "WSL detected" in out
|
|
assert "hermes gateway run" in out
|
|
assert "wsl.conf" in out
|
|
|
|
def test_install_wsl_with_systemd_warns(self, monkeypatch, capsys):
|
|
"""hermes gateway install on WSL with systemd shows warning but proceeds."""
|
|
monkeypatch.setattr(gateway, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_wsl", lambda: True)
|
|
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: True)
|
|
monkeypatch.setattr(gateway, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_managed", lambda: False)
|
|
|
|
# Mock systemd_install to capture call
|
|
install_called = []
|
|
monkeypatch.setattr(
|
|
gateway, "systemd_install",
|
|
lambda **kwargs: install_called.append(kwargs),
|
|
)
|
|
|
|
args = SimpleNamespace(
|
|
gateway_command="install", force=False, system=False,
|
|
run_as_user=None,
|
|
)
|
|
gateway.gateway_command(args)
|
|
|
|
out = capsys.readouterr().out
|
|
assert "WSL detected" in out
|
|
assert "may not survive WSL restarts" in out
|
|
assert len(install_called) == 1 # install still proceeded
|
|
|
|
def test_status_wsl_running_manual(self, monkeypatch, capsys):
|
|
"""hermes gateway status on WSL with manual process shows WSL note."""
|
|
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_wsl", lambda: True)
|
|
monkeypatch.setattr(gateway, "find_gateway_pids", lambda: [12345])
|
|
monkeypatch.setattr(gateway, "_runtime_health_lines", lambda: [])
|
|
# Stub out the systemd unit path check
|
|
monkeypatch.setattr(
|
|
gateway, "get_systemd_unit_path",
|
|
lambda system=False: SimpleNamespace(exists=lambda: False),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway, "get_launchd_plist_path",
|
|
lambda: SimpleNamespace(exists=lambda: False),
|
|
)
|
|
|
|
args = SimpleNamespace(gateway_command="status", deep=False, system=False)
|
|
gateway.gateway_command(args)
|
|
|
|
out = capsys.readouterr().out
|
|
assert "WSL note" in out
|
|
assert "tmux or screen" in out
|
|
|
|
def test_status_wsl_not_running(self, monkeypatch, capsys):
|
|
"""hermes gateway status on WSL with no process shows WSL start advice."""
|
|
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_termux", lambda: False)
|
|
monkeypatch.setattr(gateway, "is_wsl", lambda: True)
|
|
monkeypatch.setattr(gateway, "find_gateway_pids", lambda: [])
|
|
monkeypatch.setattr(gateway, "_runtime_health_lines", lambda: [])
|
|
monkeypatch.setattr(
|
|
gateway, "get_systemd_unit_path",
|
|
lambda system=False: SimpleNamespace(exists=lambda: False),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway, "get_launchd_plist_path",
|
|
lambda: SimpleNamespace(exists=lambda: False),
|
|
)
|
|
|
|
args = SimpleNamespace(gateway_command="status", deep=False, system=False)
|
|
gateway.gateway_command(args)
|
|
|
|
out = capsys.readouterr().out
|
|
assert "hermes gateway run" in out
|
|
assert "tmux" in out
|