diff --git a/muxplex/cli.py b/muxplex/cli.py index 4aff5fa..b964479 100644 --- a/muxplex/cli.py +++ b/muxplex/cli.py @@ -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,17 +436,22 @@ def doctor() -> None: if sys.platform == "darwin": plist = Path.home() / "Library" / "LaunchAgents" / "com.muxplex.plist" if plist.exists(): - uid = os.getuid() - result = subprocess.run( - ["launchctl", "print", f"gui/{uid}/com.muxplex"], - capture_output=True, - text=True, - ) - if result.returncode == 0: - print(f" {ok_mark} Service: launchd agent running") + if _have_launchctl(): + uid = os.getuid() + result = subprocess.run( + ["launchctl", "print", f"gui/{uid}/com.muxplex"], + capture_output=True, + text=True, + ) + if result.returncode == 0: + print(f" {ok_mark} Service: launchd agent running") + else: + print( + f" {warn_mark} Service: launchd agent installed but not running ({plist})" + ) else: print( - f" {warn_mark} Service: launchd agent installed but not running ({plist})" + f" {warn_mark} Service: launchctl not found — cannot check status" ) else: print( @@ -515,12 +525,39 @@ def upgrade(*, force: bool = False) -> None: else: print(" Status: --force specified — skipping version check") - # 1. Detect platform and stop service + # 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" + + # 1. Stop service if sys.platform == "darwin": - label = "com.muxplex" - uid = os.getuid() - plist = Path.home() / "Library" / "LaunchAgents" / f"{label}.plist" - if plist.exists(): + # 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,116 +582,123 @@ 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") - if uv_path: - result = subprocess.run( - [ - uv_path, - "tool", - "install", - install_target, - "--force", - ], - capture_output=True, - text=True, - ) - if result.returncode != 0: - print(f" ERROR: uv tool install failed:\n{result.stderr}") - return - print(" Installed successfully") - else: - # Fallback: pip - pip_path = shutil.which("pip") or shutil.which("pip3") - if pip_path: - result = subprocess.run( - [ - pip_path, + try: + # Bug 3: dispatch — uv-tool-managed gets --reinstall; plain uv/pip otherwise + if uv_path: + if _is_uv_managed: + # uv-tool install: always reinstall the package by name + install_cmd = [ + uv_path, + "tool", "install", - "--upgrade", - install_target, - ], - capture_output=True, - text=True, - ) + "--reinstall", + "--force", + "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: pip install failed:\n{result.stderr}") - return - print(" Installed successfully") + print(f" ERROR: uv tool install failed:\n{result.stderr}") + _install_failed = True + else: + print(" Installed successfully") else: - print(" ERROR: neither uv nor pip found — cannot upgrade") - return + # 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], + capture_output=True, + text=True, + ) + if result.returncode != 0: + print(f" ERROR: pip install failed:\n{result.stderr}") + _install_failed = True + else: + print(" Installed successfully") + else: + print(" ERROR: neither uv nor pip found — cannot upgrade") + _install_failed = True - # 3. Regenerate service file (picks up any plist/unit changes) - if sys.platform == "darwin" or _have_systemctl(): - print(" Regenerating service file...") - from muxplex.service import service_install # noqa: PLC0415 + 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...") + from muxplex.service import service_install # noqa: PLC0415 - service_install() - else: - print(" ! systemctl not found — skipping service file regeneration") - - # 4. Restart service - if sys.platform == "darwin": - label = "com.muxplex" - uid = os.getuid() - plist = Path.home() / "Library" / "LaunchAgents" / f"{label}.plist" - if plist.exists(): - print(" Starting launchd service...") + service_install() + else: + print(" ! systemctl not found — skipping service file regeneration") + finally: + # 4. Restart service — ALWAYS runs after install, even on failure (best-effort). + if sys.platform == "darwin": + # Bug 2a: guard launchctl + if _have_launchctl(): + if plist.exists(): + print(" Starting launchd service...") + result = subprocess.run( + ["launchctl", "bootstrap", f"gui/{uid}", str(plist)], + capture_output=True, + text=True, + ) + if result.returncode == 0: + print(" Service started") + else: + # Fallback to legacy load for older macOS + subprocess.run( + ["launchctl", "load", str(plist)], capture_output=True + ) + print(" Service started (legacy)") + else: + print(" Service file not found — run: muxplex service install") + elif _have_systemctl(): result = subprocess.run( - ["launchctl", "bootstrap", f"gui/{uid}", str(plist)], + ["systemctl", "--user", "is-enabled", "muxplex"], capture_output=True, text=True, ) if result.returncode == 0: + print(" Restarting systemd service...") + subprocess.run( + ["systemctl", "--user", "daemon-reload"], capture_output=True + ) + subprocess.run( + ["systemctl", "--user", "start", "muxplex"], capture_output=True + ) print(" Service started") else: - # Fallback to legacy load for older macOS - subprocess.run(["launchctl", "load", str(plist)], capture_output=True) - print(" Service started (legacy)") + print(" Service not enabled — run: muxplex service install") else: - print(" Service file not found — run: muxplex service install") - elif _have_systemctl(): - result = subprocess.run( - ["systemctl", "--user", "is-enabled", "muxplex"], - capture_output=True, - text=True, - ) - if result.returncode == 0: - print(" Restarting systemd service...") - subprocess.run( - ["systemctl", "--user", "daemon-reload"], capture_output=True + # No service manager — ask the user to restart manually + pid_hint = "" + try: + pid_result = subprocess.run( + ["pgrep", "-f", "muxplex serve"], + capture_output=True, + text=True, + ) + pid_str = pid_result.stdout.strip() + if pid_str: + pid_hint = f" (running PID: {pid_str})" + except Exception: + pass + print( + " ! systemd not detected — restart muxplex manually to pick up the new version" + + pid_hint ) - subprocess.run( - ["systemctl", "--user", "start", "muxplex"], capture_output=True - ) - print(" Service started") - else: - print(" Service not enabled — run: muxplex service install") - else: - # No systemd available — ask the user to restart manually - pid_hint = "" - try: - pid_result = subprocess.run( - ["pgrep", "-f", "muxplex serve"], - capture_output=True, - text=True, - ) - pid_str = pid_result.stdout.strip() - if pid_str: - pid_hint = f" (running PID: {pid_str})" - except Exception: - pass + + # 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( - " ! systemd not detected — restart muxplex manually to pick up the new version" - + pid_hint + "\n ERROR: upgrade failed — muxplex service has been restarted (best-effort).\n" ) + sys.exit(1) # 5. Doctor check print("\n Verifying...") diff --git a/muxplex/tests/test_cli.py b/muxplex/tests/test_cli.py index 4061466..c11185b 100644 --- a/muxplex/tests/test_cli.py +++ b/muxplex/tests/test_cli.py @@ -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"