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

refactor: remove old install_service/launchd/systemd from cli.py, replaced by service module

Browse Source
This commit is contained in:
Brian Krabach
2026-03-31 17:24:34 -07:00
parent 496c37a32f
commit e1dfe8e10c
2 changed files with 73 additions and 356 deletions
Download Patch File Download Diff File Expand all files Collapse all files
muxplex/cli.py
+3 -154
View File
@@ -364,159 +364,6 @@ def _check_dependencies() -> None:
sys.exit(1)
def _install_launchd(executable: str) -> None:
"""Install a macOS launchd agent plist to ~/Library/LaunchAgents/."""
import shutil
# Prefer the entry point script ('muxplex') so macOS shows the correct
# process name in Activity Monitor and launchctl list.
muxplex_bin = shutil.which("muxplex")
if muxplex_bin:
program_args = f""" <array>
<string>{muxplex_bin}</string>
<string>--host</string>
<string>0.0.0.0</string>
</array>"""
else:
program_args = f""" <array>
<string>{executable}</string>
<string>-m</string>
<string>muxplex</string>
<string>--host</string>
<string>0.0.0.0</string>
</array>"""
# Build PATH that includes Homebrew and common tool locations.
# launchd agents run with a minimal PATH (/usr/bin:/bin:/usr/sbin:/sbin).
# Homebrew binaries (tmux, ttyd, uv, git) won't be found without this.
_homebrew_paths = [
"/opt/homebrew/bin", # Apple Silicon Homebrew
"/usr/local/bin", # Intel Homebrew + system extras
"/opt/homebrew/sbin",
"/usr/local/sbin",
]
_user_local = str(Path.home() / ".local" / "bin") # uv/pip tool installs
_extra = [_user_local] + _homebrew_paths
_base = "/usr/bin:/bin:/usr/sbin:/sbin"
_service_path = ":".join(_extra) + ":" + _base
label = "com.muxplex"
uid = os.getuid()
plist = f"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key>
<string>{label}</string>
<key>ProgramArguments</key>
{program_args}
<key>EnvironmentVariables</key>
<dict>
<key>PATH</key>
<string>{_service_path}</string>
</dict>
<key>RunAtLoad</key>
<true/>
<key>KeepAlive</key>
<true/>
<key>StandardOutPath</key>
<string>/tmp/muxplex.log</string>
<key>StandardErrorPath</key>
<string>/tmp/muxplex.err</string>
</dict>
</plist>
"""
path = Path.home() / "Library" / "LaunchAgents" / f"{label}.plist"
path.parent.mkdir(parents=True, exist_ok=True)
# Unload existing service first (ignore errors if not loaded)
subprocess.run(
["launchctl", "bootout", f"gui/{uid}/{label}"],
capture_output=True,
)
path.write_text(plist)
print(f"Launch agent written to {path}")
print()
# Load the service using the modern bootstrap API
result = subprocess.run(
["launchctl", "bootstrap", f"gui/{uid}", str(path)],
capture_output=True,
text=True,
)
if result.returncode == 0:
print(f" Service started (launchctl bootstrap gui/{uid})")
else:
# Fallback to legacy load for older macOS
result2 = subprocess.run(
["launchctl", "load", str(path)],
capture_output=True,
text=True,
)
if result2.returncode == 0:
print(" Service started (launchctl load)")
else:
print(" Could not auto-start. Try manually:")
print(f" launchctl bootstrap gui/{uid} {path}")
print()
print("Management commands:")
print(f" Stop: launchctl bootout gui/{uid}/{label}")
print(f" Start: launchctl bootstrap gui/{uid} {path}")
print(f" Status: launchctl print gui/{uid}/{label}")
print(" Logs: tail -f /tmp/muxplex.log")
def _install_systemd(executable: str, *, system: bool = False) -> None:
"""Install a Linux systemd service unit file."""
_raw_path = os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin")
_safe_path = ":".join(p for p in _raw_path.split(":") if not p.startswith("/mnt/"))
_safe_path = _safe_path or "/usr/local/bin:/usr/bin:/bin"
unit = f"""\
[Unit]
Description=muxplex — web-based tmux session dashboard
After=network.target
[Service]
Type=simple
ExecStart={executable} -m muxplex
Restart=on-failure
RestartSec=5s
Environment=PATH={_safe_path}
[Install]
WantedBy={"multi-user.target" if system else "default.target"}
"""
if system:
path = _system_service_path
reload_cmd = (
"sudo systemctl daemon-reload && sudo systemctl enable --now muxplex"
)
else:
path = Path.home() / ".config" / "systemd" / "user" / "muxplex.service"
path.parent.mkdir(parents=True, exist_ok=True)
reload_cmd = (
"systemctl --user daemon-reload && systemctl --user enable --now muxplex"
)
path.write_text(unit)
print(f"Service file written to {path}")
print(f"Enable with:\n {reload_cmd}")
def install_service(*, system: bool = False) -> None:
"""Install muxplex as a background service (launchd on macOS, systemd on Linux)."""
executable = sys.executable
if sys.platform == "darwin":
_install_launchd(executable)
else:
_install_systemd(executable, system=system)
def upgrade(*, force: bool = False) -> None:
"""Upgrade muxplex to the latest version and restart the service."""
print("\nmuxplex upgrade\n")
@@ -609,7 +456,9 @@ def upgrade(*, force: bool = False) -> None:
# 3. Regenerate service file (picks up any plist/unit changes)
print(" Regenerating service file...")
install_service(system=False)
from muxplex.service import service_install # noqa: PLC0415
service_install()
# 4. Restart service
if sys.platform == "darwin":
muxplex/tests/test_cli.py
+70 -202
View File
@@ -92,78 +92,6 @@ def test_main_install_service_system_flag():
mock_install.assert_called_once()
def test_install_service_user_mode_writes_unit_file(tmp_path, monkeypatch):
"""install_service(system=False) writes a unit file to ~/.config/systemd/user/."""
from muxplex.cli import install_service
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
install_service(system=False)
unit_path = fake_home / ".config" / "systemd" / "user" / "muxplex.service"
assert unit_path.exists()
content = unit_path.read_text()
assert "[Unit]" in content
assert "[Service]" in content
assert "[Install]" in content
assert "muxplex" in content
assert "default.target" in content
def test_install_service_system_mode_target(tmp_path, monkeypatch):
"""install_service(system=True) targets multi-user.target in the unit file."""
from muxplex.cli import install_service
# Redirect the system path to tmp so we don't write to /etc
unit_path = tmp_path / "muxplex.service"
monkeypatch.setattr("muxplex.cli._system_service_path", unit_path)
install_service(system=True)
assert unit_path.exists()
content = unit_path.read_text()
assert "multi-user.target" in content
def test_install_service_strips_wsl_mnt_paths_from_environment(tmp_path, monkeypatch):
"""Fix 3: install_service() must strip /mnt/ paths from Environment=PATH.
WSL mounts Windows at /mnt/c/, /mnt/d/ etc. Paths like
'/mnt/c/Program Files/dotnet/' contain spaces, causing systemd to
truncate and reject the Environment= line.
"""
from muxplex.cli import install_service
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
wsl_path = (
"/usr/local/bin:/usr/bin:/bin:/mnt/c/Program Files/dotnet:/mnt/d/tools/bin"
)
monkeypatch.setenv("PATH", wsl_path)
install_service(system=False)
unit_path = fake_home / ".config" / "systemd" / "user" / "muxplex.service"
content = unit_path.read_text()
# Find the Environment=PATH line
env_line = next(
(line for line in content.splitlines() if line.startswith("Environment=PATH=")),
None,
)
assert env_line is not None, "Environment=PATH line must be present"
assert "/mnt/" not in env_line, (
f"WSL /mnt/ paths must be stripped from Environment=PATH; got: {env_line!r}"
)
# Safe paths must still be present
assert "/usr/local/bin" in env_line
assert "/usr/bin" in env_line
def test_show_password_prints_password_from_file(tmp_path, monkeypatch, capsys):
"""show_password() prints the password when MUXPLEX_AUTH=password and file exists."""
from muxplex.cli import show_password
@@ -263,130 +191,6 @@ def test_reset_secret_prints_warning(tmp_path, monkeypatch, capsys):
)
def test_install_service_writes_launchd_plist_on_macos(tmp_path, monkeypatch):
"""install_service() on macOS writes a launchd plist to ~/Library/LaunchAgents/."""
import subprocess
from muxplex.cli import install_service
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
monkeypatch.setattr("sys.platform", "darwin")
monkeypatch.setattr(
subprocess,
"run",
lambda *a, **kw: type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})(),
)
install_service(system=False)
plist_path = fake_home / "Library" / "LaunchAgents" / "com.muxplex.plist"
assert plist_path.exists(), "Plist file must be created on macOS"
content = plist_path.read_text()
assert "com.muxplex" in content
assert "RunAtLoad" in content
assert "ProgramArguments" in content
assert "LaunchAgents" in str(plist_path)
def test_install_service_does_not_write_systemd_on_macos(tmp_path, monkeypatch):
"""install_service() on macOS must NOT write a systemd unit file."""
import subprocess
from muxplex.cli import install_service
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
monkeypatch.setattr("sys.platform", "darwin")
monkeypatch.setattr(
subprocess,
"run",
lambda *a, **kw: type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})(),
)
install_service(system=False)
systemd_path = fake_home / ".config" / "systemd" / "user" / "muxplex.service"
assert not systemd_path.exists(), "No systemd unit file should be written on macOS"
def test_install_service_uses_modern_launchctl_on_macos(tmp_path, monkeypatch, capsys):
"""install-service on macOS must use launchctl bootstrap (not load)."""
import subprocess
from muxplex.cli import install_service
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
monkeypatch.setattr("sys.platform", "darwin")
monkeypatch.setattr(shutil, "which", lambda name: f"/usr/local/bin/{name}")
calls = []
def mock_run(cmd, **kwargs):
calls.append(cmd)
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
monkeypatch.setattr(subprocess, "run", mock_run)
install_service(system=False)
launchctl_cmds = [c for c in calls if isinstance(c, list) and "launchctl" in str(c)]
assert any("bootout" in str(c) for c in launchctl_cmds), (
"must bootout old service before loading"
)
assert any("bootstrap" in str(c) for c in launchctl_cmds), (
"must use launchctl bootstrap (modern API)"
)
# Must NOT use deprecated load
assert not any(
c == ["launchctl", "load"]
or (isinstance(c, list) and c[:2] == ["launchctl", "load"])
for c in launchctl_cmds
), "must NOT use deprecated launchctl load"
def test_install_service_plist_includes_host_flag(tmp_path, monkeypatch):
"""The generated plist must include --host 0.0.0.0 for network access."""
import subprocess
from muxplex.cli import install_service
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
monkeypatch.setattr("sys.platform", "darwin")
monkeypatch.setattr(shutil, "which", lambda name: f"/usr/local/bin/{name}")
monkeypatch.setattr(
subprocess,
"run",
lambda *a, **kw: type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})(),
)
install_service(system=False)
plist_path = fake_home / "Library" / "LaunchAgents" / "com.muxplex.plist"
content = plist_path.read_text()
assert "0.0.0.0" in content, "plist must bind to 0.0.0.0 for network access"
def test_install_service_writes_systemd_on_linux(tmp_path, monkeypatch):
"""install_service() on Linux writes a systemd unit to ~/.config/systemd/user/."""
from muxplex.cli import install_service
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
monkeypatch.setattr("sys.platform", "linux")
install_service(system=False)
unit_path = fake_home / ".config" / "systemd" / "user" / "muxplex.service"
assert unit_path.exists(), "Systemd unit file must be created on Linux"
content = unit_path.read_text()
assert "[Unit]" in content
assert "[Service]" in content
def test_install_service_help_text_mentions_background_service():
"""install-service help must mention 'service', not just 'systemd'."""
import io
@@ -646,7 +450,6 @@ def test_upgrade_calls_uv_tool_install(monkeypatch, capsys):
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)
# Mock version check so upgrade proceeds regardless of local install type
monkeypatch.setattr(
@@ -655,7 +458,8 @@ def test_upgrade_calls_uv_tool_install(monkeypatch, capsys):
lambda info: (True, "update available (abc12345 → def67890)"),
)
cli_mod.upgrade()
with patch("muxplex.service.service_install", 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)]
@@ -729,7 +533,6 @@ def test_upgrade_force_skips_version_check(monkeypatch, capsys):
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)
# With force=True the version check must be bypassed entirely
check_calls = []
@@ -739,7 +542,8 @@ def test_upgrade_force_skips_version_check(monkeypatch, capsys):
lambda info: check_calls.append(info) or (True, "should not be reached"),
)
cli_mod.upgrade(force=True)
with patch("muxplex.service.service_install", lambda: None):
cli_mod.upgrade(force=True)
# _check_for_update must NOT have been called when force=True
assert len(check_calls) == 0, "Version check must be skipped when force=True"
@@ -762,7 +566,6 @@ def test_upgrade_already_up_to_date_skips_install(monkeypatch, capsys):
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)
monkeypatch.setattr(
cli_mod,
@@ -770,7 +573,8 @@ def test_upgrade_already_up_to_date_skips_install(monkeypatch, capsys):
lambda info: (False, "up to date (commit abcd1234)"),
)
cli_mod.upgrade()
with patch("muxplex.service.service_install", lambda: None):
cli_mod.upgrade()
out = capsys.readouterr().out
assert "up to date" in out.lower() or "already" in out.lower()
@@ -1131,3 +935,67 @@ def test_service_subcommand_in_help():
help_text = buf.getvalue().lower()
assert "service" in help_text
# ---------------------------------------------------------------------------
# task-6: Verify old install_service/launchd/systemd removed from cli.py
# ---------------------------------------------------------------------------
def test_old_install_service_removed_from_cli():
"""install_service must no longer exist in muxplex.cli (moved to muxplex.service)."""
import muxplex.cli as cli_mod
assert not hasattr(cli_mod, "install_service"), (
"install_service should be removed from cli.py; use muxplex.service.service_install"
)
def test_old_install_launchd_removed_from_cli():
"""_install_launchd must no longer exist in muxplex.cli (moved to muxplex.service)."""
import muxplex.cli as cli_mod
assert not hasattr(cli_mod, "_install_launchd"), (
"_install_launchd should be removed from cli.py; functionality is in muxplex.service"
)
def test_old_install_systemd_removed_from_cli():
"""_install_systemd must no longer exist in muxplex.cli (moved to muxplex.service)."""
import muxplex.cli as cli_mod
assert not hasattr(cli_mod, "_install_systemd"), (
"_install_systemd should be removed from cli.py; functionality is in muxplex.service"
)
def test_upgrade_uses_service_module_install(monkeypatch, capsys):
"""upgrade() must call muxplex.service.service_install instead of cli.install_service."""
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, "doctor", lambda: None)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available (abc12345 \u2192 def67890)"),
)
service_install_calls = []
with patch(
"muxplex.service.service_install", lambda: service_install_calls.append(True)
):
cli_mod.upgrade()
assert len(service_install_calls) > 0, (
"upgrade() must call muxplex.service.service_install() to regenerate the service file"
)