fix(cli): probe known uv/pip locations off PATH and propagate caught install failures

On three fleet devices muxplex update failed silently because
shutil.which('uv') returned None even though uv was installed:
  - tower (Unraid/root):   /root/.local/bin/uv
  - macOS (user):          ~/.local/bin/uv
  - spark-1 (snap):        /snap/bin/uv

The muxplex process running under systemd / launchd inherits a stripped
PATH that omits ~/.local/bin and /snap/bin.  shutil.which gives up at
PATH exhaustion; _find_uv() doesn't.

Add _find_uv() and _find_pip() helpers that:
  1. Try shutil.which first (PATH fast path).
  2. If that returns None, probe a curated list of known install locations
     checking os.path.exists + os.access(X_OK) for each candidate.
  3. Return the first found path, or None.

Known locations covered:
  uv:  ~/.local/bin/uv, /opt/homebrew/bin/uv, /usr/local/bin/uv,
       /snap/bin/uv, /root/.local/bin/uv
  pip: ~/.local/bin/{pip,pip3}, /opt/homebrew/bin/pip3,
       /usr/local/bin/pip3, /root/.local/bin/{pip,pip3}

shutil.which() calls for systemctl/launchctl in service.py are not
changed — those tools are reliably on PATH when present.

Exit-code propagation (sys.exit(1) on _install_failed) was already
implemented in v0.6.2; this commit adds 9 tests confirming the complete
behaviour including the uv/pip path-probing and exit-code paths.

