feat: add muxplex upgrade/update — stop service, reinstall, regenerate, restart
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'.
This commit is contained in:
Brian Krabach
+111
@@ -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(
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user