"""Tests for hermes_cli.service_manager — the abstract ServiceManager protocol, the detect_service_manager() entry point, and the host-side adapter wrappers (Systemd / Launchd / Windows). The s6 backend is added in Phase 3; its tests live alongside the implementation in this same file once that phase ships. """ from __future__ import annotations import pytest from hermes_cli.service_manager import ( LaunchdServiceManager, S6ServiceManager, ServiceManager, ServiceManagerKind, SystemdServiceManager, WindowsServiceManager, detect_service_manager, get_service_manager, validate_profile_name, ) # --------------------------------------------------------------------------- # validate_profile_name # --------------------------------------------------------------------------- def test_validate_profile_name_accepts_valid_names() -> None: # Smoke: known-good names should not raise. validate_profile_name("coder") validate_profile_name("my-profile") validate_profile_name("assistant_v2") validate_profile_name("a") validate_profile_name("0") validate_profile_name("0abc") @pytest.mark.parametrize( "bad", [ "", # empty "Coder", # uppercase "foo/bar", # path traversal "../escape", # path traversal "-leading-dash", # leading dash (s6 reads as a flag) "_leading_underscore", # leading underscore "name with spaces", # whitespace "name.with.dots", # punctuation "a" * 252, # too long ], ) def test_validate_profile_name_rejects_invalid(bad: str) -> None: with pytest.raises(ValueError): validate_profile_name(bad) # --------------------------------------------------------------------------- # detect_service_manager # --------------------------------------------------------------------------- def test_detect_service_manager_returns_known_value() -> None: """Without mocking, the function must still return one of the advertised literals — anything else means a new platform branch was added without updating ServiceManagerKind.""" result = detect_service_manager() assert result in ("systemd", "launchd", "windows", "s6", "none") # --------------------------------------------------------------------------- # _s6_running — must work for unprivileged users, not just root # --------------------------------------------------------------------------- def _patch_s6_paths( monkeypatch: pytest.MonkeyPatch, *, comm: str | OSError | None, basedir_is_dir: bool, ) -> None: """Stub /proc/1/comm and /run/s6/basedir for _s6_running tests.""" from pathlib import Path as _Path real_read_text = _Path.read_text real_is_dir = _Path.is_dir def fake_read_text(self, *args, **kwargs): # type: ignore[override] if str(self) == "/proc/1/comm": if isinstance(comm, OSError): raise comm if comm is None: raise FileNotFoundError(2, "No such file or directory") return comm + "\n" return real_read_text(self, *args, **kwargs) def fake_is_dir(self): # type: ignore[override] if str(self) == "/run/s6/basedir": return basedir_is_dir return real_is_dir(self) monkeypatch.setattr(_Path, "read_text", fake_read_text) monkeypatch.setattr(_Path, "is_dir", fake_is_dir) def test_s6_running_true_when_comm_and_basedir_match( monkeypatch: pytest.MonkeyPatch, ) -> None: from hermes_cli.service_manager import _s6_running _patch_s6_paths(monkeypatch, comm="s6-svscan", basedir_is_dir=True) assert _s6_running() is True def test_s6_running_false_when_comm_is_wrong( monkeypatch: pytest.MonkeyPatch, ) -> None: from hermes_cli.service_manager import _s6_running # systemd as PID 1, basedir present from some stray s6 install _patch_s6_paths(monkeypatch, comm="systemd", basedir_is_dir=True) assert _s6_running() is False def test_s6_running_false_when_basedir_missing( monkeypatch: pytest.MonkeyPatch, ) -> None: from hermes_cli.service_manager import _s6_running # The comm matches but the basedir is missing — e.g. an unrelated # process happens to be named "s6-svscan" _patch_s6_paths(monkeypatch, comm="s6-svscan", basedir_is_dir=False) assert _s6_running() is False def test_s6_running_false_when_comm_unreadable( monkeypatch: pytest.MonkeyPatch, ) -> None: """Regression: /proc/1/exe was unreadable to UID 10000 and resolve() silently returned the unresolved path, making detection always-False inside the container under the hermes user. The new probe must FAIL CLOSED — not raise — when /proc/1/comm can't be read. """ from hermes_cli.service_manager import _s6_running _patch_s6_paths( monkeypatch, comm=PermissionError(13, "Permission denied"), basedir_is_dir=True, ) assert _s6_running() is False def test_s6_running_handles_missing_proc( monkeypatch: pytest.MonkeyPatch, ) -> None: """On macOS / Windows / WSL-without-procfs, /proc/1/comm doesn't exist. Must return False, not raise.""" from hermes_cli.service_manager import _s6_running _patch_s6_paths(monkeypatch, comm=None, basedir_is_dir=False) assert _s6_running() is False # --------------------------------------------------------------------------- # Backend wrappers — kind + registration unsupported on hosts # --------------------------------------------------------------------------- def test_systemd_manager_kind_and_registration_unsupported() -> None: mgr = SystemdServiceManager() assert mgr.kind == "systemd" assert mgr.supports_runtime_registration() is False with pytest.raises(NotImplementedError): mgr.register_profile_gateway("foo") with pytest.raises(NotImplementedError): mgr.unregister_profile_gateway("foo") assert mgr.list_profile_gateways() == [] # Protocol conformance — runtime_checkable lets us assert this. assert isinstance(mgr, ServiceManager) def test_launchd_manager_kind_and_registration_unsupported() -> None: mgr = LaunchdServiceManager() assert mgr.kind == "launchd" assert mgr.supports_runtime_registration() is False with pytest.raises(NotImplementedError): mgr.register_profile_gateway("foo") assert mgr.list_profile_gateways() == [] assert isinstance(mgr, ServiceManager) def test_windows_manager_kind_and_registration_unsupported() -> None: mgr = WindowsServiceManager() assert mgr.kind == "windows" assert mgr.supports_runtime_registration() is False with pytest.raises(NotImplementedError): mgr.register_profile_gateway("foo") assert isinstance(mgr, ServiceManager) # --------------------------------------------------------------------------- # Lifecycle delegation — wrappers must call through to module-level fns # --------------------------------------------------------------------------- def test_systemd_manager_lifecycle_delegates(monkeypatch: pytest.MonkeyPatch) -> None: called: list[str] = [] monkeypatch.setattr( "hermes_cli.gateway.systemd_start", lambda: called.append("start"), ) monkeypatch.setattr( "hermes_cli.gateway.systemd_stop", lambda: called.append("stop"), ) monkeypatch.setattr( "hermes_cli.gateway.systemd_restart", lambda: called.append("restart"), ) monkeypatch.setattr( "hermes_cli.gateway._probe_systemd_service_running", lambda *a, **kw: (False, True), ) mgr = SystemdServiceManager() mgr.start("ignored") mgr.stop("ignored") mgr.restart("ignored") assert called == ["start", "stop", "restart"] assert mgr.is_running("ignored") is True def test_launchd_manager_lifecycle_delegates(monkeypatch: pytest.MonkeyPatch) -> None: called: list[str] = [] monkeypatch.setattr( "hermes_cli.gateway.launchd_start", lambda: called.append("start"), ) monkeypatch.setattr( "hermes_cli.gateway.launchd_stop", lambda: called.append("stop"), ) monkeypatch.setattr( "hermes_cli.gateway.launchd_restart", lambda: called.append("restart"), ) monkeypatch.setattr( "hermes_cli.gateway._probe_launchd_service_running", lambda: False, ) mgr = LaunchdServiceManager() mgr.start("ignored") mgr.stop("ignored") mgr.restart("ignored") assert called == ["start", "stop", "restart"] assert mgr.is_running("ignored") is False def test_windows_manager_lifecycle_delegates(monkeypatch: pytest.MonkeyPatch) -> None: called: list[str] = [] # Force-import the submodule so monkeypatch's attribute lookup # against the `hermes_cli` package succeeds — gateway_windows is # imported lazily inside the wrapper and may not yet be loaded. import hermes_cli.gateway_windows # noqa: F401 class _FakeWindowsModule: @staticmethod def start() -> None: called.append("start") @staticmethod def stop() -> None: called.append("stop") @staticmethod def restart() -> None: called.append("restart") @staticmethod def is_installed() -> bool: return True monkeypatch.setattr("hermes_cli.gateway_windows", _FakeWindowsModule) monkeypatch.setattr( "hermes_cli.gateway.find_gateway_pids", lambda **kw: [12345], ) mgr = WindowsServiceManager() mgr.start("ignored") mgr.stop("ignored") mgr.restart("ignored") assert called == ["start", "stop", "restart"] assert mgr.is_running("ignored") is True def test_windows_manager_is_running_false_when_not_installed( monkeypatch: pytest.MonkeyPatch, ) -> None: import hermes_cli.gateway_windows # noqa: F401 class _FakeWindowsModule: @staticmethod def is_installed() -> bool: return False monkeypatch.setattr("hermes_cli.gateway_windows", _FakeWindowsModule) monkeypatch.setattr( "hermes_cli.gateway.find_gateway_pids", lambda **kw: [12345], # PIDs would otherwise vote "running" ) assert WindowsServiceManager().is_running("ignored") is False def test_windows_manager_install_forwards_kwargs(monkeypatch: pytest.MonkeyPatch) -> None: captured: dict[str, object] = {} import hermes_cli.gateway_windows # noqa: F401 class _FakeWindowsModule: @staticmethod def install(*, force, start_now, start_on_login, elevated_handoff) -> None: captured["force"] = force captured["start_now"] = start_now captured["start_on_login"] = start_on_login captured["elevated_handoff"] = elevated_handoff monkeypatch.setattr("hermes_cli.gateway_windows", _FakeWindowsModule) WindowsServiceManager().install( force=True, start_now=True, start_on_login=False, elevated_handoff=True, ) assert captured == { "force": True, "start_now": True, "start_on_login": False, "elevated_handoff": True, } # --------------------------------------------------------------------------- # get_service_manager factory # --------------------------------------------------------------------------- @pytest.mark.parametrize( "kind,cls", [ ("systemd", SystemdServiceManager), ("launchd", LaunchdServiceManager), ("windows", WindowsServiceManager), ], ) def test_get_service_manager_returns_correct_backend( monkeypatch: pytest.MonkeyPatch, kind: ServiceManagerKind, cls: type, ) -> None: monkeypatch.setattr( "hermes_cli.service_manager.detect_service_manager", lambda: kind, ) assert isinstance(get_service_manager(), cls) def test_get_service_manager_raises_when_unsupported( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( "hermes_cli.service_manager.detect_service_manager", lambda: "none", ) with pytest.raises(RuntimeError, match="no supported service manager"): get_service_manager() def test_get_service_manager_returns_s6_instance( monkeypatch: pytest.MonkeyPatch, ) -> None: """The s6 backend ships in Phase 3 — the factory must return an S6ServiceManager when running inside a container.""" monkeypatch.setattr( "hermes_cli.service_manager.detect_service_manager", lambda: "s6", ) assert isinstance(get_service_manager(), S6ServiceManager) # --------------------------------------------------------------------------- # S6ServiceManager — unit tests against a tmp-path scandir (no real s6) # --------------------------------------------------------------------------- @pytest.fixture def s6_scandir(tmp_path): """Empty scandir for the S6ServiceManager tests.""" d = tmp_path / "service" d.mkdir() return d @pytest.fixture def fake_subprocess_run(monkeypatch: pytest.MonkeyPatch): """Capture subprocess.run calls + always return success. Lets the S6ServiceManager tests run on hosts that don't have s6-svc / s6-svscanctl installed. Records are normalized: leading ``/command/`` is stripped from cmd[0] so assertions can match on the bare s6-svc / s6-svstat / s6-svscanctl name regardless of whether the manager calls them via absolute path or bare name.""" calls: list[list[str]] = [] def _fake(cmd, **kw): import subprocess as _sp seq = list(cmd) if isinstance(cmd, (list, tuple)) else [str(cmd)] if seq and seq[0].startswith("/command/"): seq[0] = seq[0][len("/command/"):] calls.append(seq) return _sp.CompletedProcess(cmd, 0, "", "") monkeypatch.setattr("subprocess.run", _fake) return calls def test_s6_manager_kind_and_supports_registration() -> None: mgr = S6ServiceManager() assert mgr.kind == "s6" assert mgr.supports_runtime_registration() is True # --------------------------------------------------------------------------- # _seed_supervise_skeleton — unit tests # --------------------------------------------------------------------------- # # The skeleton helper pre-creates the dirs and FIFOs that s6-supervise # would otherwise create as root mode 0700, locking out the # unprivileged hermes user from every lifecycle op. These tests run # against tmp_path and assert the produced layout — the live-container # verification (against real s6-svc / s6-svstat) lives in # tests/docker/test_s6_profile_gateway_integration.py. def test_seed_supervise_skeleton_creates_expected_layout(tmp_path) -> None: """Verifies the dirs + FIFO + modes the helper lays down.""" import stat from hermes_cli.service_manager import _seed_supervise_skeleton svc_dir = tmp_path / "gateway-foo" svc_dir.mkdir() _seed_supervise_skeleton(svc_dir) # Top-level event/ — s6-svlisten1 event subscription dir. event = svc_dir / "event" assert event.is_dir(), "missing top-level event/" assert stat.S_IMODE(event.stat().st_mode) == 0o3730, ( f"event/ mode = {oct(event.stat().st_mode)}, want 03730" ) # supervise/ dir. supervise = svc_dir / "supervise" assert supervise.is_dir(), "missing supervise/" assert stat.S_IMODE(supervise.stat().st_mode) == 0o755 # supervise/event/. supervise_event = supervise / "event" assert supervise_event.is_dir(), "missing supervise/event/" assert stat.S_IMODE(supervise_event.stat().st_mode) == 0o3730 # supervise/control FIFO. control = supervise / "control" assert control.exists(), "missing supervise/control FIFO" assert stat.S_ISFIFO(control.stat().st_mode), ( "supervise/control must be a FIFO" ) assert stat.S_IMODE(control.stat().st_mode) == 0o660 def test_seed_supervise_skeleton_handles_log_subservice(tmp_path) -> None: """When a log/ subdir exists, its supervise tree also gets seeded. Without this, ``unregister_profile_gateway``'s rmtree would EACCES on the logger's root-owned supervise dir even after the parent slot's supervise/ was hermes-owned. """ import stat from hermes_cli.service_manager import _seed_supervise_skeleton svc_dir = tmp_path / "gateway-foo" svc_dir.mkdir() (svc_dir / "log").mkdir() # logger subdir present _seed_supervise_skeleton(svc_dir) # Logger's own supervise tree is seeded the same way. log_event = svc_dir / "log" / "event" log_supervise = svc_dir / "log" / "supervise" log_supervise_event = log_supervise / "event" log_control = log_supervise / "control" assert log_event.is_dir() assert stat.S_IMODE(log_event.stat().st_mode) == 0o3730 assert log_supervise.is_dir() assert log_supervise_event.is_dir() assert log_control.exists() and stat.S_ISFIFO(log_control.stat().st_mode) def test_seed_supervise_skeleton_skips_when_no_log_subservice(tmp_path) -> None: """If log/ isn't present, no logger skeleton is created.""" from hermes_cli.service_manager import _seed_supervise_skeleton svc_dir = tmp_path / "gateway-foo" svc_dir.mkdir() _seed_supervise_skeleton(svc_dir) assert not (svc_dir / "log").exists(), ( "helper must not synthesize a log/ subdir on its own" ) def test_seed_supervise_skeleton_is_idempotent(tmp_path) -> None: """Calling the helper twice on the same dir is a no-op the second time. Important because s6-supervise may have already opened the FIFO when a re-register / reconcile happens; double-creation would error out. The helper short-circuits on existence. """ from hermes_cli.service_manager import _seed_supervise_skeleton svc_dir = tmp_path / "gateway-foo" svc_dir.mkdir() _seed_supervise_skeleton(svc_dir) _seed_supervise_skeleton(svc_dir) # must not raise def test_s6_register_creates_service_dir_and_triggers_scan( s6_scandir, fake_subprocess_run, ) -> None: mgr = S6ServiceManager(scandir=s6_scandir) mgr.register_profile_gateway("coder") svc_dir = s6_scandir / "gateway-coder" assert svc_dir.is_dir() assert (svc_dir / "type").read_text().strip() == "longrun" run_path = svc_dir / "run" assert run_path.is_file() assert run_path.stat().st_mode & 0o111 # executable run_text = run_path.read_text() assert "export HOME=/opt/data" in run_text assert "hermes -p coder gateway run" in run_text assert "s6-setuidgid hermes" in run_text # Sentinel marking this as the supervised-child invocation. Without # it, the supervised `gateway run` would re-enter the s6 redirect # in `_gateway_command_inner` and recurse. See the matching guard # in hermes_cli/gateway.py::_gateway_command_inner. assert "export HERMES_S6_SUPERVISED_CHILD=1" in run_text log_run = svc_dir / "log" / "run" assert log_run.is_file() log_text = log_run.read_text() # CRITICAL: HERMES_HOME must be a runtime env-var expansion, NOT # a Python-substituted absolute path. Negative-assert the wrong # form so future regressions are caught. assert "$HERMES_HOME" in log_text assert "logs/gateways/coder" in log_text assert "/opt/data/logs/gateways/coder" not in log_text, ( "log_dir was hard-coded; must use ${HERMES_HOME} at run time" ) # `1` action directive forwards lines to stdout BEFORE the file # destination so the supervised gateway's stdout (including the # rich-console banner and plain print() output) reaches docker # logs, not just the rotated file. See _render_log_run's docstring # for the full output-routing rationale. assert "s6-log 1 " in log_text, ( "log/run must include the `1` action directive before the file " "destination so supervised stdout reaches docker logs. Saw: " f"{log_text!r}" ) # s6-svscanctl -a was invoked against the scandir assert any( cmd[0] == "s6-svscanctl" and "-a" in cmd and str(s6_scandir) in cmd for cmd in fake_subprocess_run ), f"s6-svscanctl -a not invoked; saw: {fake_subprocess_run}" def test_s6_register_extra_env_is_quoted(s6_scandir, fake_subprocess_run) -> None: mgr = S6ServiceManager(scandir=s6_scandir) mgr.register_profile_gateway( "x", extra_env={"FOO": "bar baz", "QUOTED": "a'b"}, ) run_text = (s6_scandir / "gateway-x" / "run").read_text() # shlex.quote should have wrapped both values assert "export FOO='bar baz'" in run_text assert "export QUOTED='a'\"'\"'b'" in run_text def test_render_run_script_resets_home_before_exec() -> None: run_text = S6ServiceManager._render_run_script("coder", {}) assert "export HOME=/opt/data" in run_text assert "exec s6-setuidgid hermes hermes -p coder gateway run" in run_text def test_s6_register_rejects_invalid_profile_name(s6_scandir) -> None: mgr = S6ServiceManager(scandir=s6_scandir) with pytest.raises(ValueError): mgr.register_profile_gateway("Bad/Name") def test_s6_register_rejects_duplicate(s6_scandir, fake_subprocess_run) -> None: mgr = S6ServiceManager(scandir=s6_scandir) (s6_scandir / "gateway-coder").mkdir(parents=True) with pytest.raises(ValueError, match="already registered"): mgr.register_profile_gateway("coder") def test_s6_register_rolls_back_on_svscanctl_failure( s6_scandir, monkeypatch: pytest.MonkeyPatch, ) -> None: """If s6-svscanctl fails the service dir must be cleaned up so the next register call doesn't see a stale duplicate.""" import subprocess as _sp def _fail_scanctl(cmd, **kw): # Manager calls s6-svscanctl by absolute path; match on basename. if cmd[0].endswith("/s6-svscanctl"): return _sp.CompletedProcess(cmd, 1, "", "rescan failed") return _sp.CompletedProcess(cmd, 0, "", "") monkeypatch.setattr("subprocess.run", _fail_scanctl) mgr = S6ServiceManager(scandir=s6_scandir) with pytest.raises(RuntimeError, match="s6-svscanctl failed"): mgr.register_profile_gateway("coder") assert not (s6_scandir / "gateway-coder").exists() def test_s6_unregister_removes_service_dir( s6_scandir, fake_subprocess_run, ) -> None: svc_dir = s6_scandir / "gateway-coder" svc_dir.mkdir(parents=True) (svc_dir / "type").write_text("longrun\n") mgr = S6ServiceManager(scandir=s6_scandir) mgr.unregister_profile_gateway("coder") # s6-svc -d was issued assert any( cmd[0] == "s6-svc" and "-d" in cmd for cmd in fake_subprocess_run ) # Service dir was removed assert not svc_dir.exists() # Rescan was triggered assert any(cmd[0] == "s6-svscanctl" for cmd in fake_subprocess_run) def test_s6_unregister_absent_profile_is_noop(s6_scandir) -> None: # Should NOT raise even though "ghost" doesn't exist S6ServiceManager(scandir=s6_scandir).unregister_profile_gateway("ghost") def test_s6_list_profile_gateways(s6_scandir) -> None: # Three gateway profiles + one unrelated service + one hidden dir (s6_scandir / "gateway-coder").mkdir() (s6_scandir / "gateway-assistant").mkdir() (s6_scandir / "gateway-writer").mkdir() (s6_scandir / "s6-linux-init-shutdownd").mkdir() # filtered out (s6_scandir / ".lock").mkdir() # filtered out (hidden) profiles = sorted(S6ServiceManager(scandir=s6_scandir).list_profile_gateways()) assert profiles == ["assistant", "coder", "writer"] def test_s6_list_profile_gateways_empty_when_scandir_missing(tmp_path) -> None: missing = tmp_path / "does-not-exist" assert S6ServiceManager(scandir=missing).list_profile_gateways() == [] def test_s6_lifecycle_dispatches_to_s6_svc( s6_scandir, fake_subprocess_run, ) -> None: mgr = S6ServiceManager(scandir=s6_scandir) # _run_svc now verifies the slot exists before invoking s6-svc, so # we have to pre-seed the dir. In real use the slot is created by # register_profile_gateway or the cont-init.d reconciler. (s6_scandir / "gateway-coder").mkdir() mgr.start("gateway-coder") mgr.stop("gateway-coder") mgr.restart("gateway-coder") flags = [c[1] for c in fake_subprocess_run if c[0] == "s6-svc"] assert flags == ["-u", "-d", "-t"] # --------------------------------------------------------------------------- # Lifecycle errors — friendly messages, not raw CalledProcessError # --------------------------------------------------------------------------- def test_lifecycle_raises_gateway_not_registered_for_missing_slot( s6_scandir, fake_subprocess_run, ) -> None: """When the service slot doesn't exist, the lifecycle methods must raise GatewayNotRegisteredError BEFORE invoking s6-svc, so the user sees a clear 'no such gateway' message instead of an opaque CalledProcessError stacktrace.""" from hermes_cli.service_manager import ( GatewayNotRegisteredError, ) mgr = S6ServiceManager(scandir=s6_scandir) # No gateway-typo/ directory exists — slot is missing. with pytest.raises(GatewayNotRegisteredError) as excinfo: mgr.start("gateway-typo") assert excinfo.value.profile == "typo" assert excinfo.value.service == "gateway-typo" msg = str(excinfo.value) assert "'typo'" in msg assert "hermes profile create typo" in msg # And critically: s6-svc was NOT invoked. assert not any(c[0] == "s6-svc" for c in fake_subprocess_run) @pytest.mark.parametrize("action,method_name", [ ("start", "start"), ("stop", "stop"), ("restart", "restart"), ]) def test_all_lifecycle_methods_check_for_missing_slot( s6_scandir, fake_subprocess_run, action: str, method_name: str, ) -> None: """start/stop/restart all check for missing slots the same way.""" from hermes_cli.service_manager import ( GatewayNotRegisteredError, ) mgr = S6ServiceManager(scandir=s6_scandir) with pytest.raises(GatewayNotRegisteredError): getattr(mgr, method_name)("gateway-absent") def test_gateway_not_registered_unprefixed_service_name(s6_scandir) -> None: """If the caller passes a name without the 'gateway-' prefix (the Protocol allows arbitrary service names), the error still carries that name verbatim as the 'profile' so error messages don't accidentally strip user-provided text.""" from hermes_cli.service_manager import ( GatewayNotRegisteredError, ) mgr = S6ServiceManager(scandir=s6_scandir) with pytest.raises(GatewayNotRegisteredError) as excinfo: mgr.start("not-prefixed") assert excinfo.value.profile == "not-prefixed" def test_lifecycle_raises_s6_command_error_on_subprocess_failure( s6_scandir, monkeypatch: pytest.MonkeyPatch, ) -> None: """When s6-svc itself fails (non-zero exit) — e.g. EACCES on the supervise control FIFO — the lifecycle methods translate the CalledProcessError into a named S6CommandError carrying the return code and stderr.""" import subprocess as _sp from hermes_cli.service_manager import S6CommandError # Pre-create the slot so we reach the s6-svc call. (s6_scandir / "gateway-coder").mkdir() def _fail(cmd, **kw): raise _sp.CalledProcessError( returncode=111, cmd=cmd, stderr="s6-svc: fatal: unable to control supervise/control: " "Permission denied\n", ) monkeypatch.setattr("subprocess.run", _fail) mgr = S6ServiceManager(scandir=s6_scandir) with pytest.raises(S6CommandError) as excinfo: mgr.start("gateway-coder") assert excinfo.value.service == "gateway-coder" assert excinfo.value.action == "start" assert excinfo.value.returncode == 111 assert "Permission denied" in excinfo.value.stderr assert "Permission denied" in str(excinfo.value) assert "rc=111" in str(excinfo.value) def test_s6_is_running_parses_svstat( s6_scandir, monkeypatch: pytest.MonkeyPatch, ) -> None: import subprocess as _sp def _svstat(cmd, **kw): if cmd[0].endswith("/s6-svstat"): return _sp.CompletedProcess(cmd, 0, "up (pid 42) 17 seconds\n", "") return _sp.CompletedProcess(cmd, 0, "", "") monkeypatch.setattr("subprocess.run", _svstat) assert S6ServiceManager(scandir=s6_scandir).is_running("gateway-coder") is True def _svstat_down(cmd, **kw): if cmd[0].endswith("/s6-svstat"): return _sp.CompletedProcess(cmd, 0, "down 5 seconds\n", "") return _sp.CompletedProcess(cmd, 0, "", "") monkeypatch.setattr("subprocess.run", _svstat_down) assert S6ServiceManager(scandir=s6_scandir).is_running("gateway-coder") is False # --------------------------------------------------------------------------- # S6 stop writes a planned-stop marker (issue #42675) # # `hermes gateway stop` inside a container dispatches through # S6ServiceManager.stop() -> `s6-svc -d`, which SIGTERMs the gateway. # That SIGTERM is indistinguishable from the one s6/Docker sends on a # container restart unless we mark the intentional stop first. Without # the marker, the gateway's shutdown handler can't tell an operator # stop from a restart kill, and the gateway_state=stopped suppression # (run.py) would never engage for explicit stops. # --------------------------------------------------------------------------- def test_s6_supervised_pid_parses_svstat(monkeypatch, s6_scandir): """_supervised_pid extracts the PID from `up (pid NNNN) ...`.""" import subprocess as _sp def _fake(cmd, **kw): return _sp.CompletedProcess(cmd, 0, "up (pid 4242) 17 seconds\n", "") monkeypatch.setattr("subprocess.run", _fake) mgr = S6ServiceManager(scandir=s6_scandir) assert mgr._supervised_pid("gateway-coder") == 4242 def test_s6_supervised_pid_none_when_down(monkeypatch, s6_scandir): """A down service (`s6-svstat` rc!=0 or no pid) yields None.""" import subprocess as _sp def _fake(cmd, **kw): return _sp.CompletedProcess(cmd, 0, "down (exitcode 0) 3 seconds\n", "") monkeypatch.setattr("subprocess.run", _fake) mgr = S6ServiceManager(scandir=s6_scandir) assert mgr._supervised_pid("gateway-coder") is None def test_s6_stop_writes_planned_stop_marker(monkeypatch, s6_scandir): """stop() must mark the supervised PID before `s6-svc -d` so the gateway recognises the SIGTERM as an intentional stop (#42675).""" import subprocess as _sp svc_dir = s6_scandir / "gateway-coder" svc_dir.mkdir() # so _run_svc doesn't raise GatewayNotRegisteredError svc_calls: list[list[str]] = [] def _fake(cmd, **kw): seq = list(cmd) if isinstance(cmd, (list, tuple)) else [str(cmd)] if seq and seq[0].startswith("/command/"): seq[0] = seq[0][len("/command/"):] svc_calls.append(seq) if seq and seq[0] == "s6-svstat": return _sp.CompletedProcess(cmd, 0, "up (pid 9090) 5 seconds\n", "") return _sp.CompletedProcess(cmd, 0, "", "") monkeypatch.setattr("subprocess.run", _fake) marked: list[int] = [] monkeypatch.setattr( "gateway.status.write_planned_stop_marker", lambda pid: marked.append(pid) or True, ) mgr = S6ServiceManager(scandir=s6_scandir) mgr.stop("gateway-coder") assert marked == [9090], ( f"stop() must write the planned-stop marker for the supervised PID; " f"marked={marked}" ) # And it must still issue the down command. assert any( cmd[0] == "s6-svc" and "-d" in cmd for cmd in svc_calls ), f"s6-svc -d not invoked; saw: {svc_calls}" def test_s6_stop_tolerates_marker_write_failure(monkeypatch, s6_scandir): """A marker-write failure must not block the stop (best-effort).""" import subprocess as _sp svc_dir = s6_scandir / "gateway-coder" svc_dir.mkdir() svc_calls: list[list[str]] = [] def _fake(cmd, **kw): seq = list(cmd) if isinstance(cmd, (list, tuple)) else [str(cmd)] if seq and seq[0].startswith("/command/"): seq[0] = seq[0][len("/command/"):] svc_calls.append(seq) if seq and seq[0] == "s6-svstat": return _sp.CompletedProcess(cmd, 0, "up (pid 9090) 5 seconds\n", "") return _sp.CompletedProcess(cmd, 0, "", "") monkeypatch.setattr("subprocess.run", _fake) def _boom(pid): raise OSError("disk full") monkeypatch.setattr("gateway.status.write_planned_stop_marker", _boom) mgr = S6ServiceManager(scandir=s6_scandir) mgr.stop("gateway-coder") # must not raise assert any(cmd[0] == "s6-svc" and "-d" in cmd for cmd in svc_calls)