feat: add muxplex doctor subcommand — dependency + config diagnostics
Checks Python version, tmux, ttyd, muxplex version, settings file, auth status (PAM/password/auto-generate), active tmux sessions, platform, and service installation status. Platform-specific install hints for missing dependencies. Non-fatal — always completes.
This commit is contained in:
+140
-1
@@ -2,13 +2,20 @@
|
|||||||
|
|
||||||
import argparse
|
import argparse
|
||||||
import os
|
import os
|
||||||
|
import platform
|
||||||
import shutil
|
import shutil
|
||||||
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import secrets as _secrets
|
import secrets as _secrets
|
||||||
|
|
||||||
from muxplex.auth import get_secret_path, load_password, pam_available
|
from muxplex.auth import (
|
||||||
|
get_password_path,
|
||||||
|
get_secret_path,
|
||||||
|
load_password,
|
||||||
|
pam_available,
|
||||||
|
)
|
||||||
|
|
||||||
# Module-level path constants (overridable in tests via monkeypatch)
|
# Module-level path constants (overridable in tests via monkeypatch)
|
||||||
_system_service_path = Path("/etc/systemd/system/muxplex.service")
|
_system_service_path = Path("/etc/systemd/system/muxplex.service")
|
||||||
@@ -57,6 +64,134 @@ def serve(
|
|||||||
uvicorn.run(app, host=host, port=port, log_level="warning")
|
uvicorn.run(app, host=host, port=port, log_level="warning")
|
||||||
|
|
||||||
|
|
||||||
|
def doctor() -> None:
|
||||||
|
"""Run diagnostic checks and report system status."""
|
||||||
|
ok_mark = "\033[32m✓\033[0m" # green check
|
||||||
|
fail_mark = "\033[31m✗\033[0m" # red x
|
||||||
|
warn_mark = "\033[33m!\033[0m" # yellow warning
|
||||||
|
|
||||||
|
print("\nmuxplex doctor\n")
|
||||||
|
|
||||||
|
# Python version
|
||||||
|
py_version = platform.python_version()
|
||||||
|
py_ok = tuple(int(x) for x in py_version.split(".")[:2]) >= (3, 11)
|
||||||
|
print(
|
||||||
|
f" {ok_mark if py_ok else fail_mark} Python {py_version}"
|
||||||
|
+ ("" if py_ok else " (3.11+ required)")
|
||||||
|
)
|
||||||
|
|
||||||
|
# tmux
|
||||||
|
tmux_path = shutil.which("tmux")
|
||||||
|
if tmux_path:
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
["tmux", "-V"], capture_output=True, text=True, timeout=5
|
||||||
|
)
|
||||||
|
tmux_version = result.stdout.strip()
|
||||||
|
print(f" {ok_mark} {tmux_version}")
|
||||||
|
except Exception:
|
||||||
|
print(f" {ok_mark} tmux (version unknown)")
|
||||||
|
else:
|
||||||
|
print(f" {fail_mark} tmux — not found")
|
||||||
|
if sys.platform == "darwin":
|
||||||
|
print(" Install: brew install tmux")
|
||||||
|
else:
|
||||||
|
print(" Install: sudo apt install tmux")
|
||||||
|
|
||||||
|
# ttyd
|
||||||
|
ttyd_path = shutil.which("ttyd")
|
||||||
|
if ttyd_path:
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
["ttyd", "--version"], capture_output=True, text=True, timeout=5
|
||||||
|
)
|
||||||
|
ttyd_version = result.stdout.strip() or result.stderr.strip()
|
||||||
|
print(f" {ok_mark} ttyd {ttyd_version}")
|
||||||
|
except Exception:
|
||||||
|
print(f" {ok_mark} ttyd (version unknown)")
|
||||||
|
else:
|
||||||
|
print(f" {fail_mark} ttyd — not found")
|
||||||
|
if sys.platform == "darwin":
|
||||||
|
print(" Install: brew install ttyd")
|
||||||
|
else:
|
||||||
|
print(" Install: sudo apt install ttyd")
|
||||||
|
|
||||||
|
# muxplex version
|
||||||
|
try:
|
||||||
|
from importlib.metadata import version as pkg_version # noqa: PLC0415
|
||||||
|
|
||||||
|
muxplex_version = pkg_version("muxplex")
|
||||||
|
except Exception:
|
||||||
|
muxplex_version = "dev"
|
||||||
|
print(f" {ok_mark} muxplex {muxplex_version}")
|
||||||
|
|
||||||
|
# Settings file
|
||||||
|
from muxplex.settings import SETTINGS_PATH # noqa: PLC0415
|
||||||
|
|
||||||
|
if SETTINGS_PATH.exists():
|
||||||
|
print(f" {ok_mark} Settings: {SETTINGS_PATH}")
|
||||||
|
else:
|
||||||
|
print(
|
||||||
|
f" {warn_mark} Settings: {SETTINGS_PATH} (not yet created — will use defaults)"
|
||||||
|
)
|
||||||
|
|
||||||
|
# Auth status
|
||||||
|
pw_path = get_password_path()
|
||||||
|
if pam_available():
|
||||||
|
import pwd # noqa: PLC0415
|
||||||
|
|
||||||
|
username = pwd.getpwuid(os.getuid()).pw_name
|
||||||
|
print(f" {ok_mark} Auth: PAM available (user: {username})")
|
||||||
|
elif pw_path.exists():
|
||||||
|
print(f" {ok_mark} Auth: password file ({pw_path})")
|
||||||
|
elif os.environ.get("MUXPLEX_PASSWORD"):
|
||||||
|
print(f" {ok_mark} Auth: password (env var)")
|
||||||
|
else:
|
||||||
|
print(f" {warn_mark} Auth: no PAM, no password — will auto-generate on serve")
|
||||||
|
|
||||||
|
# tmux sessions (if tmux is available)
|
||||||
|
if tmux_path:
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
["tmux", "list-sessions", "-F", "#{session_name}"],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
timeout=5,
|
||||||
|
)
|
||||||
|
if result.returncode == 0:
|
||||||
|
sessions = [s for s in result.stdout.strip().split("\n") if s]
|
||||||
|
print(f" {ok_mark} tmux sessions: {len(sessions)} active")
|
||||||
|
else:
|
||||||
|
print(f" {warn_mark} tmux server not running (no sessions)")
|
||||||
|
except Exception:
|
||||||
|
print(f" {warn_mark} tmux server not running")
|
||||||
|
|
||||||
|
# Platform + service status
|
||||||
|
print(f" {ok_mark} Platform: {sys.platform} ({platform.machine()})")
|
||||||
|
if sys.platform == "darwin":
|
||||||
|
plist = Path.home() / "Library" / "LaunchAgents" / "com.muxplex.plist"
|
||||||
|
if plist.exists():
|
||||||
|
print(f" {ok_mark} Service: launchd agent installed ({plist})")
|
||||||
|
else:
|
||||||
|
print(
|
||||||
|
f" {warn_mark} Service: not installed (run: muxplex install-service)"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
systemd_user = Path.home() / ".config" / "systemd" / "user" / "muxplex.service"
|
||||||
|
if systemd_user.exists():
|
||||||
|
print(f" {ok_mark} Service: systemd user unit installed ({systemd_user})")
|
||||||
|
elif _system_service_path.exists():
|
||||||
|
print(
|
||||||
|
f" {ok_mark} Service: systemd system unit installed ({_system_service_path})"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
print(
|
||||||
|
f" {warn_mark} Service: not installed (run: muxplex install-service)"
|
||||||
|
)
|
||||||
|
|
||||||
|
print() # trailing newline
|
||||||
|
|
||||||
|
|
||||||
def _check_dependencies() -> None:
|
def _check_dependencies() -> None:
|
||||||
"""Verify required external programs are installed.
|
"""Verify required external programs are installed.
|
||||||
|
|
||||||
@@ -206,6 +341,8 @@ def main() -> None:
|
|||||||
"reset-secret", help="Regenerate signing secret (invalidates sessions)"
|
"reset-secret", help="Regenerate signing secret (invalidates sessions)"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
sub.add_parser("doctor", help="Check dependencies and system status")
|
||||||
|
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
|
|
||||||
if args.command == "install-service":
|
if args.command == "install-service":
|
||||||
@@ -214,6 +351,8 @@ def main() -> None:
|
|||||||
show_password()
|
show_password()
|
||||||
elif args.command == "reset-secret":
|
elif args.command == "reset-secret":
|
||||||
reset_secret()
|
reset_secret()
|
||||||
|
elif args.command == "doctor":
|
||||||
|
doctor()
|
||||||
else:
|
else:
|
||||||
_check_dependencies()
|
_check_dependencies()
|
||||||
serve(
|
serve(
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
"""Tests for muxplex/cli.py — CLI entry point."""
|
"""Tests for muxplex/cli.py — CLI entry point."""
|
||||||
|
|
||||||
|
import shutil
|
||||||
import stat
|
import stat
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
@@ -420,3 +421,98 @@ def test_dunder_main_calls_main():
|
|||||||
with patch("muxplex.cli.main") as mock_main:
|
with patch("muxplex.cli.main") as mock_main:
|
||||||
exec(Path(spec.origin).read_text()) # noqa: S102
|
exec(Path(spec.origin).read_text()) # noqa: S102
|
||||||
mock_main.assert_called_once()
|
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"
|
||||||
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user