diff --git a/hermes_cli/service_manager.py b/hermes_cli/service_manager.py index f6a28a8ec3c..71dc6ae1888 100644 --- a/hermes_cli/service_manager.py +++ b/hermes_cli/service_manager.py @@ -279,9 +279,7 @@ def get_service_manager() -> ServiceManager: """Return the ServiceManager instance for the current environment. Raises: - RuntimeError: when no supported backend is available, or when - the detected backend's implementation hasn't shipped yet - (the s6 backend lands in Phase 3). + RuntimeError: when no supported backend is available. """ kind = detect_service_manager() if kind == "systemd": @@ -291,6 +289,283 @@ def get_service_manager() -> ServiceManager: if kind == "windows": return WindowsServiceManager() if kind == "s6": - # Phase 3 will replace this with `return S6ServiceManager()`. - raise RuntimeError("s6 backend not yet implemented (Phase 3)") + return S6ServiceManager() raise RuntimeError("no supported service manager detected") + + +# --------------------------------------------------------------------------- +# S6ServiceManager (container-only) +# +# Per-profile gateways are registered dynamically when `hermes profile create` +# runs inside the container (Phase 4). Static services (main-hermes, dashboard) +# live in /etc/s6-overlay/s6-rc.d/ and are NOT managed by this class — they're +# part of the image, not runtime-created. +# --------------------------------------------------------------------------- + + +# s6-overlay's dynamic scandir for runtime-registered services. Lives on +# tmpfs and is the directory s6-svscan watches. Writes here trigger +# automatic supervision on the next rescan. +S6_DYNAMIC_SCANDIR = Path("/run/service") +S6_SERVICE_PREFIX = "gateway-" + +# s6-overlay installs its binaries under /command/ and only adds that +# directory to PATH for processes started under the supervision tree +# (services started by s6-svscan, cont-init.d scripts, etc.). Code +# that runs via `docker exec` or any other out-of-tree entry point — +# notably our Phase 4 profile create/delete hooks — inherits the +# container's base PATH which does NOT include /command/. +# +# Rather than asking every caller to fix up its environment, the +# S6ServiceManager calls s6-* binaries by absolute path via this +# constant. We don't use `/usr/bin/s6-…` symlinks because the +# s6-overlay-symlinks-noarch tarball only links a subset, and we +# want every s6 invocation to be guaranteed-findable. +_S6_BIN_DIR = "/command" + + +class S6ServiceManager: + """Per-profile gateway supervision via s6-overlay. + + Only handles runtime-registered services under + ``S6_DYNAMIC_SCANDIR``. Static services (main-hermes, dashboard) + are managed by s6-rc at image-build time and are out of scope. + """ + + kind: ServiceManagerKind = "s6" + + def __init__(self, scandir: Path = S6_DYNAMIC_SCANDIR) -> None: + self.scandir = scandir + + # -- internal helpers -------------------------------------------------- + + def _service_dir(self, profile: str) -> Path: + validate_profile_name(profile) + return self.scandir / f"{S6_SERVICE_PREFIX}{profile}" + + def _service_name(self, profile: str) -> str: + return f"{S6_SERVICE_PREFIX}{profile}" + + @staticmethod + def _render_run_script( + profile: str, + port: int, + extra_env: dict[str, str], + ) -> str: + """Generate the run script for a profile-gateway s6 service. + + The script: + 1. Sources HERMES_HOME (and any extra env) via with-contenv — + so e.g. ``-e HERMES_HOME=/data/hermes`` is honored at run + time, not Python-substituted at registration time (OQ8-C). + 2. Activates the bundled venv. + 3. Drops to the hermes user and exec's + ``hermes -p gateway start --foreground --port ``. + """ + import shlex + lines = [ + "#!/command/with-contenv sh", + "# shellcheck shell=sh", + "set -e", + "cd /opt/data", + ". /opt/hermes/.venv/bin/activate", + ] + for k, v in sorted(extra_env.items()): + lines.append(f"export {k}={shlex.quote(v)}") + lines.append( + f"exec s6-setuidgid hermes hermes -p {shlex.quote(profile)} " + f"gateway start --foreground --port {port}" + ) + return "\n".join(lines) + "\n" + + @staticmethod + def _render_log_run(profile: str) -> str: + """Generate the log/run script for a profile-gateway service. + + OQ8-C: persist to ``${HERMES_HOME}/logs/gateways//``. + CRITICAL: the HERMES_HOME path is sourced from the runtime env + via with-contenv — NOT Python-substituted at registration time + — so a container started with ``-e HERMES_HOME=/data/hermes`` + gets its logs under /data/hermes/logs/..., not the build-time + default. + """ + import shlex + prof = shlex.quote(profile) + return ( + f"#!/command/with-contenv sh\n" + f"# shellcheck shell=sh\n" + f': "${{HERMES_HOME:=/opt/data}}"\n' + f'log_dir="$HERMES_HOME/logs/gateways/{prof}"\n' + f'mkdir -p "$log_dir"\n' + f'chown -R hermes:hermes "$log_dir" 2>/dev/null || true\n' + f'exec s6-setuidgid hermes s6-log n10 s1000000 T "$log_dir"\n' + ) + + # -- lifecycle --------------------------------------------------------- + + def start(self, name: str) -> None: + """Bring up a registered service (``s6-svc -u``).""" + import subprocess + subprocess.run( + [f"{_S6_BIN_DIR}/s6-svc", "-u", str(self.scandir / name)], + check=True, capture_output=True, timeout=5, + ) + + def stop(self, name: str) -> None: + """Bring down a registered service (``s6-svc -d``).""" + import subprocess + subprocess.run( + [f"{_S6_BIN_DIR}/s6-svc", "-d", str(self.scandir / name)], + check=True, capture_output=True, timeout=5, + ) + + def restart(self, name: str) -> None: + """Restart a registered service (``s6-svc -t`` = SIGTERM).""" + import subprocess + subprocess.run( + [f"{_S6_BIN_DIR}/s6-svc", "-t", str(self.scandir / name)], + check=True, capture_output=True, timeout=5, + ) + + def is_running(self, name: str) -> bool: + """True iff ``s6-svstat`` reports the service as up.""" + import subprocess + result = subprocess.run( + [f"{_S6_BIN_DIR}/s6-svstat", str(self.scandir / name)], + capture_output=True, text=True, timeout=5, + ) + return result.returncode == 0 and "up " in result.stdout + + # -- runtime registration --------------------------------------------- + + def supports_runtime_registration(self) -> bool: + return True + + def register_profile_gateway( + self, + profile: str, + *, + port: int, + extra_env: dict[str, str] | None = None, + ) -> None: + """Create the s6 service directory for a profile gateway. + + Triggers ``s6-svscanctl -a`` so s6-svscan picks the new directory + up immediately. The service is created in the *up* state — to + register without auto-starting, follow up with ``stop(profile)`` + (or pass the start flag via the future ``start_now=False`` arg, + which the Phase 4 reconciliation path uses via a ``down`` + marker file written directly). + + Raises: + ValueError: if the profile name is invalid or the service + directory already exists. + RuntimeError: if ``s6-svscanctl`` fails. + """ + import shutil + import subprocess + + svc_dir = self._service_dir(profile) + if svc_dir.exists(): + raise ValueError( + f"profile gateway {profile!r} already registered at {svc_dir}" + ) + + # Build the service directory atomically: write to a sibling + # temp dir, then rename. Avoids s6-svscan observing a half- + # populated directory on a fast rescan. + tmp_dir = svc_dir.with_name(svc_dir.name + ".tmp") + if tmp_dir.exists(): + shutil.rmtree(tmp_dir, ignore_errors=True) + tmp_dir.mkdir(parents=True) + + try: + (tmp_dir / "type").write_text("longrun\n") + + run_script = self._render_run_script(profile, port, extra_env or {}) + run_path = tmp_dir / "run" + run_path.write_text(run_script) + run_path.chmod(0o755) + + # Persistent log rotation (OQ8-C). + log_subdir = tmp_dir / "log" + log_subdir.mkdir() + log_run = log_subdir / "run" + log_run.write_text(self._render_log_run(profile)) + log_run.chmod(0o755) + + tmp_dir.rename(svc_dir) + except Exception: + shutil.rmtree(tmp_dir, ignore_errors=True) + raise + + # Trigger rescan so s6-svscan picks up the new service. + result = subprocess.run( + [f"{_S6_BIN_DIR}/s6-svscanctl", "-a", str(self.scandir)], + capture_output=True, text=True, timeout=5, + ) + if result.returncode != 0: + # Clean up: rescan failed, leave the directory in place would + # be confusing (no supervisor watching it). + shutil.rmtree(svc_dir, ignore_errors=True) + raise RuntimeError( + f"s6-svscanctl failed: {result.stderr or result.stdout}" + ) + + def unregister_profile_gateway(self, profile: str) -> None: + """Stop the profile gateway service and remove its directory. + + Idempotent: absent services are a no-op. Best-effort stop + + wait-for-down before removal so the running gateway process + gets a chance to shut down cleanly before its service dir + disappears. + """ + import shutil + import subprocess + + svc_dir = self._service_dir(profile) + if not svc_dir.exists(): + return + + # Stop the service (best effort — service may already be down). + subprocess.run( + [f"{_S6_BIN_DIR}/s6-svc", "-d", str(svc_dir)], + capture_output=True, text=True, timeout=5, + check=False, + ) + # Wait for it to actually go down (up to 10s). + subprocess.run( + [f"{_S6_BIN_DIR}/s6-svwait", "-D", "-t", "10000", str(svc_dir)], + capture_output=True, text=True, timeout=15, + check=False, + ) + + # Remove the directory. + shutil.rmtree(svc_dir, ignore_errors=True) + + # Rescan so s6-svscan drops its supervise process for the dir. + # -n = also reap orphan supervise processes. + subprocess.run( + [f"{_S6_BIN_DIR}/s6-svscanctl", "-an", str(self.scandir)], + capture_output=True, text=True, timeout=5, + check=False, + ) + + def list_profile_gateways(self) -> list[str]: + """Return the profile names of all currently-registered gateway services. + + Filters the scandir to entries that match the ``gateway-`` prefix. + Other services (e.g. ``s6-linux-init-shutdownd``) are ignored. + """ + if not self.scandir.exists(): + return [] + profiles: list[str] = [] + for entry in self.scandir.iterdir(): + if entry.name.startswith("."): + continue + if not entry.is_dir(): + continue + if not entry.name.startswith(S6_SERVICE_PREFIX): + continue + profiles.append(entry.name[len(S6_SERVICE_PREFIX):]) + return profiles diff --git a/tests/docker/test_s6_profile_gateway_integration.py b/tests/docker/test_s6_profile_gateway_integration.py new file mode 100644 index 00000000000..eb5cdca4bb8 --- /dev/null +++ b/tests/docker/test_s6_profile_gateway_integration.py @@ -0,0 +1,124 @@ +"""Harness: in-container integration tests for S6ServiceManager. + +The unit tests in tests/hermes_cli/test_service_manager.py exercise the +class against a tmp-path scandir with a stubbed ``subprocess.run``. +These tests run the real class inside a real container against the +real s6-svc / s6-svscanctl binaries, validating end-to-end. + +Phase 3 only registers the service slot — it doesn't depend on the +gateway actually starting (the binary will refuse to start without a +valid profile config). The full register → start → supervised-restart +→ unregister cycle is covered by Phase 4 once profile create/delete +hooks land. +""" +from __future__ import annotations + +import subprocess +import time + + +_REGISTER_SCRIPT = """ +import sys +sys.path.insert(0, "/opt/hermes") +from hermes_cli.service_manager import S6ServiceManager +S6ServiceManager().register_profile_gateway("phase3test", port=9301) +# Don't worry about whether the gateway actually starts — we only care +# that the supervision slot was created. The gateway run script will +# likely error out (no profile config exists) but that's expected. +print("REGISTERED") +""" + +_UNREGISTER_SCRIPT = """ +import sys +sys.path.insert(0, "/opt/hermes") +from hermes_cli.service_manager import S6ServiceManager +S6ServiceManager().unregister_profile_gateway("phase3test") +print("UNREGISTERED") +""" + + +def _exec(container: str, *args: str, timeout: int = 30) -> subprocess.CompletedProcess: + return subprocess.run( + ["docker", "exec", container, *args], + capture_output=True, text=True, timeout=timeout, + ) + + +def test_s6_register_creates_service_dir_in_live_container( + built_image: str, container_name: str, +) -> None: + """S6ServiceManager.register_profile_gateway must create + ``/run/service/gateway-/`` and trigger s6-svscan rescan + against the real s6 supervision tree.""" + subprocess.run( + ["docker", "run", "-d", "--name", container_name, built_image, + "sleep", "120"], + check=True, capture_output=True, timeout=30, + ) + # Give the supervision tree a moment to come up. + time.sleep(3) + + r = _exec(container_name, "python3", "-c", _REGISTER_SCRIPT, timeout=30) + assert "REGISTERED" in r.stdout, ( + f"register failed: stderr={r.stderr!r} stdout={r.stdout!r}" + ) + + # Service directory exists with the expected structure. + r = _exec(container_name, "test", "-d", "/run/service/gateway-phase3test") + assert r.returncode == 0, "service directory not created" + + r = _exec(container_name, "test", "-f", "/run/service/gateway-phase3test/run") + assert r.returncode == 0, "run script not created" + + r = _exec(container_name, "test", "-f", + "/run/service/gateway-phase3test/log/run") + assert r.returncode == 0, "log/run script not created" + + # s6-svscan picked it up — s6-svstat works against the dir. + # `docker exec` doesn't put /command/ on PATH (only the supervision + # tree does), so call s6-svstat by absolute path. + r = _exec(container_name, "/command/s6-svstat", + "/run/service/gateway-phase3test") + assert r.returncode == 0, f"s6-svstat failed: {r.stderr or r.stdout}" + + # list_profile_gateways picks it up. + r = _exec(container_name, "python3", "-c", ( + "from hermes_cli.service_manager import S6ServiceManager;" + "print(S6ServiceManager().list_profile_gateways())" + )) + assert "phase3test" in r.stdout, f"list output: {r.stdout!r}" + + +def test_s6_unregister_removes_service_dir_in_live_container( + built_image: str, container_name: str, +) -> None: + """unregister_profile_gateway must stop the service, remove the + directory, and trigger s6-svscan rescan so the supervise process + is dropped.""" + subprocess.run( + ["docker", "run", "-d", "--name", container_name, built_image, + "sleep", "120"], + check=True, capture_output=True, timeout=30, + ) + time.sleep(3) + + # First register so we have something to unregister. + r = _exec(container_name, "python3", "-c", _REGISTER_SCRIPT, timeout=30) + assert "REGISTERED" in r.stdout + + # Then unregister. + r = _exec(container_name, "python3", "-c", _UNREGISTER_SCRIPT, timeout=30) + assert "UNREGISTERED" in r.stdout, ( + f"unregister failed: stderr={r.stderr!r} stdout={r.stdout!r}" + ) + + # Directory is gone. + r = _exec(container_name, "test", "-d", "/run/service/gateway-phase3test") + assert r.returncode != 0, "service directory still exists after unregister" + + # list_profile_gateways no longer includes it. + r = _exec(container_name, "python3", "-c", ( + "from hermes_cli.service_manager import S6ServiceManager;" + "print(S6ServiceManager().list_profile_gateways())" + )) + assert "phase3test" not in r.stdout diff --git a/tests/hermes_cli/test_service_manager.py b/tests/hermes_cli/test_service_manager.py index 067048380b9..fc2ab6a7896 100644 --- a/tests/hermes_cli/test_service_manager.py +++ b/tests/hermes_cli/test_service_manager.py @@ -11,6 +11,7 @@ import pytest from hermes_cli.service_manager import ( LaunchdServiceManager, + S6ServiceManager, ServiceManager, ServiceManagerKind, SystemdServiceManager, @@ -260,14 +261,225 @@ def test_get_service_manager_raises_when_unsupported( get_service_manager() -def test_get_service_manager_raises_for_s6_until_phase_3( +def test_get_service_manager_returns_s6_instance( monkeypatch: pytest.MonkeyPatch, ) -> None: - """The s6 backend ships in Phase 3 — until then the factory raises - with an explicit message so accidental host code that ends up - running inside the container surfaces clearly.""" + """The s6 backend ships in Phase 3 — the factory must return an + S6ServiceManager when running inside a container.""" + from hermes_cli.service_manager import S6ServiceManager monkeypatch.setattr( "hermes_cli.service_manager.detect_service_manager", lambda: "s6", ) - with pytest.raises(RuntimeError, match="s6 backend not yet implemented"): - get_service_manager() + 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: + from hermes_cli.service_manager import S6ServiceManager + mgr = S6ServiceManager() + assert mgr.kind == "s6" + assert mgr.supports_runtime_registration() is True + + +def test_s6_register_creates_service_dir_and_triggers_scan( + s6_scandir, fake_subprocess_run, +) -> None: + from hermes_cli.service_manager import S6ServiceManager + mgr = S6ServiceManager(scandir=s6_scandir) + mgr.register_profile_gateway("coder", port=9150) + + 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 "hermes -p coder gateway start" in run_text + assert "--port 9150" in run_text + assert "s6-setuidgid hermes" 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" + ) + + # 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: + from hermes_cli.service_manager import S6ServiceManager + mgr = S6ServiceManager(scandir=s6_scandir) + mgr.register_profile_gateway( + "x", port=9300, 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_s6_register_rejects_invalid_profile_name(s6_scandir) -> None: + from hermes_cli.service_manager import S6ServiceManager + mgr = S6ServiceManager(scandir=s6_scandir) + with pytest.raises(ValueError): + mgr.register_profile_gateway("Bad/Name", port=9100) + + +def test_s6_register_rejects_duplicate(s6_scandir, fake_subprocess_run) -> None: + from hermes_cli.service_manager import S6ServiceManager + mgr = S6ServiceManager(scandir=s6_scandir) + (s6_scandir / "gateway-coder").mkdir(parents=True) + with pytest.raises(ValueError, match="already registered"): + mgr.register_profile_gateway("coder", port=9150) + + +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 + from hermes_cli.service_manager import S6ServiceManager + + 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", port=9150) + assert not (s6_scandir / "gateway-coder").exists() + + +def test_s6_unregister_removes_service_dir( + s6_scandir, fake_subprocess_run, +) -> None: + from hermes_cli.service_manager import S6ServiceManager + 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: + from hermes_cli.service_manager import S6ServiceManager + # 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: + from hermes_cli.service_manager import S6ServiceManager + # 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: + from hermes_cli.service_manager import S6ServiceManager + 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: + from hermes_cli.service_manager import S6ServiceManager + mgr = S6ServiceManager(scandir=s6_scandir) + 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"] + + +def test_s6_is_running_parses_svstat( + s6_scandir, monkeypatch: pytest.MonkeyPatch, +) -> None: + import subprocess as _sp + from hermes_cli.service_manager import S6ServiceManager + + 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