feat: add muxplex config subcommand — list/get/set/reset settings via CLI
Exposes ~/.config/muxplex/settings.json management without hand-editing: config list — show all settings with (modified) markers config get <key> — print one value config set <key> <value> — set with auto type coercion (bool/int/str/list) config reset [key] — reset one or all to defaults Example: muxplex config set host 0.0.0.0 && muxplex service restart
This commit is contained in:
@@ -104,6 +104,11 @@ All serve options read from `~/.config/muxplex/settings.json` by default. CLI fl
|
||||
| `muxplex upgrade` | Upgrade to latest version and restart service |
|
||||
| `muxplex show-password` | Show the current muxplex password |
|
||||
| `muxplex reset-secret` | Regenerate signing secret (invalidates sessions) |
|
||||
| `muxplex config` | Show all settings with current values |
|
||||
| `muxplex config list` | Show all settings with current values |
|
||||
| `muxplex config get <key>` | Show one setting |
|
||||
| `muxplex config set <key> <value>` | Set a setting (auto-detects type) |
|
||||
| `muxplex config reset [key]` | Reset one or all settings to defaults |
|
||||
|
||||
### Service management
|
||||
|
||||
|
||||
+133
@@ -503,6 +503,115 @@ def upgrade(*, force: bool = False) -> None:
|
||||
doctor()
|
||||
|
||||
|
||||
def config_list() -> None:
|
||||
"""Show all settings with current values."""
|
||||
from muxplex.settings import DEFAULT_SETTINGS, SETTINGS_PATH, load_settings # noqa: PLC0415
|
||||
|
||||
settings = load_settings()
|
||||
print(f"\nmuxplex config ({SETTINGS_PATH})\n")
|
||||
|
||||
for key in DEFAULT_SETTINGS:
|
||||
value = settings.get(key)
|
||||
default = DEFAULT_SETTINGS[key]
|
||||
is_default = value == default
|
||||
marker = "" if is_default else " (modified)"
|
||||
if isinstance(value, str):
|
||||
display = f'"{value}"'
|
||||
elif value is None:
|
||||
display = "null"
|
||||
elif isinstance(value, bool):
|
||||
display = "true" if value else "false"
|
||||
elif isinstance(value, list):
|
||||
display = str(value) if value else "[]"
|
||||
else:
|
||||
display = str(value)
|
||||
print(f" {key}: {display}{marker}")
|
||||
print()
|
||||
|
||||
|
||||
def config_get(key: str) -> None:
|
||||
"""Show one setting value."""
|
||||
from muxplex.settings import DEFAULT_SETTINGS, load_settings # noqa: PLC0415
|
||||
|
||||
if key not in DEFAULT_SETTINGS:
|
||||
print(f"Unknown setting: {key}", file=sys.stderr)
|
||||
print(
|
||||
f"Valid keys: {', '.join(sorted(DEFAULT_SETTINGS.keys()))}", file=sys.stderr
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
settings = load_settings()
|
||||
value = settings.get(key)
|
||||
if isinstance(value, str):
|
||||
print(value)
|
||||
elif value is None:
|
||||
print("null")
|
||||
elif isinstance(value, bool):
|
||||
print("true" if value else "false")
|
||||
else:
|
||||
print(value)
|
||||
|
||||
|
||||
def config_set(key: str, raw_value: str) -> None:
|
||||
"""Set a setting value. Auto-detects type from the default."""
|
||||
import json # noqa: PLC0415
|
||||
|
||||
from muxplex.settings import DEFAULT_SETTINGS, patch_settings # noqa: PLC0415
|
||||
|
||||
if key not in DEFAULT_SETTINGS:
|
||||
print(f"Unknown setting: {key}", file=sys.stderr)
|
||||
print(
|
||||
f"Valid keys: {', '.join(sorted(DEFAULT_SETTINGS.keys()))}", file=sys.stderr
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
default = DEFAULT_SETTINGS[key]
|
||||
|
||||
try:
|
||||
if isinstance(default, bool):
|
||||
value: object = raw_value.lower() in ("true", "1", "yes", "on")
|
||||
elif isinstance(default, int):
|
||||
value = int(raw_value)
|
||||
elif default is None:
|
||||
value = None if raw_value.lower() in ("null", "none", "") else raw_value
|
||||
elif isinstance(default, list):
|
||||
value = json.loads(raw_value) if raw_value else []
|
||||
else:
|
||||
value = raw_value
|
||||
except (ValueError, json.JSONDecodeError) as e:
|
||||
print(f"Invalid value for {key}: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
patch_settings({key: value})
|
||||
print(f" {key}: {value}")
|
||||
|
||||
|
||||
def config_reset(key: str | None = None) -> None:
|
||||
"""Reset one or all settings to defaults."""
|
||||
import copy # noqa: PLC0415
|
||||
|
||||
from muxplex.settings import ( # noqa: PLC0415
|
||||
DEFAULT_SETTINGS,
|
||||
SETTINGS_PATH,
|
||||
patch_settings,
|
||||
save_settings,
|
||||
)
|
||||
|
||||
if key is not None:
|
||||
if key not in DEFAULT_SETTINGS:
|
||||
print(f"Unknown setting: {key}", file=sys.stderr)
|
||||
print(
|
||||
f"Valid keys: {', '.join(sorted(DEFAULT_SETTINGS.keys()))}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
patch_settings({key: DEFAULT_SETTINGS[key]})
|
||||
print(f" {key} reset to: {DEFAULT_SETTINGS[key]}")
|
||||
else:
|
||||
save_settings(copy.deepcopy(DEFAULT_SETTINGS))
|
||||
print(f" All settings reset to defaults ({SETTINGS_PATH})")
|
||||
|
||||
|
||||
def _add_serve_flags(parser: argparse.ArgumentParser) -> None:
|
||||
"""Add --host, --port, --auth, --session-ttl flags to a parser.
|
||||
|
||||
@@ -579,6 +688,19 @@ def main() -> None:
|
||||
help="Force reinstall even if already up to date",
|
||||
)
|
||||
|
||||
config_parser = sub.add_parser("config", help="View and manage settings")
|
||||
config_sub = config_parser.add_subparsers(dest="config_command")
|
||||
config_sub.add_parser("list", help="Show all settings (default)")
|
||||
config_get_parser = config_sub.add_parser("get", help="Show one setting")
|
||||
config_get_parser.add_argument("key", help="Setting key")
|
||||
config_set_parser = config_sub.add_parser("set", help="Set a setting value")
|
||||
config_set_parser.add_argument("key", help="Setting key")
|
||||
config_set_parser.add_argument("value", help="New value")
|
||||
config_reset_parser = config_sub.add_parser("reset", help="Reset to defaults")
|
||||
config_reset_parser.add_argument(
|
||||
"key", nargs="?", help="Setting key (omit to reset all)"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.command == "show-password":
|
||||
@@ -589,6 +711,17 @@ def main() -> None:
|
||||
doctor()
|
||||
elif args.command in ("upgrade", "update"):
|
||||
upgrade(force=getattr(args, "force", False))
|
||||
elif args.command == "config":
|
||||
cmd = getattr(args, "config_command", None)
|
||||
if cmd == "get":
|
||||
config_get(args.key)
|
||||
elif cmd == "set":
|
||||
config_set(args.key, args.value)
|
||||
elif cmd == "reset":
|
||||
config_reset(getattr(args, "key", None))
|
||||
else:
|
||||
# Default: list (no subcommand or explicit "list")
|
||||
config_list()
|
||||
elif args.command == "service":
|
||||
from muxplex.service import ( # noqa: PLC0415
|
||||
service_install,
|
||||
|
||||
@@ -7,10 +7,12 @@ import stat
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, 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 None defaults (settings layer resolves)."""
|
||||
from muxplex.cli import main
|
||||
@@ -22,6 +24,7 @@ def test_main_calls_serve_by_default():
|
||||
host=None, port=None, auth=None, session_ttl=None
|
||||
)
|
||||
|
||||
|
||||
def test_main_passes_custom_host_and_port():
|
||||
"""main() with --host/--port must forward them to serve(); unset flags are None."""
|
||||
from muxplex.cli import main
|
||||
@@ -33,6 +36,7 @@ def test_main_passes_custom_host_and_port():
|
||||
host="192.168.1.1", port=9000, auth=None, session_ttl=None
|
||||
)
|
||||
|
||||
|
||||
def test_main_default_host_is_localhost():
|
||||
"""Default --host must be None (settings layer resolves to 127.0.0.1)."""
|
||||
from muxplex.cli import main
|
||||
@@ -43,6 +47,7 @@ def test_main_default_host_is_localhost():
|
||||
_, kwargs = mock_serve.call_args
|
||||
assert kwargs["host"] is None
|
||||
|
||||
|
||||
def test_main_passes_auth_flag():
|
||||
"""main() with --auth password must forward auth='password'; unset flags are None."""
|
||||
from muxplex.cli import main
|
||||
@@ -54,6 +59,7 @@ def test_main_passes_auth_flag():
|
||||
host=None, port=None, auth="password", session_ttl=None
|
||||
)
|
||||
|
||||
|
||||
def test_main_passes_session_ttl_flag():
|
||||
"""main() with --session-ttl 3600 must forward session_ttl=3600; unset flags are None."""
|
||||
from muxplex.cli import main
|
||||
@@ -65,6 +71,7 @@ def test_main_passes_session_ttl_flag():
|
||||
host=None, port=None, auth=None, session_ttl=3600
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
@@ -84,6 +91,7 @@ def test_show_password_prints_password_from_file(tmp_path, monkeypatch, capsys):
|
||||
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
|
||||
@@ -101,6 +109,7 @@ def test_show_password_no_file(tmp_path, monkeypatch, capsys):
|
||||
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
|
||||
@@ -113,6 +122,7 @@ def test_show_password_pam_mode(monkeypatch, capsys):
|
||||
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
|
||||
@@ -128,6 +138,7 @@ def test_reset_secret_writes_new_secret(tmp_path, monkeypatch):
|
||||
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
|
||||
@@ -142,6 +153,7 @@ def test_reset_secret_sets_0600_permissions(tmp_path, monkeypatch):
|
||||
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
|
||||
@@ -158,6 +170,7 @@ def test_reset_secret_prints_warning(tmp_path, monkeypatch, capsys):
|
||||
f"Expected 'invalid' or 'warning' in output, got: {captured.out!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_check_dependencies_exits_when_ttyd_missing(monkeypatch):
|
||||
"""_check_dependencies() must sys.exit(1) when ttyd is not in PATH."""
|
||||
import shutil
|
||||
@@ -177,6 +190,7 @@ def test_check_dependencies_exits_when_ttyd_missing(monkeypatch):
|
||||
_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
|
||||
@@ -196,6 +210,7 @@ def test_check_dependencies_exits_when_tmux_missing(monkeypatch):
|
||||
_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
|
||||
@@ -206,6 +221,7 @@ def test_check_dependencies_passes_when_all_present(monkeypatch):
|
||||
# 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
|
||||
@@ -219,6 +235,7 @@ def test_main_check_dependencies_called_for_serve(monkeypatch):
|
||||
|
||||
assert len(calls) == 1, "_check_dependencies must be called once for serve"
|
||||
|
||||
|
||||
def test_dunder_main_calls_main():
|
||||
"""python -m muxplex must call cli.main()."""
|
||||
import importlib.util
|
||||
@@ -231,10 +248,12 @@ def test_dunder_main_calls_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
|
||||
@@ -243,6 +262,7 @@ def test_doctor_shows_python_version(capsys):
|
||||
out = capsys.readouterr().out
|
||||
assert "Python" in out
|
||||
|
||||
|
||||
def test_doctor_checks_tmux(capsys, monkeypatch):
|
||||
"""doctor must check for tmux."""
|
||||
import subprocess
|
||||
@@ -263,6 +283,7 @@ def test_doctor_checks_tmux(capsys, monkeypatch):
|
||||
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
|
||||
@@ -280,6 +301,7 @@ def test_doctor_reports_missing_ttyd(capsys, monkeypatch):
|
||||
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
|
||||
@@ -288,6 +310,7 @@ def test_doctor_shows_platform(capsys):
|
||||
out = capsys.readouterr().out
|
||||
assert "Platform" in out
|
||||
|
||||
|
||||
def test_doctor_subcommand_registered():
|
||||
"""doctor must be a valid subcommand in main() argparse."""
|
||||
import io
|
||||
@@ -305,6 +328,7 @@ def test_doctor_subcommand_registered():
|
||||
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
|
||||
@@ -319,10 +343,12 @@ def test_main_dispatches_to_doctor(monkeypatch):
|
||||
"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
|
||||
@@ -340,6 +366,7 @@ def test_upgrade_subcommand_registered():
|
||||
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
|
||||
@@ -357,6 +384,7 @@ def test_update_alias_registered():
|
||||
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
|
||||
@@ -386,6 +414,7 @@ def test_upgrade_calls_uv_tool_install(monkeypatch, capsys):
|
||||
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
|
||||
@@ -398,6 +427,7 @@ def test_main_dispatches_to_upgrade(monkeypatch):
|
||||
|
||||
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
|
||||
@@ -410,10 +440,12 @@ def test_main_dispatches_update_to_upgrade(monkeypatch):
|
||||
|
||||
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
|
||||
@@ -425,6 +457,7 @@ def test_get_install_info_returns_dict():
|
||||
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
|
||||
@@ -434,6 +467,7 @@ def test_check_for_update_editable_returns_false():
|
||||
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
|
||||
@@ -466,6 +500,7 @@ def test_upgrade_force_skips_version_check(monkeypatch, capsys):
|
||||
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
|
||||
@@ -496,6 +531,7 @@ def test_upgrade_already_up_to_date_skips_install(monkeypatch, capsys):
|
||||
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
|
||||
@@ -513,10 +549,12 @@ def test_upgrade_force_flag_registered():
|
||||
help_text = buf.getvalue()
|
||||
assert "--force" in help_text
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# serve() settings.json integration tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_serve_reads_host_from_settings(tmp_path, monkeypatch):
|
||||
"""serve(host=None) must use host from settings.json."""
|
||||
settings_file = tmp_path / "settings.json"
|
||||
@@ -538,6 +576,7 @@ def test_serve_reads_host_from_settings(tmp_path, monkeypatch):
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["host"] == "192.168.0.1"
|
||||
|
||||
|
||||
def test_serve_cli_flag_overrides_settings(tmp_path, monkeypatch):
|
||||
"""serve(host='10.0.0.1') must override settings.json host."""
|
||||
settings_file = tmp_path / "settings.json"
|
||||
@@ -559,6 +598,7 @@ def test_serve_cli_flag_overrides_settings(tmp_path, monkeypatch):
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["host"] == "10.0.0.1"
|
||||
|
||||
|
||||
def test_serve_falls_back_to_default_when_no_settings_file(tmp_path, monkeypatch):
|
||||
"""serve() with no settings file and no CLI flags uses hardcoded defaults."""
|
||||
settings_file = tmp_path / "nonexistent_settings.json"
|
||||
@@ -581,6 +621,7 @@ def test_serve_falls_back_to_default_when_no_settings_file(tmp_path, monkeypatch
|
||||
assert calls[0]["host"] == "127.0.0.1"
|
||||
assert calls[0]["port"] == 8088
|
||||
|
||||
|
||||
def test_serve_port_from_settings(tmp_path, monkeypatch):
|
||||
"""serve(port=None) must use port from settings.json."""
|
||||
settings_file = tmp_path / "settings.json"
|
||||
@@ -602,6 +643,7 @@ def test_serve_port_from_settings(tmp_path, monkeypatch):
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["port"] == 9999
|
||||
|
||||
|
||||
def test_serve_session_ttl_from_settings(tmp_path, monkeypatch):
|
||||
"""serve(session_ttl=None) must use session_ttl from settings.json."""
|
||||
settings_file = tmp_path / "settings.json"
|
||||
@@ -618,6 +660,7 @@ def test_serve_session_ttl_from_settings(tmp_path, monkeypatch):
|
||||
|
||||
assert os.environ.get("MUXPLEX_SESSION_TTL") == "3600"
|
||||
|
||||
|
||||
def test_serve_session_ttl_zero_is_valid(tmp_path, monkeypatch):
|
||||
"""serve(session_ttl=0) must work — 0 means browser session, a valid value."""
|
||||
settings_file = tmp_path / "settings.json"
|
||||
@@ -634,11 +677,13 @@ def test_serve_session_ttl_zero_is_valid(tmp_path, monkeypatch):
|
||||
|
||||
assert os.environ.get("MUXPLEX_SESSION_TTL") == "0"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# argparse refactoring tests — None defaults, serve flags on both parsers,
|
||||
# upgrade alias
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_main_passes_none_for_unset_flags():
|
||||
"""main() with no flags passes None for host/port/auth/session_ttl to serve()."""
|
||||
from muxplex.cli import main
|
||||
@@ -650,6 +695,7 @@ def test_main_passes_none_for_unset_flags():
|
||||
host=None, port=None, auth=None, session_ttl=None
|
||||
)
|
||||
|
||||
|
||||
def test_main_passes_explicit_host_only():
|
||||
"""main() with --host 10.0.0.1 passes host='10.0.0.1', others as None."""
|
||||
from muxplex.cli import main
|
||||
@@ -661,6 +707,7 @@ def test_main_passes_explicit_host_only():
|
||||
host="10.0.0.1", port=None, auth=None, session_ttl=None
|
||||
)
|
||||
|
||||
|
||||
def test_main_serve_subcommand_accepts_flags():
|
||||
"""'muxplex serve --host 10.0.0.1 --port 9000' passes values to serve()."""
|
||||
from muxplex.cli import main
|
||||
@@ -674,6 +721,7 @@ def test_main_serve_subcommand_accepts_flags():
|
||||
host="10.0.0.1", port=9000, auth=None, session_ttl=None
|
||||
)
|
||||
|
||||
|
||||
def test_help_shows_single_upgrade_line():
|
||||
"""Help output shows 'upgrade (update)' alias notation, not two separate subcommand entries."""
|
||||
import io
|
||||
@@ -696,6 +744,7 @@ def test_help_shows_single_upgrade_line():
|
||||
f"Got help text:\n{help_text}"
|
||||
)
|
||||
|
||||
|
||||
def test_doctor_shows_serve_config(tmp_path, monkeypatch, capsys):
|
||||
"""doctor() must show the current serve config (host, port, auth)."""
|
||||
import json
|
||||
@@ -717,10 +766,12 @@ def test_doctor_shows_serve_config(tmp_path, monkeypatch, capsys):
|
||||
assert "9999" in out
|
||||
assert "password" in out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# service subcommand dispatch tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_service_install_dispatches():
|
||||
"""muxplex service install must call service_install()."""
|
||||
from muxplex.cli import main
|
||||
@@ -730,6 +781,7 @@ def test_service_install_dispatches():
|
||||
main()
|
||||
mock_fn.assert_called_once()
|
||||
|
||||
|
||||
def test_service_uninstall_dispatches():
|
||||
"""muxplex service uninstall must call service_uninstall()."""
|
||||
from muxplex.cli import main
|
||||
@@ -739,6 +791,7 @@ def test_service_uninstall_dispatches():
|
||||
main()
|
||||
mock_fn.assert_called_once()
|
||||
|
||||
|
||||
def test_service_start_dispatches():
|
||||
"""muxplex service start must call service_start()."""
|
||||
from muxplex.cli import main
|
||||
@@ -748,6 +801,7 @@ def test_service_start_dispatches():
|
||||
main()
|
||||
mock_fn.assert_called_once()
|
||||
|
||||
|
||||
def test_service_stop_dispatches():
|
||||
"""muxplex service stop must call service_stop()."""
|
||||
from muxplex.cli import main
|
||||
@@ -757,6 +811,7 @@ def test_service_stop_dispatches():
|
||||
main()
|
||||
mock_fn.assert_called_once()
|
||||
|
||||
|
||||
def test_service_restart_dispatches():
|
||||
"""muxplex service restart must call service_restart()."""
|
||||
from muxplex.cli import main
|
||||
@@ -766,6 +821,7 @@ def test_service_restart_dispatches():
|
||||
main()
|
||||
mock_fn.assert_called_once()
|
||||
|
||||
|
||||
def test_service_status_dispatches():
|
||||
"""muxplex service status must call service_status()."""
|
||||
from muxplex.cli import main
|
||||
@@ -775,6 +831,7 @@ def test_service_status_dispatches():
|
||||
main()
|
||||
mock_fn.assert_called_once()
|
||||
|
||||
|
||||
def test_service_logs_dispatches():
|
||||
"""muxplex service logs must call service_logs()."""
|
||||
from muxplex.cli import main
|
||||
@@ -784,6 +841,7 @@ def test_service_logs_dispatches():
|
||||
main()
|
||||
mock_fn.assert_called_once()
|
||||
|
||||
|
||||
def test_service_subcommand_in_help():
|
||||
"""'service' must appear in muxplex --help output."""
|
||||
import io
|
||||
@@ -801,10 +859,12 @@ def test_service_subcommand_in_help():
|
||||
help_text = buf.getvalue().lower()
|
||||
assert "service" in help_text
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# task-6: Verify old launchd/systemd helpers removed from cli.py
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
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
|
||||
@@ -813,6 +873,7 @@ def test_old_install_launchd_removed_from_cli():
|
||||
"_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
|
||||
@@ -821,6 +882,145 @@ def test_old_install_systemd_removed_from_cli():
|
||||
"_install_systemd should be removed from cli.py; functionality is in muxplex.service"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# config subcommand tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_config_list_shows_all_keys(capsys, tmp_path, monkeypatch):
|
||||
"""config list must show all DEFAULT_SETTINGS keys."""
|
||||
import muxplex.settings as settings_mod
|
||||
|
||||
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "s.json")
|
||||
|
||||
from muxplex.cli import config_list
|
||||
|
||||
config_list()
|
||||
out = capsys.readouterr().out
|
||||
for key in settings_mod.DEFAULT_SETTINGS:
|
||||
assert key in out, f"config list must show '{key}'"
|
||||
|
||||
|
||||
def test_config_get_returns_value(capsys, tmp_path, monkeypatch):
|
||||
"""config get must return the value of a known key."""
|
||||
import muxplex.settings as settings_mod
|
||||
|
||||
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "s.json")
|
||||
|
||||
from muxplex.cli import config_get
|
||||
|
||||
config_get("port")
|
||||
out = capsys.readouterr().out.strip()
|
||||
assert out == "8088"
|
||||
|
||||
|
||||
def test_config_get_unknown_key_exits(tmp_path, monkeypatch):
|
||||
"""config get with unknown key must exit 1."""
|
||||
import pytest
|
||||
import muxplex.settings as settings_mod
|
||||
|
||||
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "s.json")
|
||||
|
||||
from muxplex.cli import config_get
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
config_get("nonexistent_key")
|
||||
|
||||
|
||||
def test_config_set_persists_value(tmp_path, monkeypatch):
|
||||
"""config set must persist the value to settings.json."""
|
||||
import muxplex.settings as settings_mod
|
||||
|
||||
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "s.json")
|
||||
|
||||
from muxplex.cli import config_set
|
||||
|
||||
config_set("host", "0.0.0.0")
|
||||
|
||||
settings = settings_mod.load_settings()
|
||||
assert settings["host"] == "0.0.0.0"
|
||||
|
||||
|
||||
def test_config_set_coerces_int(tmp_path, monkeypatch):
|
||||
"""config set must coerce port to int."""
|
||||
import muxplex.settings as settings_mod
|
||||
|
||||
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "s.json")
|
||||
|
||||
from muxplex.cli import config_set
|
||||
|
||||
config_set("port", "9090")
|
||||
|
||||
settings = settings_mod.load_settings()
|
||||
assert settings["port"] == 9090
|
||||
|
||||
|
||||
def test_config_set_coerces_bool(tmp_path, monkeypatch):
|
||||
"""config set must coerce booleans."""
|
||||
import muxplex.settings as settings_mod
|
||||
|
||||
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "s.json")
|
||||
|
||||
from muxplex.cli import config_set
|
||||
|
||||
config_set("window_size_largest", "true")
|
||||
|
||||
settings = settings_mod.load_settings()
|
||||
assert settings["window_size_largest"] is True
|
||||
|
||||
|
||||
def test_config_reset_all(tmp_path, monkeypatch):
|
||||
"""config reset (no key) must reset all settings to defaults."""
|
||||
import muxplex.settings as settings_mod
|
||||
|
||||
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "s.json")
|
||||
|
||||
from muxplex.cli import config_set, config_reset
|
||||
|
||||
config_set("host", "0.0.0.0")
|
||||
config_set("port", "9090")
|
||||
config_reset(None)
|
||||
|
||||
settings = settings_mod.load_settings()
|
||||
assert settings["host"] == "127.0.0.1"
|
||||
assert settings["port"] == 8088
|
||||
|
||||
|
||||
def test_config_reset_single_key(tmp_path, monkeypatch):
|
||||
"""config reset <key> must reset only that key."""
|
||||
import muxplex.settings as settings_mod
|
||||
|
||||
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "s.json")
|
||||
|
||||
from muxplex.cli import config_set, config_reset
|
||||
|
||||
config_set("host", "0.0.0.0")
|
||||
config_set("port", "9090")
|
||||
config_reset("host")
|
||||
|
||||
settings = settings_mod.load_settings()
|
||||
assert settings["host"] == "127.0.0.1"
|
||||
assert settings["port"] == 9090 # unchanged
|
||||
|
||||
|
||||
def test_config_subcommand_registered():
|
||||
"""config must appear in --help."""
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "muxplex", "config", "--help"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert "list" in result.stdout
|
||||
assert "get" in result.stdout
|
||||
assert "set" in result.stdout
|
||||
assert "reset" in result.stdout
|
||||
|
||||
|
||||
def test_upgrade_uses_service_module_install(monkeypatch, capsys):
|
||||
"""upgrade() must call muxplex.service.service_install."""
|
||||
import subprocess
|
||||
|
||||
Reference in New Issue
Block a user