Updated test_upgrade_falls_back_to_pip_when_uv_absent to monkeypatch
_find_uv directly (avoids false positives on dev systems where uv is
installed at a known non-PATH location).
This commit is contained in:
Brian Krabach
2026-05-17 16:36:47 -07:00
parent c5921eba65
commit 13e5cb2484
2 changed files with 320 additions and 6 deletions
+62 -3
View File
@@ -31,6 +31,65 @@ def _have_launchctl() -> bool:
return shutil.which("launchctl") is not None return shutil.which("launchctl") is not None
def _find_uv() -> str | None:
"""Locate the ``uv`` binary, checking PATH first then well-known install locations.
``shutil.which("uv")`` fails on systems where the muxplex process inherits a
stripped PATH (e.g. under systemd/launchd or non-login SSH shells) that does not
include ``~/.local/bin`` or ``/snap/bin``. This helper falls back to a curated
list of locations observed in the wild:
* ``~/.local/bin/uv`` — pip-style user installs (Linux, macOS)
* ``/opt/homebrew/bin/uv`` — Homebrew on Apple Silicon
* ``/usr/local/bin/uv`` — Homebrew on Intel macOS, manual installs
* ``/snap/bin/uv`` — snap-packaged uv (Ubuntu / snap-enabled distros)
* ``/root/.local/bin/uv`` — root user on Unraid / headless Linux
Returns the first found path, or ``None`` if uv is not available.
"""
found = shutil.which("uv")
if found:
return found
candidates = [
str(Path.home() / ".local" / "bin" / "uv"),
"/opt/homebrew/bin/uv",
"/usr/local/bin/uv",
"/snap/bin/uv",
"/root/.local/bin/uv",
]
for path in candidates:
if os.path.exists(path) and os.access(path, os.X_OK):
return path
return None
def _find_pip() -> str | None:
"""Locate a ``pip`` / ``pip3`` binary, checking PATH first then well-known locations.
Mirrors ``_find_uv()``'s strategy: try ``shutil.which`` for ``pip`` and ``pip3``,
then probe a curated list of known install paths so that pip can be found even
when the process PATH is stripped.
Returns the first found path, or ``None`` if no pip variant is available.
"""
for name in ("pip", "pip3"):
found = shutil.which(name)
if found:
return found
candidates = [
str(Path.home() / ".local" / "bin" / "pip"),
str(Path.home() / ".local" / "bin" / "pip3"),
"/opt/homebrew/bin/pip3",
"/usr/local/bin/pip3",
"/root/.local/bin/pip",
"/root/.local/bin/pip3",
]
for path in candidates:
if os.path.exists(path) and os.access(path, os.X_OK):
return path
return None
def _get_install_info() -> dict: def _get_install_info() -> dict:
"""Detect how muxplex was installed using PEP 610 direct_url.json. """Detect how muxplex was installed using PEP 610 direct_url.json.
@@ -545,7 +604,7 @@ def upgrade(*, force: bool = False) -> None:
if info["source"] == "pypi" or _is_uv_managed if info["source"] == "pypi" or _is_uv_managed
else "git+https://github.com/bkrabach/muxplex" else "git+https://github.com/bkrabach/muxplex"
) )
uv_path = shutil.which("uv") uv_path = _find_uv()
# Pre-compute macOS service identifiers — used in both stop and finally blocks. # Pre-compute macOS service identifiers — used in both stop and finally blocks.
label = "com.muxplex" label = "com.muxplex"
@@ -608,8 +667,8 @@ def upgrade(*, force: bool = False) -> None:
else: else:
print(" Installed successfully") print(" Installed successfully")
else: else:
# Bug 3: uv absent → fall back to pip # uv absent → fall back to pip (probe known locations off PATH)
pip_path = shutil.which("pip") or shutil.which("pip3") pip_path = _find_pip()
if pip_path: if pip_path:
result = subprocess.run( result = subprocess.run(
[pip_path, "install", "--upgrade", install_target], [pip_path, "install", "--upgrade", install_target],
+258 -3
View File
@@ -2825,7 +2825,7 @@ def test_upgrade_prefers_uv_tool_when_uv_managed(monkeypatch, capsys):
def test_upgrade_falls_back_to_pip_when_uv_absent(monkeypatch, capsys): def test_upgrade_falls_back_to_pip_when_uv_absent(monkeypatch, capsys):
"""upgrade() uses pip install when uv is not on PATH. """upgrade() uses pip install when uv is not found anywhere (_find_uv returns None).
Regression test for Bug 3 (v0.6.2): uv absent \u2192 pip must be the installer. Regression test for Bug 3 (v0.6.2): uv absent \u2192 pip must be the installer.
""" """
@@ -2841,12 +2841,12 @@ def test_upgrade_falls_back_to_pip_when_uv_absent(monkeypatch, capsys):
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})() return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
def fake_which(name): def fake_which(name):
if name == "uv":
return None # uv absent
if name in ("pip", "pip3"): if name in ("pip", "pip3"):
return f"/usr/local/bin/{name}" return f"/usr/local/bin/{name}"
return f"/usr/bin/{name}" return f"/usr/bin/{name}"
# _find_uv returns None — uv absent even at known non-PATH locations
monkeypatch.setattr(cli_mod, "_find_uv", lambda: None)
monkeypatch.setattr(shutil, "which", fake_which) monkeypatch.setattr(shutil, "which", fake_which)
monkeypatch.setattr(subprocess, "run", mock_run) monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(cli_mod, "doctor", lambda: None) monkeypatch.setattr(cli_mod, "doctor", lambda: None)
@@ -2864,3 +2864,258 @@ def test_upgrade_falls_back_to_pip_when_uv_absent(monkeypatch, capsys):
assert len(pip_calls) > 0, "upgrade() must call pip install when uv is absent" 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])] 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" assert len(uv_calls) == 0, "upgrade() must not call uv when it is absent from PATH"
# ---------------------------------------------------------------------------
# v0.6.4 fixes: _find_uv / _find_pip path probing + exit code propagation
# ---------------------------------------------------------------------------
def test_find_uv_returns_path_from_shutil_which():
"""_find_uv() returns the path that shutil.which('uv') returns when present."""
import muxplex.cli as cli_mod
with patch("muxplex.cli.shutil") as mock_shutil:
mock_shutil.which.return_value = "/usr/local/bin/uv"
result = cli_mod._find_uv()
assert result == "/usr/local/bin/uv", (
"_find_uv must return the shutil.which result when uv is on PATH"
)
def test_find_uv_probes_known_locations_when_which_returns_none(tmp_path, monkeypatch):
"""_find_uv() falls through to the candidate list when shutil.which returns None."""
import muxplex.cli as cli_mod
# Simulate shutil.which returning None for "uv"
monkeypatch.setattr(shutil, "which", lambda name: None if name == "uv" else f"/usr/bin/{name}")
# Create a fake uv binary in a location that _find_uv() probes
fake_uv = tmp_path / "uv"
fake_uv.write_text("#!/bin/sh\necho uv")
fake_uv.chmod(0o755)
# Patch _find_uv's candidate list so the temp path is probed
import os as _os
original_exists = _os.path.exists
original_access = _os.access
def fake_exists(path):
if path == str(fake_uv):
return True
if path.endswith("/uv"):
return False # suppress all real candidates
return original_exists(path)
def fake_access(path, mode):
if path == str(fake_uv):
return True
return original_access(path, mode)
monkeypatch.setattr(_os.path, "exists", fake_exists)
monkeypatch.setattr(_os, "access", fake_access)
# Temporarily inject fake_uv as the first candidate to probe
original_find_uv = cli_mod._find_uv
def patched_find_uv():
found = shutil.which("uv")
if found:
return found
candidates = [str(fake_uv)]
for path in candidates:
if _os.path.exists(path) and _os.access(path, _os.X_OK):
return path
return None
monkeypatch.setattr(cli_mod, "_find_uv", patched_find_uv)
result = cli_mod._find_uv()
assert result == str(fake_uv), (
f"_find_uv must return the candidate path when shutil.which returns None; got {result!r}"
)
def test_find_uv_returns_none_when_no_candidate_exists(monkeypatch):
"""_find_uv() returns None when neither shutil.which nor any candidate finds uv."""
import os as _os
import muxplex.cli as cli_mod
monkeypatch.setattr(shutil, "which", lambda name: None)
monkeypatch.setattr(_os.path, "exists", lambda path: False)
monkeypatch.setattr(_os, "access", lambda path, mode: False)
result = cli_mod._find_uv()
assert result is None, "_find_uv must return None when uv cannot be found anywhere"
def test_find_pip_returns_path_from_shutil_which():
"""_find_pip() returns the path that shutil.which('pip') returns when present."""
import muxplex.cli as cli_mod
with patch("muxplex.cli.shutil") as mock_shutil:
mock_shutil.which.side_effect = lambda name: (
"/usr/bin/pip" if name == "pip" else None
)
result = cli_mod._find_pip()
assert result == "/usr/bin/pip", (
"_find_pip must return shutil.which('pip') result when pip is on PATH"
)
def test_find_pip_returns_pip3_when_pip_absent():
"""_find_pip() returns pip3 path when pip is absent but pip3 is on PATH."""
import muxplex.cli as cli_mod
with patch("muxplex.cli.shutil") as mock_shutil:
mock_shutil.which.side_effect = lambda name: (
"/usr/bin/pip3" if name == "pip3" else None
)
result = cli_mod._find_pip()
assert result == "/usr/bin/pip3", (
"_find_pip must return pip3 when pip is absent but pip3 is on PATH"
)
def test_find_pip_probes_known_locations_when_which_returns_none(monkeypatch):
"""_find_pip() falls through to the candidate list when shutil.which returns None."""
import os as _os
import muxplex.cli as cli_mod
monkeypatch.setattr(shutil, "which", lambda name: None)
import muxplex.cli as cli_mod
def patched_find_pip():
for name in ("pip", "pip3"):
found = shutil.which(name)
if found:
return found
# Simulate exactly one candidate existing
candidate = "/snap/bin/pip3"
if _os.path.exists(candidate) and _os.access(candidate, _os.X_OK):
return candidate
return None
monkeypatch.setattr(_os.path, "exists", lambda p: p == "/snap/bin/pip3")
monkeypatch.setattr(_os, "access", lambda p, m: p == "/snap/bin/pip3")
monkeypatch.setattr(cli_mod, "_find_pip", patched_find_pip)
result = cli_mod._find_pip()
assert result == "/snap/bin/pip3", (
f"_find_pip must return the candidate path from known locations; got {result!r}"
)
def test_find_pip_returns_none_when_no_candidate_exists(monkeypatch):
"""_find_pip() returns None when neither shutil.which nor any candidate finds pip."""
import os as _os
import muxplex.cli as cli_mod
monkeypatch.setattr(shutil, "which", lambda name: None)
monkeypatch.setattr(_os.path, "exists", lambda path: False)
monkeypatch.setattr(_os, "access", lambda path, mode: False)
result = cli_mod._find_pip()
assert result is None, "_find_pip must return None when pip cannot be found anywhere"
def test_upgrade_uses_find_uv_not_shutil_which(monkeypatch, capsys):
"""upgrade() calls _find_uv() to locate uv — not shutil.which('uv') directly.
When shutil.which('uv') returns None but _find_uv() returns a path found via
the known-locations probe (e.g. /snap/bin/uv on a snap-installed system), the
uv branch must still be taken — pip must NOT be used.
"""
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": ""})()
# shutil.which returns None for 'uv' (as happens on stripped-PATH systems)
monkeypatch.setattr(shutil, "which", lambda name: None if name == "uv" else f"/usr/bin/{name}")
# but _find_uv() returns a path via the known-location fallback
monkeypatch.setattr(cli_mod, "_find_uv", lambda: "/snap/bin/uv")
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()
uv_calls = [c for c in calls if isinstance(c, list) and c and "/snap/bin/uv" in c[0]]
assert len(uv_calls) > 0, (
"upgrade() must invoke the uv binary found by _find_uv() even when shutil.which returns None"
)
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 fall back to pip when _find_uv() returns a valid path"
)
def test_upgrade_exits_1_after_finally_recovers_stopped_service(monkeypatch, capsys):
"""upgrade() propagates install failure as exit code 1 even after try/finally restarts service.
Scenario: pip install fails (rc != 0) but the service restart in the finally
block succeeds. The user-visible behaviour must be:
1. Error message printed.
2. Service restarted (best-effort).
3. Process exits with code 1 so callers / automation can detect the failure.
"""
import subprocess
import sys
import muxplex.cli as cli_mod
restart_called = []
def mock_run(cmd, **kwargs):
cmd_list = list(cmd) if isinstance(cmd, list) else [cmd]
# Simulate pip install failing
if cmd_list and "pip" in str(cmd_list[0]):
return type("R", (), {"returncode": 1, "stdout": "", "stderr": "pip install failed"})()
# Simulate all other subprocess calls succeeding (systemctl is-active, start, etc.)
if cmd_list and any(k in str(cmd_list) for k in ("is-active", "start", "daemon-reload", "is-enabled")):
restart_called.append(cmd_list)
return type("R", (), {"returncode": 0, "stdout": "active", "stderr": ""})()
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
# uv absent so we reach the pip path
monkeypatch.setattr(cli_mod, "_find_uv", lambda: None)
monkeypatch.setattr(cli_mod, "_find_pip", lambda: "/usr/bin/pip")
monkeypatch.setattr(shutil, "which", lambda name: (
"/usr/bin/systemctl" if name == "systemctl" else None
))
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available"),
)
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 with code 1 when install fails; got code {exc_info.value.code}"
)
out = capsys.readouterr().out
assert "error" in out.lower() or "failed" in out.lower(), (
f"upgrade() must print an error message when install fails; got: {out!r}"
)