From 85798783b648d236db22473e36cda81858408292 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Mon, 30 Mar 2026 08:19:08 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20add=20muxplex=20upgrade/update=20?= =?UTF-8?q?=E2=80=94=20stop=20service,=20reinstall,=20regenerate,=20restar?= =?UTF-8?q?t?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Platform-aware: launchctl on macOS, systemd on Linux/WSL. Prefers uv tool install, falls back to pip. Regenerates service file (picks up any plist/unit changes). Runs doctor at the end for verification. 'update' is an alias for 'upgrade'. --- muxplex/cli.py | 111 ++++++++++++++++++++++++++++++++++++++ muxplex/tests/test_cli.py | 91 +++++++++++++++++++++++++++++++ 2 files changed, 202 insertions(+) diff --git a/muxplex/cli.py b/muxplex/cli.py index c7b8292..055093d 100644 --- a/muxplex/cli.py +++ b/muxplex/cli.py @@ -313,6 +313,110 @@ def install_service(*, system: bool = False) -> None: _install_systemd(executable, system=system) +def upgrade() -> None: + """Upgrade muxplex to the latest version and restart the service.""" + print("\nmuxplex upgrade\n") + + # 1. Detect platform and stop service + if sys.platform == "darwin": + plist = Path.home() / "Library" / "LaunchAgents" / "com.muxplex.plist" + if plist.exists(): + print(" Stopping launchd service...") + subprocess.run(["launchctl", "unload", str(plist)], capture_output=True) + else: + print(" No launchd service found (skipping stop)") + else: + # Linux/WSL — check systemd + result = subprocess.run( + ["systemctl", "--user", "is-active", "muxplex"], + capture_output=True, + text=True, + ) + if result.returncode == 0: + print(" Stopping systemd service...") + subprocess.run( + ["systemctl", "--user", "stop", "muxplex"], capture_output=True + ) + else: + print(" No active systemd service found (skipping stop)") + + # 2. Reinstall via uv tool install + print(" Installing latest version...") + uv_path = shutil.which("uv") + if uv_path: + result = subprocess.run( + [ + uv_path, + "tool", + "install", + "git+https://github.com/bkrabach/muxplex", + "--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, + "install", + "--upgrade", + "git+https://github.com/bkrabach/muxplex", + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + print(f" ERROR: pip install failed:\n{result.stderr}") + return + print(" Installed successfully") + else: + print(" ERROR: neither uv nor pip found — cannot upgrade") + return + + # 3. Regenerate service file (picks up any plist/unit changes) + print(" Regenerating service file...") + install_service(system=False) + + # 4. Restart service + if sys.platform == "darwin": + plist = Path.home() / "Library" / "LaunchAgents" / "com.muxplex.plist" + if plist.exists(): + print(" Starting launchd service...") + subprocess.run(["launchctl", "load", str(plist)], capture_output=True) + print(" Service started") + else: + print(" Service file not found — run: muxplex install-service") + else: + 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 + ) + subprocess.run( + ["systemctl", "--user", "start", "muxplex"], capture_output=True + ) + print(" Service started") + else: + print(" Service not enabled — run: muxplex install-service") + + # 5. Doctor check + print("\n Verifying...") + doctor() + + def main() -> None: """CLI entry point.""" parser = argparse.ArgumentParser( @@ -356,6 +460,11 @@ def main() -> None: sub.add_parser("doctor", help="Check dependencies and system status") + sub.add_parser( + "upgrade", help="Upgrade muxplex to latest version and restart service" + ) + sub.add_parser("update", help="Alias for upgrade") + args = parser.parse_args() if args.command == "install-service": @@ -366,6 +475,8 @@ def main() -> None: reset_secret() elif args.command == "doctor": doctor() + elif args.command in ("upgrade", "update"): + upgrade() else: _check_dependencies() serve( diff --git a/muxplex/tests/test_cli.py b/muxplex/tests/test_cli.py index 8b620f3..ea81056 100644 --- a/muxplex/tests/test_cli.py +++ b/muxplex/tests/test_cli.py @@ -516,3 +516,94 @@ def test_main_dispatches_to_doctor(monkeypatch): assert len(calls) == 1, ( "doctor() must be called once when 'doctor' subcommand is used" ) + + +# --------------------------------------------------------------------------- +# upgrade / update subcommand tests +# --------------------------------------------------------------------------- + + +def test_upgrade_subcommand_registered(): + """upgrade must be a valid subcommand.""" + import io + + from muxplex.cli import main + + buf = io.StringIO() + with patch("sys.argv", ["muxplex", "--help"]): + try: + with patch("sys.stdout", buf): + main() + except SystemExit: + pass + + help_text = buf.getvalue().lower() + assert "upgrade" in help_text + + +def test_update_alias_registered(): + """update must be a valid subcommand (alias for upgrade).""" + import io + + from muxplex.cli import main + + buf = io.StringIO() + with patch("sys.argv", ["muxplex", "--help"]): + try: + with patch("sys.stdout", buf): + main() + except SystemExit: + pass + + help_text = buf.getvalue().lower() + assert "update" in help_text + + +def test_upgrade_calls_uv_tool_install(monkeypatch, capsys): + """upgrade must attempt uv tool install.""" + import subprocess + + import muxplex.cli as cli_mod + + calls = [] + + def mock_run(cmd, **kwargs): + calls.append(cmd) + 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, "install_service", lambda system=False: None) + monkeypatch.setattr(cli_mod, "doctor", lambda: None) + + cli_mod.upgrade() + + # Should have called uv tool install + uv_calls = [c for c in calls if isinstance(c, list) and "uv" in str(c)] + assert len(uv_calls) > 0, "upgrade must call uv tool install" + + +def test_main_dispatches_to_upgrade(monkeypatch): + """main() with 'upgrade' subcommand must invoke upgrade().""" + from muxplex.cli import main + + calls = [] + monkeypatch.setattr("muxplex.cli.upgrade", lambda: calls.append(True)) + + with patch("sys.argv", ["muxplex", "upgrade"]): + main() + + assert len(calls) == 1, "upgrade() must be called once for 'upgrade' subcommand" + + +def test_main_dispatches_update_to_upgrade(monkeypatch): + """main() with 'update' subcommand must also invoke upgrade().""" + from muxplex.cli import main + + calls = [] + monkeypatch.setattr("muxplex.cli.upgrade", lambda: calls.append(True)) + + with patch("sys.argv", ["muxplex", "update"]): + main() + + assert len(calls) == 1, "upgrade() must be called once for 'update' subcommand"