Brian Krabach
320c9f3ba7
Replace deprecated launchctl load/unload with bootstrap/bootout (modern macOS API). Auto-bootout existing service before reinstall (fixes 'I/O error' on reload). Add --host 0.0.0.0 to plist ProgramArguments so the service is actually accessible from the network. install-service now auto-starts the service and prints management commands. upgrade() also uses modern API. doctor() checks if service is actually running.
795 lines
25 KiB
Python
795 lines
25 KiB
Python
"""Tests for muxplex/cli.py — CLI entry point."""
|
|
|
|
import shutil
|
|
import stat
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
def test_cli_module_importable():
|
|
"""muxplex.cli must be importable."""
|
|
from muxplex.cli import main # noqa: F401
|
|
|
|
|
|
def test_main_calls_serve_by_default():
|
|
"""Calling main() with no args must invoke serve() with new defaults."""
|
|
from muxplex.cli import main
|
|
|
|
with patch("muxplex.cli.serve") as mock_serve:
|
|
with patch("sys.argv", ["muxplex"]):
|
|
main()
|
|
mock_serve.assert_called_once_with(
|
|
host="127.0.0.1", port=8088, auth="pam", session_ttl=604800
|
|
)
|
|
|
|
|
|
def test_main_passes_custom_host_and_port():
|
|
"""main() with --host/--port must forward them to serve()."""
|
|
from muxplex.cli import main
|
|
|
|
with patch("muxplex.cli.serve") as mock_serve:
|
|
with patch("sys.argv", ["muxplex", "--host", "192.168.1.1", "--port", "9000"]):
|
|
main()
|
|
mock_serve.assert_called_once_with(
|
|
host="192.168.1.1", port=9000, auth="pam", session_ttl=604800
|
|
)
|
|
|
|
|
|
def test_main_default_host_is_localhost():
|
|
"""Default --host must be 127.0.0.1 (not 0.0.0.0)."""
|
|
from muxplex.cli import main
|
|
|
|
with patch("muxplex.cli.serve") as mock_serve:
|
|
with patch("sys.argv", ["muxplex"]):
|
|
main()
|
|
_, kwargs = mock_serve.call_args
|
|
assert kwargs["host"] == "127.0.0.1"
|
|
|
|
|
|
def test_main_passes_auth_flag():
|
|
"""main() with --auth password must forward auth='password' to serve()."""
|
|
from muxplex.cli import main
|
|
|
|
with patch("muxplex.cli.serve") as mock_serve:
|
|
with patch("sys.argv", ["muxplex", "--auth", "password"]):
|
|
main()
|
|
mock_serve.assert_called_once_with(
|
|
host="127.0.0.1", port=8088, auth="password", session_ttl=604800
|
|
)
|
|
|
|
|
|
def test_main_passes_session_ttl_flag():
|
|
"""main() with --session-ttl 3600 must forward session_ttl=3600 to serve()."""
|
|
from muxplex.cli import main
|
|
|
|
with patch("muxplex.cli.serve") as mock_serve:
|
|
with patch("sys.argv", ["muxplex", "--session-ttl", "3600"]):
|
|
main()
|
|
mock_serve.assert_called_once_with(
|
|
host="127.0.0.1", port=8088, auth="pam", session_ttl=3600
|
|
)
|
|
|
|
|
|
def test_main_install_service_subcommand():
|
|
"""main() with 'install-service' must invoke install_service()."""
|
|
from muxplex.cli import main
|
|
|
|
with patch("muxplex.cli.install_service") as mock_install:
|
|
with patch("sys.argv", ["muxplex", "install-service"]):
|
|
main()
|
|
mock_install.assert_called_once_with(system=False)
|
|
|
|
|
|
def test_main_install_service_system_flag():
|
|
"""main() with 'install-service --system' passes system=True."""
|
|
from muxplex.cli import main
|
|
|
|
with patch("muxplex.cli.install_service") as mock_install:
|
|
with patch("sys.argv", ["muxplex", "install-service", "--system"]):
|
|
main()
|
|
mock_install.assert_called_once_with(system=True)
|
|
|
|
|
|
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
|
|
|
|
# Set up fake home with password file
|
|
fake_home = tmp_path / "home"
|
|
pw_dir = fake_home / ".config" / "muxplex"
|
|
pw_dir.mkdir(parents=True)
|
|
pw_file = pw_dir / "password"
|
|
pw_file.write_text("my-test-password\n")
|
|
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
|
|
monkeypatch.setenv("MUXPLEX_AUTH", "password")
|
|
|
|
show_password()
|
|
|
|
captured = capsys.readouterr()
|
|
assert "my-test-password" in captured.out
|
|
|
|
|
|
def test_show_password_no_file(tmp_path, monkeypatch, capsys):
|
|
"""show_password() tells user no file found when in password mode with no file."""
|
|
from muxplex.cli import show_password
|
|
|
|
# Set up fake home WITHOUT password file
|
|
fake_home = tmp_path / "home"
|
|
fake_home.mkdir(parents=True)
|
|
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
|
|
monkeypatch.setenv("MUXPLEX_AUTH", "password")
|
|
|
|
show_password()
|
|
|
|
captured = capsys.readouterr()
|
|
output_lower = captured.out.lower()
|
|
assert "no password" in output_lower or "not found" in output_lower
|
|
|
|
|
|
def test_show_password_pam_mode(monkeypatch, capsys):
|
|
"""show_password() reports PAM mode when pam_available() is True and not password mode."""
|
|
from muxplex.cli import show_password
|
|
|
|
monkeypatch.delenv("MUXPLEX_AUTH", raising=False)
|
|
|
|
with patch("muxplex.cli.pam_available", return_value=True):
|
|
show_password()
|
|
|
|
captured = capsys.readouterr()
|
|
assert "pam" in captured.out.lower()
|
|
|
|
|
|
def test_reset_secret_writes_new_secret(tmp_path, monkeypatch):
|
|
"""reset_secret() writes a new secret file with content longer than 20 chars."""
|
|
from muxplex.cli import reset_secret
|
|
|
|
fake_home = tmp_path / "home"
|
|
fake_home.mkdir()
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
|
|
|
|
reset_secret()
|
|
|
|
secret_path = fake_home / ".config" / "muxplex" / "secret"
|
|
assert secret_path.exists(), "Secret file must be created"
|
|
content = secret_path.read_text().strip()
|
|
assert len(content) > 20, f"Secret must be longer than 20 chars, got {len(content)}"
|
|
|
|
|
|
def test_reset_secret_sets_0600_permissions(tmp_path, monkeypatch):
|
|
"""reset_secret() sets file permissions to 0o600."""
|
|
from muxplex.cli import reset_secret
|
|
|
|
fake_home = tmp_path / "home"
|
|
fake_home.mkdir()
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
|
|
|
|
reset_secret()
|
|
|
|
secret_path = fake_home / ".config" / "muxplex" / "secret"
|
|
file_mode = stat.S_IMODE(secret_path.stat().st_mode)
|
|
assert file_mode == 0o600, f"Expected 0o600, got {oct(file_mode)}"
|
|
|
|
|
|
def test_reset_secret_prints_warning(tmp_path, monkeypatch, capsys):
|
|
"""reset_secret() prints a warning that sessions are now invalid."""
|
|
from muxplex.cli import reset_secret
|
|
|
|
fake_home = tmp_path / "home"
|
|
fake_home.mkdir()
|
|
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
|
|
|
|
reset_secret()
|
|
|
|
captured = capsys.readouterr()
|
|
output_lower = captured.out.lower()
|
|
assert "invalid" in output_lower or "warning" in output_lower, (
|
|
f"Expected 'invalid' or 'warning' in output, got: {captured.out!r}"
|
|
)
|
|
|
|
|
|
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
|
|
from muxplex.cli import main
|
|
|
|
buf = io.StringIO()
|
|
with patch("sys.argv", ["muxplex", "install-service", "--help"]):
|
|
try:
|
|
with patch("sys.stdout", buf):
|
|
main()
|
|
except SystemExit:
|
|
pass
|
|
|
|
help_text = buf.getvalue().lower()
|
|
assert "service" in help_text
|
|
|
|
|
|
def test_check_dependencies_exits_when_ttyd_missing(monkeypatch):
|
|
"""_check_dependencies() must sys.exit(1) when ttyd is not in PATH."""
|
|
import shutil
|
|
import pytest
|
|
from muxplex.cli import _check_dependencies
|
|
|
|
orig_which = shutil.which
|
|
|
|
def fake_which(name):
|
|
if name == "ttyd":
|
|
return None
|
|
return orig_which(name)
|
|
|
|
monkeypatch.setattr(shutil, "which", fake_which)
|
|
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
_check_dependencies()
|
|
assert exc_info.value.code == 1
|
|
|
|
|
|
def test_check_dependencies_exits_when_tmux_missing(monkeypatch):
|
|
"""_check_dependencies() must sys.exit(1) when tmux is not in PATH."""
|
|
import shutil
|
|
import pytest
|
|
from muxplex.cli import _check_dependencies
|
|
|
|
orig_which = shutil.which
|
|
|
|
def fake_which(name):
|
|
if name == "tmux":
|
|
return None
|
|
return orig_which(name)
|
|
|
|
monkeypatch.setattr(shutil, "which", fake_which)
|
|
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
_check_dependencies()
|
|
assert exc_info.value.code == 1
|
|
|
|
|
|
def test_check_dependencies_passes_when_all_present(monkeypatch):
|
|
"""_check_dependencies() must not raise when both tmux and ttyd are found."""
|
|
import shutil
|
|
from muxplex.cli import _check_dependencies
|
|
|
|
monkeypatch.setattr(shutil, "which", lambda name: f"/usr/bin/{name}")
|
|
|
|
# Should not raise
|
|
_check_dependencies()
|
|
|
|
|
|
def test_main_check_dependencies_called_for_serve(monkeypatch):
|
|
"""main() must call _check_dependencies() when subcommand is serve."""
|
|
from muxplex.cli import main
|
|
|
|
calls = []
|
|
monkeypatch.setattr("muxplex.cli._check_dependencies", lambda: calls.append(True))
|
|
|
|
with patch("muxplex.cli.serve"):
|
|
with patch("sys.argv", ["muxplex"]):
|
|
main()
|
|
|
|
assert len(calls) == 1, "_check_dependencies must be called once for serve"
|
|
|
|
|
|
def test_main_check_dependencies_not_called_for_install_service(monkeypatch):
|
|
"""main() must NOT call _check_dependencies() for install-service subcommand."""
|
|
from muxplex.cli import main
|
|
|
|
calls = []
|
|
monkeypatch.setattr("muxplex.cli._check_dependencies", lambda: calls.append(True))
|
|
|
|
with patch("muxplex.cli.install_service"):
|
|
with patch("sys.argv", ["muxplex", "install-service"]):
|
|
main()
|
|
|
|
assert len(calls) == 0, "_check_dependencies must NOT be called for install-service"
|
|
|
|
|
|
def test_dunder_main_calls_main():
|
|
"""python -m muxplex must call cli.main()."""
|
|
import importlib.util
|
|
|
|
# Locate __main__.py without executing it (find_spec does not import)
|
|
spec = importlib.util.find_spec("muxplex.__main__")
|
|
assert spec is not None and spec.origin is not None
|
|
|
|
with patch("muxplex.cli.main") as mock_main:
|
|
exec(Path(spec.origin).read_text()) # noqa: S102
|
|
mock_main.assert_called_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# doctor() tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_doctor_shows_python_version(capsys):
|
|
"""doctor must show Python version."""
|
|
from muxplex.cli import doctor
|
|
|
|
doctor()
|
|
out = capsys.readouterr().out
|
|
assert "Python" in out
|
|
|
|
|
|
def test_doctor_checks_tmux(capsys, monkeypatch):
|
|
"""doctor must check for tmux."""
|
|
import subprocess
|
|
|
|
from muxplex.cli import doctor
|
|
|
|
monkeypatch.setattr(
|
|
"shutil.which", lambda name: "/usr/bin/tmux" if name == "tmux" else None
|
|
)
|
|
monkeypatch.setattr(
|
|
subprocess,
|
|
"run",
|
|
lambda *a, **kw: type(
|
|
"R", (), {"returncode": 0, "stdout": "tmux 3.4", "stderr": ""}
|
|
)(),
|
|
)
|
|
doctor()
|
|
out = capsys.readouterr().out
|
|
assert "tmux" in out
|
|
|
|
|
|
def test_doctor_reports_missing_ttyd(capsys, monkeypatch):
|
|
"""doctor must report when ttyd is missing."""
|
|
from muxplex.cli import doctor
|
|
|
|
original_which = shutil.which
|
|
|
|
def mock_which(name):
|
|
if name == "ttyd":
|
|
return None
|
|
return original_which(name)
|
|
|
|
monkeypatch.setattr("shutil.which", mock_which)
|
|
doctor()
|
|
out = capsys.readouterr().out
|
|
assert "ttyd" in out
|
|
assert "not found" in out
|
|
|
|
|
|
def test_doctor_shows_platform(capsys):
|
|
"""doctor must show platform info."""
|
|
from muxplex.cli import doctor
|
|
|
|
doctor()
|
|
out = capsys.readouterr().out
|
|
assert "Platform" in out
|
|
|
|
|
|
def test_doctor_subcommand_registered():
|
|
"""doctor must be a valid subcommand in main() argparse."""
|
|
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 "doctor" in help_text
|
|
|
|
|
|
def test_main_dispatches_to_doctor(monkeypatch):
|
|
"""main() with 'doctor' subcommand must invoke doctor()."""
|
|
from muxplex.cli import main
|
|
|
|
calls = []
|
|
monkeypatch.setattr("muxplex.cli.doctor", lambda: calls.append(True))
|
|
|
|
with patch("sys.argv", ["muxplex", "doctor"]):
|
|
main()
|
|
|
|
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 when update is available."""
|
|
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)
|
|
# Mock version check so upgrade proceeds regardless of local install type
|
|
monkeypatch.setattr(
|
|
cli_mod,
|
|
"_check_for_update",
|
|
lambda info: (True, "update available (abc12345 → def67890)"),
|
|
)
|
|
|
|
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 force=False: 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 force=False: calls.append(True))
|
|
|
|
with patch("sys.argv", ["muxplex", "update"]):
|
|
main()
|
|
|
|
assert len(calls) == 1, "upgrade() must be called once for 'update' subcommand"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Smart version-check tests (_get_install_info / _check_for_update)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_install_info_returns_dict():
|
|
"""_get_install_info must return a dict with all required keys."""
|
|
from muxplex.cli import _get_install_info
|
|
|
|
info = _get_install_info()
|
|
assert "source" in info
|
|
assert "version" in info
|
|
assert "commit" in info
|
|
assert "url" in info
|
|
assert info["source"] in ("git", "editable", "pypi", "unknown")
|
|
|
|
|
|
def test_check_for_update_editable_returns_false():
|
|
"""Editable installs must never suggest an update."""
|
|
from muxplex.cli import _check_for_update
|
|
|
|
info = {"source": "editable", "version": "0.1.0", "commit": None, "url": None}
|
|
available, msg = _check_for_update(info)
|
|
assert available is False
|
|
assert "editable" in msg
|
|
|
|
|
|
def test_upgrade_force_skips_version_check(monkeypatch, capsys):
|
|
"""upgrade(force=True) must skip the version check and proceed to 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)
|
|
# With force=True the version check must be bypassed entirely
|
|
check_calls = []
|
|
monkeypatch.setattr(
|
|
cli_mod,
|
|
"_check_for_update",
|
|
lambda info: check_calls.append(info) or (True, "should not be reached"),
|
|
)
|
|
|
|
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"
|
|
# uv install must still be attempted
|
|
uv_calls = [c for c in calls if isinstance(c, list) and "uv" in str(c)]
|
|
assert len(uv_calls) > 0, "upgrade(force=True) must still call uv tool install"
|
|
|
|
|
|
def test_upgrade_already_up_to_date_skips_install(monkeypatch, capsys):
|
|
"""upgrade() must print 'up to date' and NOT call uv when version check says current."""
|
|
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)
|
|
monkeypatch.setattr(
|
|
cli_mod,
|
|
"_check_for_update",
|
|
lambda info: (False, "up to date (commit abcd1234)"),
|
|
)
|
|
|
|
cli_mod.upgrade()
|
|
|
|
out = capsys.readouterr().out
|
|
assert "up to date" in out.lower() or "already" in out.lower()
|
|
# uv install must NOT have been called
|
|
uv_calls = [c for c in calls if isinstance(c, list) and "uv" in str(c)]
|
|
assert len(uv_calls) == 0, "uv must NOT be called when already up to date"
|
|
|
|
|
|
def test_upgrade_force_flag_registered():
|
|
"""upgrade --force must be accepted by argparse without error."""
|
|
import io
|
|
|
|
from muxplex.cli import main
|
|
|
|
buf = io.StringIO()
|
|
with patch("sys.argv", ["muxplex", "upgrade", "--help"]):
|
|
try:
|
|
with patch("sys.stdout", buf):
|
|
main()
|
|
except SystemExit:
|
|
pass
|
|
|
|
help_text = buf.getvalue()
|
|
assert "--force" in help_text
|