From 91faab9b9459ce5c8bd2eca27d70cb613eee4fa7 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Tue, 31 Mar 2026 18:12:35 -0700 Subject: [PATCH] fix: remove check=True from idempotent service ops; guard input() against EOFError MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit C1: _systemd_status, _systemd_stop, _systemd_uninstall (stop+disable), _launchd_status, _launchd_stop, _launchd_uninstall (bootout) no longer pass check=True. systemctl status exits 3 on stopped service; bootout exits non-zero on unloaded service — both are informational, not errors. C2: _prompt_host_if_localhost wraps input() in try/except (EOFError, KeyboardInterrupt) so service install doesn't crash in CI or piped stdin environments. Also fixes settings["host"] → settings.get("host") to prevent KeyError on partial/missing settings files. Added 9 regression tests in test_service.py: - test_systemd_status_no_check_true - test_systemd_stop_no_check_true - test_systemd_uninstall_stop_and_disable_no_check_true - test_launchd_status_no_check_true - test_launchd_stop_no_check_true - test_launchd_uninstall_no_check_true - test_prompt_host_eoferror_defaults_to_no_change - test_prompt_host_keyboard_interrupt_defaults_to_no_change - test_prompt_host_missing_host_key_no_keyerror Co-authored-by: Amplifier --- muxplex/service.py | 33 +++--- muxplex/tests/test_service.py | 209 ++++++++++++++++++++++++++++++++++ 2 files changed, 228 insertions(+), 14 deletions(-) diff --git a/muxplex/service.py b/muxplex/service.py index 18019f2..bbad5f3 100644 --- a/muxplex/service.py +++ b/muxplex/service.py @@ -94,11 +94,18 @@ def _prompt_host_if_localhost() -> None: from muxplex.settings import load_settings, patch_settings settings = load_settings() - if settings["host"] == "127.0.0.1": - answer = input( - "Host is 127.0.0.1 — change to 0.0.0.0 so the service is reachable? [Y/n] " - ) - if answer.strip().lower() in ("y", ""): + if settings.get("host") == "127.0.0.1": + try: + answer = ( + input( + "Host is 127.0.0.1 — change to 0.0.0.0 so the service is reachable? [Y/n] " + ) + .strip() + .lower() + ) + except (EOFError, KeyboardInterrupt): + answer = "n" + if answer in ("y", ""): patch_settings({"host": "0.0.0.0"}) @@ -122,8 +129,8 @@ def _systemd_install() -> None: def _systemd_uninstall() -> None: - subprocess.run(["systemctl", "--user", "stop", "muxplex"], check=True) - subprocess.run(["systemctl", "--user", "disable", "muxplex"], check=True) + subprocess.run(["systemctl", "--user", "stop", "muxplex"]) + subprocess.run(["systemctl", "--user", "disable", "muxplex"]) _SYSTEMD_UNIT_PATH.unlink(missing_ok=True) subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) @@ -133,7 +140,7 @@ def _systemd_start() -> None: def _systemd_stop() -> None: - subprocess.run(["systemctl", "--user", "stop", "muxplex"], check=True) + subprocess.run(["systemctl", "--user", "stop", "muxplex"]) def _systemd_restart() -> None: @@ -141,9 +148,7 @@ def _systemd_restart() -> None: def _systemd_status() -> None: - subprocess.run( - ["systemctl", "--user", "status", "muxplex", "--no-pager"], check=True - ) + subprocess.run(["systemctl", "--user", "status", "muxplex", "--no-pager"]) def _systemd_logs() -> None: @@ -173,7 +178,7 @@ def _launchd_install() -> None: def _launchd_uninstall() -> None: uid = os.getuid() - subprocess.run(["launchctl", "bootout", f"gui/{uid}/{_LAUNCHD_LABEL}"], check=True) + subprocess.run(["launchctl", "bootout", f"gui/{uid}/{_LAUNCHD_LABEL}"]) _LAUNCHD_PLIST_PATH.unlink(missing_ok=True) @@ -186,7 +191,7 @@ def _launchd_start() -> None: def _launchd_stop() -> None: uid = os.getuid() - subprocess.run(["launchctl", "bootout", f"gui/{uid}/{_LAUNCHD_LABEL}"], check=True) + subprocess.run(["launchctl", "bootout", f"gui/{uid}/{_LAUNCHD_LABEL}"]) def _launchd_restart() -> None: @@ -196,7 +201,7 @@ def _launchd_restart() -> None: def _launchd_status() -> None: uid = os.getuid() - subprocess.run(["launchctl", "print", f"gui/{uid}/{_LAUNCHD_LABEL}"], check=True) + subprocess.run(["launchctl", "print", f"gui/{uid}/{_LAUNCHD_LABEL}"]) def _launchd_logs() -> None: diff --git a/muxplex/tests/test_service.py b/muxplex/tests/test_service.py index e6cbdbe..d1c23bc 100644 --- a/muxplex/tests/test_service.py +++ b/muxplex/tests/test_service.py @@ -313,3 +313,212 @@ def test_launchd_status_runs_print_command(monkeypatch): assert "gui/501/com.muxplex" in " ".join(print_cmd), ( f"print must reference gui/501/com.muxplex, got: {print_cmd}" ) + + +# --------------------------------------------------------------------------- +# C1 regression tests — check=True must NOT be set on idempotent/informational +# operations: status, stop, uninstall (stop+disable for systemd, bootout for launchd) +# --------------------------------------------------------------------------- + + +def _make_kwargs_capture(): + """Return (calls_with_kw, monkeypatch_fn) for capturing subprocess.run kwargs.""" + calls_with_kw: list[tuple[list[str], dict]] = [] + + def fake_run(cmd, **kw): + calls_with_kw.append((list(cmd), dict(kw))) + + return calls_with_kw, fake_run + + +def test_systemd_status_no_check_true(monkeypatch): + """_systemd_status must NOT pass check=True — a stopped service yields exit code 3.""" + import subprocess + + import muxplex.service as svc + + calls_with_kw, fake_run = _make_kwargs_capture() + monkeypatch.setattr(subprocess, "run", fake_run) + svc._systemd_status() + + assert calls_with_kw, "subprocess.run was not called" + for cmd, kw in calls_with_kw: + assert kw.get("check") is not True, ( + f"check=True must not be set on status command {cmd}" + ) + + +def test_systemd_stop_no_check_true(monkeypatch): + """_systemd_stop must NOT pass check=True — stopping an already-stopped service is ok.""" + import subprocess + + import muxplex.service as svc + + calls_with_kw, fake_run = _make_kwargs_capture() + monkeypatch.setattr(subprocess, "run", fake_run) + svc._systemd_stop() + + assert calls_with_kw, "subprocess.run was not called" + for cmd, kw in calls_with_kw: + assert kw.get("check") is not True, ( + f"check=True must not be set on stop command {cmd}" + ) + + +def test_systemd_uninstall_stop_and_disable_no_check_true(monkeypatch, tmp_path): + """_systemd_uninstall's stop and disable calls must NOT pass check=True.""" + import subprocess + + import muxplex.service as svc + + unit_dir = tmp_path / "systemd" / "user" + unit_dir.mkdir(parents=True) + unit_path = unit_dir / "muxplex.service" + unit_path.write_text("[Unit]\n") + monkeypatch.setattr(svc, "_SYSTEMD_UNIT_DIR", unit_dir) + monkeypatch.setattr(svc, "_SYSTEMD_UNIT_PATH", unit_path) + + calls_with_kw, fake_run = _make_kwargs_capture() + monkeypatch.setattr(subprocess, "run", fake_run) + svc._systemd_uninstall() + + # stop and disable must not have check=True + for cmd, kw in calls_with_kw: + if "stop" in cmd or "disable" in cmd: + assert kw.get("check") is not True, ( + f"check=True must not be set on uninstall subcommand {cmd}" + ) + + +def test_launchd_status_no_check_true(monkeypatch): + """_launchd_status must NOT pass check=True — service may not be loaded.""" + import os + import subprocess + + import muxplex.service as svc + + monkeypatch.setattr(os, "getuid", lambda: 501) + calls_with_kw, fake_run = _make_kwargs_capture() + monkeypatch.setattr(subprocess, "run", fake_run) + svc._launchd_status() + + assert calls_with_kw, "subprocess.run was not called" + for cmd, kw in calls_with_kw: + assert kw.get("check") is not True, ( + f"check=True must not be set on launchd status command {cmd}" + ) + + +def test_launchd_stop_no_check_true(monkeypatch): + """_launchd_stop must NOT pass check=True — bootout on unloaded service is ok.""" + import os + import subprocess + + import muxplex.service as svc + + monkeypatch.setattr(os, "getuid", lambda: 501) + calls_with_kw, fake_run = _make_kwargs_capture() + monkeypatch.setattr(subprocess, "run", fake_run) + svc._launchd_stop() + + assert calls_with_kw, "subprocess.run was not called" + for cmd, kw in calls_with_kw: + assert kw.get("check") is not True, ( + f"check=True must not be set on launchd stop command {cmd}" + ) + + +def test_launchd_uninstall_no_check_true(monkeypatch, tmp_path): + """_launchd_uninstall's bootout must NOT pass check=True.""" + import os + import subprocess + + import muxplex.service as svc + + plist_dir = tmp_path / "LaunchAgents" + plist_dir.mkdir(parents=True) + plist_path = plist_dir / "com.muxplex.plist" + plist_path.write_text("") + monkeypatch.setattr(svc, "_LAUNCHD_PLIST_DIR", plist_dir) + monkeypatch.setattr(svc, "_LAUNCHD_PLIST_PATH", plist_path) + monkeypatch.setattr(os, "getuid", lambda: 501) + + calls_with_kw, fake_run = _make_kwargs_capture() + monkeypatch.setattr(subprocess, "run", fake_run) + svc._launchd_uninstall() + + for cmd, kw in calls_with_kw: + assert kw.get("check") is not True, ( + f"check=True must not be set on launchd uninstall command {cmd}" + ) + + +# --------------------------------------------------------------------------- +# C2 regression tests — _prompt_host_if_localhost must be resilient +# --------------------------------------------------------------------------- + + +def test_prompt_host_eoferror_defaults_to_no_change(monkeypatch): + """_prompt_host_if_localhost must not crash on EOFError (CI/piped stdin).""" + import muxplex.service as svc + + patched: list[dict] = [] + + def fake_load(): + return {"host": "127.0.0.1"} + + def fake_patch(settings): + patched.append(settings) + + def fake_input(_prompt): + raise EOFError + + monkeypatch.setattr("muxplex.settings.load_settings", fake_load) + monkeypatch.setattr("muxplex.settings.patch_settings", fake_patch) + monkeypatch.setattr("builtins.input", fake_input) + + # Must not raise, and must NOT patch settings (default to "n") + svc._prompt_host_if_localhost() + assert patched == [], "patch_settings must not be called when EOFError occurs" + + +def test_prompt_host_keyboard_interrupt_defaults_to_no_change(monkeypatch): + """_prompt_host_if_localhost must not crash on KeyboardInterrupt.""" + import muxplex.service as svc + + patched: list[dict] = [] + + def fake_load(): + return {"host": "127.0.0.1"} + + def fake_patch(settings): + patched.append(settings) + + def fake_input(_prompt): + raise KeyboardInterrupt + + monkeypatch.setattr("muxplex.settings.load_settings", fake_load) + monkeypatch.setattr("muxplex.settings.patch_settings", fake_patch) + monkeypatch.setattr("builtins.input", fake_input) + + svc._prompt_host_if_localhost() + assert patched == [], ( + "patch_settings must not be called when KeyboardInterrupt occurs" + ) + + +def test_prompt_host_missing_host_key_no_keyerror(monkeypatch): + """_prompt_host_if_localhost must not raise KeyError when 'host' key is absent.""" + import muxplex.service as svc + + def fake_load(): + return {} # no 'host' key + + def fake_patch(settings): + pass # should never be called + + monkeypatch.setattr("muxplex.settings.load_settings", fake_load) + monkeypatch.setattr("muxplex.settings.patch_settings", fake_patch) + + # Must not raise KeyError + svc._prompt_host_if_localhost()