ken/muxplex
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(cli): propagate install failures, restart service on partial upgrade, prefer uv tool over pip

Browse Source 
Bug 1 — upgrade returned exit 0 on partial failure (macOS pip ImportError case):
  When the install subprocess fails (non-zero exit), upgrade() now sets
  _install_failed=True and calls sys.exit(1) after the finally block instead
  of silently returning.  A clear error message is printed so the caller /
  script can detect the failure.
  Files: muxplex/cli.py upgrade() install dispatch blocks.

Bug 2 — macOS launchd agent left unloaded when install fails:
  (a) Added _have_launchctl() helper mirroring the _have_systemctl() guard
      introduced in v0.6.1.  Every launchctl subprocess call in upgrade() and
      doctor() is now guarded — if launchctl is absent a clear note is printed
      and the step is skipped.
  (b) Wrapped stop + install + start in try/finally so the service-start step
      ALWAYS executes regardless of install outcome (success or failure).
      Applied to both the launchctl path (macOS) and the systemctl path (Linux/
      WSL) — same structural issue existed in both, only the macOS variant was
      observed in the field.
  Files: muxplex/cli.py _have_launchctl(), upgrade() try/finally.

Bug 3 — upgrade chose system pip3 over uv even when install was uv-tool-managed:
  Added uv-tool-managed detection: resolves the muxplex script on PATH and
  checks whether the target lives under ~/.local/share/uv/tools/.  When
  detected, uses 'uv tool install --reinstall --force muxplex' (package name,
  not a git URL) so the correct uv tool-environment is upgraded.  Falls back to
  pip when uv is absent from PATH — unchanged.
  Files: muxplex/cli.py upgrade() uv-tool detection + install dispatch.

Refs: v0.6.1 added _have_systemctl() and the Linux no-systemctl guard;
this commit brings parity for macOS and closes the remaining failure modes
observed during the fleet update.
This commit is contained in:
Brian Krabach
2026-05-17 14:45:44 -07:00
parent 831292279f
commit 78950cbf02
2 changed files with 532 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
muxplex/cli.py
+78 -34
View File
@@ -26,6 +26,11 @@ def _have_systemctl() -> bool:
return shutil.which("systemctl") is not None
def _have_launchctl() -> bool:
"""Return True if launchctl is on PATH (used to gate service-management steps)."""
return shutil.which("launchctl") is not None
def _get_install_info() -> dict:
"""Detect how muxplex was installed using PEP 610 direct_url.json.
@@ -431,6 +436,7 @@ def doctor() -> None:
if sys.platform == "darwin":
plist = Path.home() / "Library" / "LaunchAgents" / "com.muxplex.plist"
if plist.exists():
if _have_launchctl():
uid = os.getuid()
result = subprocess.run(
["launchctl", "print", f"gui/{uid}/com.muxplex"],
@@ -443,6 +449,10 @@ def doctor() -> None:
print(
f" {warn_mark} Service: launchd agent installed but not running ({plist})"
)
else:
print(
f" {warn_mark} Service: launchctl not found — cannot check status"
)
else:
print(
f" {warn_mark} Service: not installed (run: muxplex service install)"
@@ -515,12 +525,39 @@ def upgrade(*, force: bool = False) -> None:
else:
print(" Status: --force specified — skipping version check")
# 1. Detect platform and stop service
if sys.platform == "darwin":
# Bug 3: Detect whether this install is managed by uv tool.
# If the resolved muxplex binary lives inside the uv tools directory we
# always use `uv tool install --reinstall --force muxplex` regardless of
# the PEP-610 source tag so that the correct tool-environment is upgraded.
_uv_tools_dir = str(Path.home() / ".local" / "share" / "uv" / "tools")
_muxplex_script = shutil.which("muxplex")
_is_uv_managed = False
if _muxplex_script:
try:
if _uv_tools_dir in str(Path(_muxplex_script).resolve()):
_is_uv_managed = True
except Exception:
pass
# install_target: pypi / uv-tool-managed → package name; git → git URL
install_target = (
"muxplex"
if info["source"] == "pypi" or _is_uv_managed
else "git+https://github.com/bkrabach/muxplex"
)
uv_path = shutil.which("uv")
# Pre-compute macOS service identifiers — used in both stop and finally blocks.
label = "com.muxplex"
uid = os.getuid()
plist = Path.home() / "Library" / "LaunchAgents" / f"{label}.plist"
if plist.exists():
# 1. Stop service
if sys.platform == "darwin":
# Bug 2a: guard every launchctl call with _have_launchctl()
if not _have_launchctl():
print(" ! launchctl not found — skipping service management step")
elif plist.exists():
print(" Stopping launchd service...")
subprocess.run(
["launchctl", "bootout", f"gui/{uid}/{label}"], capture_output=True
@@ -545,52 +582,50 @@ def upgrade(*, force: bool = False) -> None:
else:
print(" No active systemd service found (skipping stop)")
# 2. Reinstall via uv tool install
# 2+4. Install (try) with guaranteed service restart in finally.
# Bug 1+2b: try/finally ensures the start step always runs — success OR failure.
_install_failed = False
print(" Installing latest version...")
install_target = (
"muxplex"
if info["source"] == "pypi"
else "git+https://github.com/bkrabach/muxplex"
)
uv_path = shutil.which("uv")
try:
# Bug 3: dispatch — uv-tool-managed gets --reinstall; plain uv/pip otherwise
if uv_path:
result = subprocess.run(
[
if _is_uv_managed:
# uv-tool install: always reinstall the package by name
install_cmd = [
uv_path,
"tool",
"install",
install_target,
"--reinstall",
"--force",
],
capture_output=True,
text=True,
)
"muxplex",
]
else:
install_cmd = [uv_path, "tool", "install", install_target, "--force"]
result = subprocess.run(install_cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f" ERROR: uv tool install failed:\n{result.stderr}")
return
_install_failed = True
else:
print(" Installed successfully")
else:
# Fallback: pip
# Bug 3: uv absent → fall back to pip
pip_path = shutil.which("pip") or shutil.which("pip3")
if pip_path:
result = subprocess.run(
[
pip_path,
"install",
"--upgrade",
install_target,
],
[pip_path, "install", "--upgrade", install_target],
capture_output=True,
text=True,
)
if result.returncode != 0:
print(f" ERROR: pip install failed:\n{result.stderr}")
return
_install_failed = True
else:
print(" Installed successfully")
else:
print(" ERROR: neither uv nor pip found — cannot upgrade")
return
_install_failed = True
if not _install_failed:
# 3. Regenerate service file (picks up any plist/unit changes)
if sys.platform == "darwin" or _have_systemctl():
print(" Regenerating service file...")
@@ -599,12 +634,11 @@ def upgrade(*, force: bool = False) -> None:
service_install()
else:
print(" ! systemctl not found — skipping service file regeneration")
# 4. Restart service
finally:
# 4. Restart service — ALWAYS runs after install, even on failure (best-effort).
if sys.platform == "darwin":
label = "com.muxplex"
uid = os.getuid()
plist = Path.home() / "Library" / "LaunchAgents" / f"{label}.plist"
# Bug 2a: guard launchctl
if _have_launchctl():
if plist.exists():
print(" Starting launchd service...")
result = subprocess.run(
@@ -616,7 +650,9 @@ def upgrade(*, force: bool = False) -> None:
print(" Service started")
else:
# Fallback to legacy load for older macOS
subprocess.run(["launchctl", "load", str(plist)], capture_output=True)
subprocess.run(
["launchctl", "load", str(plist)], capture_output=True
)
print(" Service started (legacy)")
else:
print(" Service file not found — run: muxplex service install")
@@ -638,7 +674,7 @@ def upgrade(*, force: bool = False) -> None:
else:
print(" Service not enabled — run: muxplex service install")
else:
# No systemd available — ask the user to restart manually
# No service manager — ask the user to restart manually
pid_hint = ""
try:
pid_result = subprocess.run(
@@ -656,6 +692,14 @@ def upgrade(*, force: bool = False) -> None:
+ pid_hint
)
# Bug 1: propagate install failure as a non-zero exit so callers / scripts
# can detect the partial upgrade. Service restart was attempted above (best-effort).
if _install_failed:
print(
"\n ERROR: upgrade failed — muxplex service has been restarted (best-effort).\n"
)
sys.exit(1)
# 5. Doctor check
print("\n Verifying...")
doctor()
muxplex/tests/test_cli.py
+381
View File
@@ -2483,3 +2483,384 @@ def test_doctor_no_systemctl_does_not_crash(monkeypatch, capsys):
# Must not raise — the original bug was FileNotFoundError on systems without systemctl
doctor()
# ---------------------------------------------------------------------------
# v0.6.2 fixes: install failure propagation, service-restart-on-failure,
# prefer uv over pip for uv-tool-managed installs
# ---------------------------------------------------------------------------
def test_have_launchctl_helper_exists():
"""_have_launchctl must be importable from muxplex.cli."""
from muxplex.cli import _have_launchctl # noqa: F401
def test_have_launchctl_returns_true_when_present(monkeypatch):
"""_have_launchctl() returns True when launchctl is on PATH."""
import shutil as _shutil
from muxplex.cli import _have_launchctl
monkeypatch.setattr(
_shutil,
"which",
lambda name: "/usr/bin/launchctl" if name == "launchctl" else None,
)
assert _have_launchctl() is True
def test_have_launchctl_returns_false_when_absent(monkeypatch):
"""_have_launchctl() returns False when launchctl is absent."""
import shutil as _shutil
from muxplex.cli import _have_launchctl
monkeypatch.setattr(_shutil, "which", lambda name: None)
assert _have_launchctl() is False
def test_upgrade_propagates_install_failure_as_exit1(monkeypatch, capsys):
"""upgrade() must sys.exit(1) when the install subprocess returns non-zero.
Regression test for Bug 1 (v0.6.2): previously upgrade() silently returned
exit 0 even when pip/uv failed mid-flight.
"""
import subprocess
import sys
import muxplex.cli as cli_mod
calls = []
def mock_run(cmd, **kwargs):
calls.append(cmd)
# Fail only on the install command (uv or pip)
if (
isinstance(cmd, list)
and cmd
and ("uv" in str(cmd[0]) or "pip" in str(cmd[0]))
):
return type(
"R", (), {"returncode": 1, "stdout": "", "stderr": "install error"}
)()
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(shutil, "which", lambda name: f"/usr/bin/{name}")
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available (v0.6.1 \u2192 v0.6.2)"),
)
monkeypatch.setattr(sys, "platform", "linux")
with pytest.raises(SystemExit) as exc_info:
cli_mod.upgrade()
assert exc_info.value.code == 1, (
f"upgrade() must exit(1) on install failure, got code {exc_info.value.code}"
)
out = capsys.readouterr().out
assert "error" in out.lower() or "fail" in out.lower(), (
f"upgrade() must print an error message on install failure; got: {out!r}"
)
def test_upgrade_restarts_systemctl_after_failed_install(monkeypatch, capsys):
"""upgrade() must attempt to restart the systemctl service even when install fails.
Regression test for Bug 2 (v0.6.2) — systemctl path: the start step must
run in the finally block regardless of install outcome.
"""
import subprocess
import sys
import muxplex.cli as cli_mod
calls: list = []
def mock_run(cmd, **kwargs):
calls.append(list(cmd) if isinstance(cmd, list) else cmd)
# Fail the install command
if isinstance(cmd, list) and cmd and "uv" in str(cmd[0]):
return type("R", (), {"returncode": 1, "stdout": "", "stderr": "uv fail"})()
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(shutil, "which", lambda name: f"/usr/bin/{name}")
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available"),
)
monkeypatch.setattr(sys, "platform", "linux")
with pytest.raises(SystemExit):
cli_mod.upgrade()
# systemctl start must still have been called after the failed install
start_calls = [
c for c in calls if isinstance(c, list) and "systemctl" in c and "start" in c
]
assert len(start_calls) > 0, (
"upgrade() must call systemctl start after a failed install (try/finally)"
)
# Verify call ordering: install attempt precedes start
install_idx = next(
(
i
for i, c in enumerate(calls)
if isinstance(c, list) and c and "uv" in str(c[0])
),
-1,
)
start_idx = next(
(
i
for i, c in enumerate(calls)
if isinstance(c, list) and "systemctl" in c and "start" in c
),
-1,
)
assert install_idx >= 0, "install command must have been attempted"
assert start_idx > install_idx, (
"systemctl start must be called AFTER the failed install attempt"
)
def test_upgrade_restarts_launchctl_after_failed_install(monkeypatch, capsys, tmp_path):
"""upgrade() must attempt to re-load the launchd agent even when install fails (macOS).
Regression test for Bug 2 (v0.6.2) — launchctl path: the bootstrap step
must run in the finally block regardless of install outcome.
"""
import subprocess
import sys
import muxplex.cli as cli_mod
# Set up a fake macOS home with the launchd plist present
fake_home = tmp_path / "home"
plist_dir = fake_home / "Library" / "LaunchAgents"
plist_dir.mkdir(parents=True)
(plist_dir / "com.muxplex.plist").write_text("")
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
monkeypatch.setattr(sys, "platform", "darwin")
monkeypatch.setattr(cli_mod, "_have_launchctl", lambda: True)
calls: list = []
def mock_run(cmd, **kwargs):
calls.append(list(cmd) if isinstance(cmd, list) else cmd)
if isinstance(cmd, list) and cmd and "uv" in str(cmd[0]):
return type("R", (), {"returncode": 1, "stdout": "", "stderr": "uv fail"})()
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(shutil, "which", lambda name: f"/usr/bin/{name}")
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available"),
)
with pytest.raises(SystemExit):
cli_mod.upgrade()
# launchctl bootstrap (or legacy load) must have been called after the failed install
restart_calls = [
c
for c in calls
if isinstance(c, list)
and "launchctl" in c
and ("bootstrap" in c or "load" in c)
]
assert len(restart_calls) > 0, (
"upgrade() must call launchctl bootstrap/load after a failed install (try/finally)"
)
# Verify ordering: install attempt before restart
install_idx = next(
(
i
for i, c in enumerate(calls)
if isinstance(c, list) and c and "uv" in str(c[0])
),
-1,
)
restart_idx = next(
(
i
for i, c in enumerate(calls)
if isinstance(c, list)
and "launchctl" in c
and ("bootstrap" in c or "load" in c)
),
-1,
)
assert install_idx >= 0, "install command must have been attempted"
assert restart_idx > install_idx, (
"launchctl bootstrap/load must be called AFTER the failed install attempt"
)
def test_upgrade_no_launchctl_on_linux_uses_systemctl(monkeypatch, capsys):
"""On Linux without launchctl, upgrade() uses systemctl and never calls launchctl.
Simulates a Linux host where launchctl is not installed. The upgrade must
complete normally (no SystemExit) using the systemctl path.
"""
import subprocess
import sys
import muxplex.cli as cli_mod
subprocess_calls: list = []
def mock_run(cmd, **kwargs):
subprocess_calls.append(list(cmd) if isinstance(cmd, list) else cmd)
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
def fake_which(name):
if name == "launchctl":
return None # launchctl absent on Linux
return f"/usr/bin/{name}"
monkeypatch.setattr(shutil, "which", fake_which)
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available"),
)
monkeypatch.setattr(sys, "platform", "linux")
with patch("muxplex.service.service_install", lambda: None):
cli_mod.upgrade() # must not raise
launchctl_calls = [
c for c in subprocess_calls if isinstance(c, list) and "launchctl" in c
]
assert len(launchctl_calls) == 0, (
f"upgrade() must not call launchctl on Linux; got: {launchctl_calls}"
)
systemctl_calls = [
c for c in subprocess_calls if isinstance(c, list) and "systemctl" in c
]
assert len(systemctl_calls) > 0, (
"upgrade() must call systemctl on Linux when available"
)
def test_upgrade_prefers_uv_tool_when_uv_managed(monkeypatch, capsys):
"""upgrade() calls 'uv tool install --reinstall --force muxplex' for uv-tool installs.
Regression test for Bug 3 (v0.6.2): when the running muxplex binary
resolves to inside the uv tools directory, the upgrade must use the uv
tool reinstall path (not pip).
"""
import subprocess
import sys
import muxplex.cli as cli_mod
# Build a fake muxplex path that looks like a uv-tool-managed install.
# Use Path.home() so _uv_tools_dir computation matches inside upgrade().
uv_tools_fake = str(Path.home() / ".local" / "share" / "uv" / "tools")
fake_muxplex_resolved = f"{uv_tools_fake}/muxplex/bin/muxplex"
fake_uv_path = str(Path.home() / ".local" / "bin" / "uv")
calls: list = []
def mock_run(cmd, **kwargs):
calls.append(list(cmd) if isinstance(cmd, list) else cmd)
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
def fake_which(name):
if name == "uv":
return fake_uv_path
if name == "muxplex":
# Return the already-resolved path under the uv tools dir so that
# Path(fake_muxplex_resolved).resolve() == fake_muxplex_resolved
return fake_muxplex_resolved
return f"/usr/bin/{name}"
monkeypatch.setattr(shutil, "which", fake_which)
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available (v0.6.1 \u2192 v0.6.2)"),
)
monkeypatch.setattr(sys, "platform", "linux")
with patch("muxplex.service.service_install", lambda: None):
cli_mod.upgrade()
uv_reinstall_calls = [
c
for c in calls
if isinstance(c, list) and "tool" in c and "install" in c and "--reinstall" in c
]
assert len(uv_reinstall_calls) > 0, (
"upgrade() must call 'uv tool install --reinstall' when install is uv-tool-managed"
)
install_cmd = uv_reinstall_calls[0]
assert "--reinstall" in install_cmd, "command must include --reinstall"
assert "--force" in install_cmd, "command must include --force"
assert "muxplex" in install_cmd, "command must target 'muxplex' package"
# Must not have fallen through to pip
pip_calls = [c for c in calls if isinstance(c, list) and c and "pip" in str(c[0])]
assert len(pip_calls) == 0, (
"upgrade() must not call pip when uv is available and install is uv-managed"
)
def test_upgrade_falls_back_to_pip_when_uv_absent(monkeypatch, capsys):
"""upgrade() uses pip install when uv is not on PATH.
Regression test for Bug 3 (v0.6.2): uv absent \u2192 pip must be the installer.
"""
import subprocess
import sys
import muxplex.cli as cli_mod
calls: list = []
def mock_run(cmd, **kwargs):
calls.append(list(cmd) if isinstance(cmd, list) else cmd)
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
def fake_which(name):
if name == "uv":
return None # uv absent
if name in ("pip", "pip3"):
return f"/usr/local/bin/{name}"
return f"/usr/bin/{name}"
monkeypatch.setattr(shutil, "which", fake_which)
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available"),
)
monkeypatch.setattr(sys, "platform", "linux")
with patch("muxplex.service.service_install", lambda: None):
cli_mod.upgrade()
pip_calls = [c for c in calls if isinstance(c, list) and c and "pip" in str(c[0])]
assert len(pip_calls) > 0, "upgrade() must call pip install when uv is absent"
uv_calls = [c for c in calls if isinstance(c, list) and c and "uv" in str(c[0])]
assert len(uv_calls) == 0, "upgrade() must not call uv when it is absent from PATH"