the_og/better-claude-config
1
0
Fork 0
Code Issues 8 Pull Requests 2 Actions Packages Projects Releases 5 Wiki Activity
Files
2e5d0351b42f203905e18df60f67d4f519c73ebe
better-claude-config/tests/test_core.py
T
Cowork SupervisorandClaude Opus 4.8 2e5d0351b4
CI / Lint (ruff) (pull_request) Successful in 9s
Details
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 22s
Details
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 13s
Details
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 15s
Details
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 13s
Details
CI / Catalog signature (pull_request) Successful in 8s
Details
test(#91): assert sidecar paths via as_posix() for the Windows CI runner 
The CI matrix includes a windows-latest job where str(Path("/Users/…")) renders
with backslashes, so the platform-parameterised sidecar-path assertions failed
there (macOS/Linux passed). Compare .as_posix() instead — separator-normalised
and portable — and note why. Pure test fix; no production-code change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-12 03:07:47 -04:00

3597 lines
138 KiB
Python
Raw Blame History

"""Pytest port of the original test_core.py script (same 23 behaviours, now
proper test functions with tmp_path/monkeypatch fixtures)."""
import dataclasses
import json
import os
import re
import sys
import urllib.error
import urllib.request
from pathlib import Path
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
import bcc_core as c
# --------------------------------------------------------------------------- #
# 1. Discovery
# --------------------------------------------------------------------------- #
@pytest.fixture
def fake_base(tmp_path, monkeypatch):
base = tmp_path / "AppSupport"
for d in ("Claude", "Claude-Work", "NotClaude"):
(base / d).mkdir(parents=True)
(base / "Claude" / c.CONFIG_FILENAME).write_text("{}")
monkeypatch.setattr(c, "app_support_base", lambda: base)
return base
def test_discovers_claude_dirs_not_others(fake_base):
labels = sorted(p.label for p in c.discover_profiles())
assert "Claude" in labels
assert "Claude-Work" in labels
assert "NotClaude" not in labels
def test_flags_missing_config_file(fake_base):
profs = c.discover_profiles()
assert any(p.label == "Claude-Work" and not p.config_exists for p in profs)
@pytest.fixture
def fake_home(tmp_path, monkeypatch, fake_base):
home = tmp_path / "home"
(home / ".claude").mkdir(parents=True)
monkeypatch.setattr(c.Path, "home", lambda: home)
return home
def test_claude_code_uses_dot_claude_json(fake_home):
(fake_home / ".claude.json").write_text('{"mcpServers": {"x": {"command": "npx"}}}')
profs = c.discover_profiles()
cc = [p for p in profs if p.label == "Claude Code"]
assert len(cc) == 1
assert cc[0].path == fake_home / ".claude.json"
assert cc[0].config_exists
def test_legacy_settings_json_surfaced_only_with_servers(fake_home):
# settings.json without mcpServers -> not listed
(fake_home / ".claude" / "settings.json").write_text('{"permissions": {}}')
assert not any("legacy" in p.label for p in c.discover_profiles())
# settings.json WITH parked mcpServers -> listed so the user can migrate
(fake_home / ".claude" / "settings.json").write_text(
'{"mcpServers": {"old": {"command": "npx"}}}'
)
assert any("legacy" in p.label for p in c.discover_profiles())
def test_project_with_mcp_json_is_discovered(tmp_path):
proj = tmp_path / "my-project"
proj.mkdir()
(proj / ".mcp.json").write_text('{"mcpServers": {"x": {"command": "npx"}}}')
claude_json = tmp_path / ".claude.json"
claude_json.write_text(json.dumps({"projects": {str(proj): {}}}))
profs = c.discover_project_configs(claude_json)
assert len(profs) == 1
assert profs[0].label == f"Project: {proj.name}"
assert profs[0].path == proj / ".mcp.json"
assert profs[0].config_exists
def test_project_without_mcp_json_not_listed(tmp_path):
proj = tmp_path / "no-mcp-project"
proj.mkdir()
claude_json = tmp_path / ".claude.json"
claude_json.write_text(json.dumps({"projects": {str(proj): {}}}))
assert c.discover_project_configs(claude_json) == []
def test_projects_missing_or_not_dict_yields_no_profiles(tmp_path):
claude_json = tmp_path / ".claude.json"
claude_json.write_text(json.dumps({}))
assert c.discover_project_configs(claude_json) == []
claude_json.write_text(json.dumps({"projects": ["not", "a", "dict"]}))
assert c.discover_project_configs(claude_json) == []
def test_malformed_claude_json_fails_quiet(fake_home):
(fake_home / ".claude.json").write_text("{not valid json")
assert c.discover_project_configs(fake_home / ".claude.json") == []
# discover_profiles as a whole must still work and just skip project profiles
profs = c.discover_profiles()
assert not any(p.label.startswith("Project:") for p in profs)
def test_project_configs_deduped_against_existing_profiles(fake_home):
# Point a project directly at the Claude Code profile's own .mcp.json-shaped
# path to prove discover_profiles() won't duplicate an already-listed path.
proj = fake_home / "dup-project"
proj.mkdir()
mcp_path = proj / ".mcp.json"
mcp_path.write_text('{"mcpServers": {"x": {"command": "npx"}}}')
(fake_home / ".claude.json").write_text(json.dumps({"projects": {str(proj): {}}}))
direct = c.discover_project_configs(fake_home / ".claude.json")
assert len(direct) == 1
profs = c.discover_profiles()
project_profiles = [p for p in profs if str(p.path) == str(mcp_path)]
assert len(project_profiles) == 1
# --------------------------------------------------------------------------- #
# 2. Write pipeline: preserves other keys + order, only touches mcpServers
# --------------------------------------------------------------------------- #
@pytest.fixture
def written_config(tmp_path):
cfgpath = tmp_path / "real.json"
original = {
"globalShortcut": "Cmd+Shift+Space",
"mcpServers": {"old": {"command": "node", "args": ["x.js"]}},
"someOtherTool": {"keep": True},
}
cfgpath.write_text(json.dumps(original, indent=2))
cfg = c.load_config(cfgpath)
servers = c.extract_servers(cfg)
servers.append(c.ServerEntry("brave", {"command": "npx", "args": ["-y", "@x/brave"]}, True))
servers[0].enabled = False
c.apply_servers(cfg, servers)
c.write_config(cfgpath, cfg)
return cfgpath, json.loads(cfgpath.read_text())
def test_preserves_unrelated_keys(written_config):
_, written = written_config
assert written.get("globalShortcut") == "Cmd+Shift+Space"
assert written.get("someOtherTool") == {"keep": True}
def test_disabled_server_parked(written_config):
_, written = written_config
assert "old" in written.get(c.DISABLED_KEY, {})
assert "old" not in written.get("mcpServers", {})
def test_enabled_server_active(written_config):
_, written = written_config
assert "brave" in written.get("mcpServers", {})
def test_key_order_kept(written_config):
_, written = written_config
keys = list(written.keys())
assert keys.index("globalShortcut") < keys.index("mcpServers")
def test_backup_created(written_config):
cfgpath, _ = written_config
bdir = cfgpath.parent / c.BACKUP_DIRNAME
assert bdir.is_dir()
assert any(bdir.iterdir())
# --------------------------------------------------------------------------- #
# 3. Paste parser — all accepted shapes
# --------------------------------------------------------------------------- #
def test_parse_full_config_shape():
full = '{"mcpServers": {"a": {"command": "node", "args": ["a.js"]}}}'
assert list(c.parse_pasted_json(full)) == ["a"]
def test_parse_inner_block_shape():
inner = '{"b": {"command": "uvx", "args": ["mcp-server-git"]}}'
assert list(c.parse_pasted_json(inner)) == ["b"]
def test_parse_bare_server_suggests_name():
bare = '{"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]}'
assert "filesystem" in c.parse_pasted_json(bare)
def test_parse_remote_suggests_host_name():
remote = '{"url": "https://mcp.asana.com/sse"}'
assert "mcp" in c.parse_pasted_json(remote)
def test_rejects_junk():
with pytest.raises(ValueError):
c.parse_pasted_json('{"junk": 5}')
# --------------------------------------------------------------------------- #
# 4. Validation
# --------------------------------------------------------------------------- #
@pytest.fixture
def problems():
bad = [
c.ServerEntry("", {"command": "x"}, True),
c.ServerEntry("dup", {"command": "x"}, True),
c.ServerEntry("dup", {"command": "x"}, True),
c.ServerEntry("r", {"url": ""}, True),
c.ServerEntry("nocmd", {"command": ""}, True),
]
return c.validate_servers(bad)
def test_validation_catches_empty_name(problems):
assert any("empty name" in p for p in problems)
def test_validation_catches_duplicate(problems):
assert any("Duplicate" in p for p in problems)
def test_validation_catches_missing_url(problems):
assert any("no URL" in p for p in problems)
def test_validation_catches_missing_command(problems):
assert any("no command" in p for p in problems)
def test_valid_set_passes_clean():
assert c.validate_servers([c.ServerEntry("good", {"command": "node"}, True)]) == []
# --------------------------------------------------------------------------- #
# 4b. Lint: non-blocking structural warnings
# --------------------------------------------------------------------------- #
def test_lint_flags_non_string_command():
warnings = c.lint_server("x", {"command": 5})
assert any("'command' should be a string" in w for w in warnings)
def test_lint_flags_args_not_a_list():
warnings = c.lint_server("x", {"command": "node", "args": "-y foo"})
assert any("'args' should be a list" in w for w in warnings)
def test_lint_flags_args_with_non_string_items():
warnings = c.lint_server("x", {"command": "node", "args": ["-y", 5]})
assert any("'args' contains non-string values" in w for w in warnings)
def test_lint_flags_env_not_a_dict():
warnings = c.lint_server("x", {"command": "node", "env": ["FOO=bar"]})
assert any("'env' should be an object" in w for w in warnings)
def test_lint_flags_env_with_non_string_values():
warnings = c.lint_server("x", {"command": "node", "env": {"FOO": 5}})
assert any("'env' contains non-string values" in w for w in warnings)
def test_lint_flags_headers_not_a_dict():
warnings = c.lint_server("x", {"url": "https://x", "headers": ["Authorization: x"]})
assert any("'headers' should be an object" in w for w in warnings)
def test_lint_flags_headers_with_non_string_values():
warnings = c.lint_server("x", {"url": "https://x", "headers": {"Authorization": 5}})
assert any("'headers' contains non-string values" in w for w in warnings)
def test_lint_flags_bad_type_value():
warnings = c.lint_server("x", {"url": "https://x", "type": "websocket"})
assert any("'type'" in w and "websocket" in w for w in warnings)
def test_lint_flags_extra_unknown_fields():
warnings = c.lint_server("x", {"command": "node", "cwd": "/tmp", "timeout": 30})
matches = [w for w in warnings if "extra fields preserved as-is" in w]
assert len(matches) == 1
assert "cwd" in matches[0]
assert "timeout" in matches[0]
def test_lint_clean_server_yields_no_warnings():
assert c.lint_server("good", {"command": "node", "args": ["a.js"]}) == []
assert (
c.lint_server(
"remote", {"url": "https://x", "type": "http", "headers": {"Authorization": "x"}}
)
== []
)
def test_lint_servers_aggregates_across_entries():
entries = [
c.ServerEntry("a", {"command": 5}, True),
c.ServerEntry("b", {"url": "https://x", "type": "bogus"}, True),
]
warnings = c.lint_servers(entries)
assert any("'a'" in w and "'command' should be a string" in w for w in warnings)
assert any("'b'" in w and "'type'" in w for w in warnings)
def test_lint_warning_is_not_a_blocking_problem():
# args given as a single string isn't caught by validate_servers (a
# command is still present), but it's a lint warning.
entry = c.ServerEntry("x", {"command": "node", "args": "-y foo"}, True)
assert c.validate_servers([entry]) == []
assert c.lint_servers([entry]) != []
# --------------------------------------------------------------------------- #
# 5. Secrets: detection + redaction
# --------------------------------------------------------------------------- #
def test_is_secret_key_positives():
for k in (
"BRAVE_API_KEY",
"BEARER_TOKEN",
"password",
"GH_TOKEN",
"Authorization",
"client_secret",
"PRIVATE_KEY",
"aws_access_key_id",
):
assert c.is_secret_key(k), k
def test_is_secret_key_negatives():
for k in ("JOOMLA_BASE_URL", "PORT", "HOME", "PATH", "debug", "TIMEOUT"):
assert not c.is_secret_key(k), k
def test_redact_args_flag_value():
assert c.redact_args(["--token", "abc123", "run"]) == ["--token", c.MASK, "run"]
def test_redact_args_inline_flag():
assert c.redact_args(["--api-key=abc123"]) == [f"--api-key={c.MASK}"]
def test_redact_args_known_prefix():
assert c.redact_args(["ghp_abcdef123456"]) == [c.MASK]
def test_redact_args_leaves_normal_args():
args = ["--directory", "/path/to/server", "run", "main.py", "-y"]
assert c.redact_args(args) == args
def test_diagnostics_redacts_token_args():
txt = c.diagnostics_text("t", {"command": "npx", "args": ["--token", "s3cret-value"]})
assert "s3cret-value" not in txt
assert c.MASK in txt
# --------------------------------------------------------------------------- #
# 6. Dependency / PATH checking
# --------------------------------------------------------------------------- #
def test_dep_check_finds_python():
# "python3" does not exist on a stock Windows install; "warn" (found only
# on an augmented PATH) still proves resolution works on a real machine.
cmd = "python" if os.name == "nt" else "python3"
assert c.check_dependency({"command": cmd})["status"] in ("ok", "warn")
def test_dep_check_flags_missing():
assert c.check_dependency({"command": "definitely-not-real-binary-xyz"})["status"] == "missing"
def test_dep_check_marks_remote():
assert c.check_dependency({"url": "https://example.com/mcp"})["status"] == "remote"
def test_dep_check_sees_through_shell_wrapper():
r = c.check_dependency({"command": "cmd", "args": ["/c", "definitely-not-real-xyz", "foo"]})
assert "definitely-not-real-xyz" in r["label"]
# --------------------------------------------------------------------------- #
# 7. Spawn test
# --------------------------------------------------------------------------- #
def test_spawn_test_ok_still_running():
# A process that sleeps longer than the timeout → "ok" (server started)
r = c.spawn_test(
{"command": sys.executable, "args": ["-c", "import time; time.sleep(60)"]}, timeout=0.3
)
assert r["outcome"] == "ok"
assert r["returncode"] is None
def test_spawn_test_crashed():
# A process that exits with a non-zero code immediately.
# Timeout is generous (not 0.3s) so a slow CI runner's interpreter startup
# can't outlast it and misclassify the crash as "ok" (still running). See #12.
r = c.spawn_test(
{"command": sys.executable, "args": ["-c", "raise SystemExit(1)"]}, timeout=2.0
)
assert r["outcome"] == "crashed"
assert r["returncode"] == 1
def test_spawn_test_exited_cleanly():
# A process that exits 0 before timeout (unusual for a server).
# Generous timeout so slow-runner startup can't flip this to "ok". See #12.
r = c.spawn_test({"command": sys.executable, "args": ["-c", "pass"]}, timeout=2.0)
assert r["outcome"] == "exited"
assert r["returncode"] == 0
def test_spawn_test_not_found():
r = c.spawn_test({"command": "definitely-not-a-real-binary-xyz"}, timeout=0.3)
assert r["outcome"] == "not_found"
def test_spawn_test_remote_not_applicable():
r = c.spawn_test({"url": "https://example.com/mcp"}, timeout=0.3)
assert r["outcome"] == "not_applicable"
def test_spawn_test_stderr_captured():
# A process that writes to stderr before crashing
r = c.spawn_test(
{
"command": sys.executable,
"args": ["-c", "import sys; sys.stderr.write('test-err\\n'); sys.exit(2)"],
},
timeout=2.0,
)
assert r["outcome"] == "crashed"
assert "test-err" in r["stderr"]
def test_spawn_test_stdin_not_closed():
# A process that blocks reading stdin should stay alive (outcome=ok), not exit 0.
# This guards against the regression where stdin=DEVNULL sends EOF and a server
# that reads stdin exits cleanly, being misclassified as "exited" instead of "ok".
r = c.spawn_test(
{"command": sys.executable, "args": ["-c", "import sys; sys.stdin.read()"]},
timeout=0.3,
)
assert r["outcome"] == "ok", f"expected ok (blocking on stdin), got {r['outcome']}"
def test_spawn_test_large_stderr_does_not_deadlock():
# A process that writes >4 KB (pipe buffer territory) to stderr then exits non-zero.
# Without EOF-draining, the subprocess blocks on a full pipe and is wrongly
# classified as "ok" (still running) instead of "crashed".
script = "import sys; sys.stderr.write('x' * 65536); sys.stderr.flush(); sys.exit(3)"
r = c.spawn_test(
{"command": sys.executable, "args": ["-c", script]},
timeout=2.0,
)
assert r["outcome"] == "crashed", f"expected crashed, got {r['outcome']}"
assert r["returncode"] == 3
# stderr is capped, not the full 64 KB
assert len(r["stderr"]) <= c._STDERR_CAP
def test_windows_tree_kill_uses_taskkill(monkeypatch):
"""The Windows kill path must walk the process tree (taskkill /T /F);
Popen.kill() alone orphans grandchildren (issue #13)."""
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return _FakeRC()
class _FakeRC:
returncode = 0
monkeypatch.setattr(c.subprocess, "run", fake_run)
c._kill_process_tree_windows(1234)
assert calls == [["taskkill", "/PID", "1234", "/T", "/F"]]
@pytest.mark.skipif(os.name != "nt", reason="Windows-only: real process-tree kill")
def test_spawn_test_windows_kills_child_process_tree(tmp_path):
"""A spawn-tested parent that has its own child (npx -> node style) must
not leave the child running after the timeout kill (issue #13)."""
import subprocess as sp
import time as _t
pid_file = tmp_path / "child.pid"
parent_script = (
"import subprocess, sys, time\n"
"p = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n"
f"open({str(pid_file)!r}, 'w').write(str(p.pid))\n"
"time.sleep(60)\n"
)
r = c.spawn_test({"command": sys.executable, "args": ["-c", parent_script]}, timeout=2.0)
assert r["outcome"] == "ok" # parent was still running at timeout, then tree-killed
assert pid_file.is_file(), "parent never spawned its child"
child_pid = int(pid_file.read_text())
deadline = _t.time() + 5.0
alive = True
while _t.time() < deadline:
out = sp.run(
["tasklist", "/FI", f"PID eq {child_pid}"], capture_output=True, text=True
).stdout
alive = str(child_pid) in out
if not alive:
break
_t.sleep(0.25)
assert not alive, f"child pid {child_pid} survived the spawn-test kill"
def test_spawn_test_non_string_env_value():
# Pasted JSON can carry numeric env values ("env": {"PORT": 8080}) that never
# round-trip through the editor. Popen rejects non-str env; spawn_test must
# coerce instead of letting TypeError escape (which would leave the GUI's
# Test launch / Test all buttons stuck disabled).
r = c.spawn_test(
{
"command": sys.executable,
"args": ["-c", "import os; raise SystemExit(0 if os.environ['PORT'] == '8080' else 1)"],
"env": {"PORT": 8080},
},
timeout=2.0,
)
assert r["outcome"] == "exited"
assert r["returncode"] == 0
def test_spawn_test_non_string_command():
# A non-string command must classify, not raise (str(123) resolves to nothing).
r = c.spawn_test({"command": 123}, timeout=0.3)
assert r["outcome"] == "not_found"
def test_spawn_test_internal_error_yields_result(monkeypatch):
# Any unexpected exception inside the spawn path must come back as a result
# dict (outcome "error"), never escape — the UI only re-enables its buttons
# when a result arrives.
def boom(*a, **k):
raise RuntimeError("simulated internal failure")
monkeypatch.setattr(c.shutil, "which", boom)
r = c.spawn_test({"command": "python3"}, timeout=0.3)
assert r["outcome"] == "error"
assert "simulated internal failure" in r["detail"]
# --------------------------------------------------------------------------- #
# 8. Backup restore
# --------------------------------------------------------------------------- #
@pytest.fixture
def config_with_backup(tmp_path):
"""A config file with two backups already present."""
cfgpath = tmp_path / "claude_desktop_config.json"
original = {
"globalShortcut": "Cmd+Shift+Space",
"mcpServers": {"server-v1": {"command": "node", "args": ["v1.js"]}},
}
cfgpath.write_text(json.dumps(original, indent=2))
# Create first backup (v1 state) by writing a new version
cfg = c.load_config(cfgpath)
servers = c.extract_servers(cfg)
servers[0] = c.ServerEntry("server-v2", {"command": "node", "args": ["v2.js"]}, True)
c.apply_servers(cfg, servers)
c.write_config(cfgpath, cfg)
return cfgpath
def test_list_backups_returns_newest_first(config_with_backup):
backups = c.list_backups(config_with_backup)
assert len(backups) >= 1
# Sorted newest-first means descending lexicographic order on timestamps
names = [b.name for b in backups]
assert names == sorted(names, reverse=True)
def test_list_backups_empty_when_no_bdir(tmp_path):
cfgpath = tmp_path / "claude_desktop_config.json"
cfgpath.write_text("{}")
assert c.list_backups(cfgpath) == []
def test_backup_label_parses_timestamp(config_with_backup):
backup = c.list_backups(config_with_backup)[0]
label = c.backup_label(backup)
# label should look like "2026-07-02 00:41:24" (date + time, separated by two spaces)
assert "-" in label and ":" in label
assert len(label) == 20 # "YYYY-MM-DD HH:MM:SS" = 10 + 2 + 8
def test_backup_diff_shows_server_change(config_with_backup):
backup = c.list_backups(config_with_backup)[0]
diff = c.backup_diff(config_with_backup, backup)
# The diff should show the transition back to v1 servers
assert "server-v1" in diff or "server-v2" in diff
def test_backup_diff_no_diff_when_identical(tmp_path):
cfgpath = tmp_path / "cfg.json"
content = {"mcpServers": {"s": {"command": "node"}}}
cfgpath.write_text(json.dumps(content, indent=2))
bdir = cfgpath.parent / c.BACKUP_DIRNAME
bdir.mkdir(exist_ok=True)
import time as _time
stamp = _time.strftime("%Y%m%d-%H%M%S")
backup_path = bdir / f"cfg.{stamp}.json"
backup_path.write_text(json.dumps(content, indent=2))
diff = c.backup_diff(cfgpath, backup_path)
assert "(no differences" in diff
def test_restore_backup_replaces_servers(config_with_backup):
# Get the v1 backup (only backup we have)
backups = c.list_backups(config_with_backup)
assert len(backups) >= 1
v1_backup = backups[0]
# Current file has server-v2; restore should bring back server-v1
c.restore_backup(config_with_backup, v1_backup)
restored = json.loads(config_with_backup.read_text())
assert "server-v1" in restored.get("mcpServers", {})
assert "server-v2" not in restored.get("mcpServers", {})
def test_restore_preserves_non_server_keys(config_with_backup):
backups = c.list_backups(config_with_backup)
c.restore_backup(config_with_backup, backups[0])
restored = json.loads(config_with_backup.read_text())
# Non-server key must survive
assert restored.get("globalShortcut") == "Cmd+Shift+Space"
def test_restore_creates_pre_restore_backup(config_with_backup):
before_count = len(c.list_backups(config_with_backup))
backups = c.list_backups(config_with_backup)
c.restore_backup(config_with_backup, backups[0])
after_count = len(c.list_backups(config_with_backup))
# restore_backup goes through write_config which makes a backup first
assert after_count == before_count + 1
def test_restore_with_current_cfg_passed(config_with_backup):
# When current_cfg is passed in (the GUI's in-memory config), it should be
# used instead of re-reading from disk.
backups = c.list_backups(config_with_backup)
# Simulate the GUI passing a repaired in-memory version
in_memory_cfg = {
"globalShortcut": "Cmd+Shift+Space",
"extraKey": "preserved",
"mcpServers": {"server-v2": {"command": "node", "args": ["v2.js"]}},
}
c.restore_backup(config_with_backup, backups[0], current_cfg=in_memory_cfg)
restored = json.loads(config_with_backup.read_text())
assert restored.get("extraKey") == "preserved"
assert "server-v1" in restored.get("mcpServers", {})
def test_backup_diff_masks_secrets(tmp_path):
cfgpath = tmp_path / "cfg.json"
current = {
"mcpServers": {
"s": {
"command": "node",
"args": ["--token", "super-secret"],
"env": {"API_KEY": "my-api-key-value"},
}
}
}
cfgpath.write_text(json.dumps(current, indent=2))
bdir = cfgpath.parent / c.BACKUP_DIRNAME
bdir.mkdir()
import time as _t
bp = bdir / f"cfg.{_t.strftime('%Y%m%d-%H%M%S')}.json"
bp.write_text(
json.dumps(
{
"mcpServers": {
"s2": {
"command": "python",
"args": ["--api-key", "another-secret"],
"env": {"SECRET_KEY": "backup-secret-value"},
}
}
},
indent=2,
)
)
diff = c.backup_diff(cfgpath, bp)
# Raw secret values must not appear in the diff
assert "super-secret" not in diff
assert "my-api-key-value" not in diff
assert "another-secret" not in diff
assert "backup-secret-value" not in diff
# The mask placeholder must appear (secrets were present in both sides)
assert c.MASK in diff
def test_restore_backup_restores_disabled_servers(tmp_path):
cfgpath = tmp_path / "cfg.json"
original = {
"mcpServers": {"active": {"command": "node"}},
c.DISABLED_KEY: {"parked": {"command": "python"}},
}
cfgpath.write_text(json.dumps(original, indent=2))
# Overwrite: add new server, drop the disabled one
cfg = c.load_config(cfgpath)
servers = c.extract_servers(cfg)
servers = [s for s in servers if s.name != "parked"]
servers.append(c.ServerEntry("newcomer", {"command": "uvx"}, True))
c.apply_servers(cfg, servers)
c.write_config(cfgpath, cfg)
# Restore from the v1 backup
backups = c.list_backups(cfgpath)
c.restore_backup(cfgpath, backups[0])
restored = json.loads(cfgpath.read_text())
assert "parked" in restored.get(c.DISABLED_KEY, {})
assert "active" in restored.get("mcpServers", {})
assert "newcomer" not in restored.get("mcpServers", {})
# --------------------------------------------------------------------------- #
# 9 — Stale-file protection
# --------------------------------------------------------------------------- #
def test_config_mtime_returns_float_for_existing_file(tmp_path):
f = tmp_path / "cfg.json"
f.write_text("{}")
assert isinstance(c.config_mtime(f), float)
def test_config_mtime_returns_none_for_missing_file(tmp_path):
assert c.config_mtime(tmp_path / "nonexistent.json") is None
def test_config_fingerprint_returns_mtime_and_size_for_existing_file(tmp_path):
f = tmp_path / "cfg.json"
f.write_text("{}")
fp = c.config_fingerprint(f)
assert isinstance(fp, c.ConfigStat)
assert isinstance(fp.mtime, float)
assert fp.size == f.stat().st_size
def test_config_fingerprint_returns_none_for_missing_file(tmp_path):
assert c.config_fingerprint(tmp_path / "nonexistent.json") is None
def test_config_fingerprint_detects_size_change_when_mtime_is_unchanged(tmp_path):
"""
Guards against the exact hole bare-mtime comparison has: an external write
that lands within the filesystem's mtime resolution (or that restores the
original mtime) must still be caught, because the file's size differs.
"""
f = tmp_path / "cfg.json"
f.write_text('{"mcpServers": {}}')
original = c.config_fingerprint(f)
original_mtime_ns = f.stat().st_mtime_ns
# Overwrite with materially different (larger) content, then force the
# mtime back to its original value -- simulating a same-second external
# write, or a writer that preserves mtime.
f.write_text('{"mcpServers": {"new-server": {"command": "node", "args": ["a", "b", "c"]}}}')
os.utime(f, ns=(original_mtime_ns, original_mtime_ns))
updated = c.config_fingerprint(f)
assert updated.mtime == original.mtime # mtime alone would say "unchanged"
assert updated.size != original.size # size catches what mtime missed
assert updated != original # the fingerprint as a whole detects the change
def test_external_change_summary_detects_non_server_key_change(tmp_path):
cfgpath = tmp_path / "cfg.json"
original = {"numStartups": 1, "mcpServers": {"s": {"command": "node"}}}
cfgpath.write_text(json.dumps({"numStartups": 2, "mcpServers": {"s": {"command": "node"}}}))
changed_keys, server_diff = c.external_change_summary(original, cfgpath)
assert "numStartups" in changed_keys
assert server_diff == "" # server sections unchanged
def test_external_change_summary_detects_server_section_change(tmp_path):
cfgpath = tmp_path / "cfg.json"
original = {"mcpServers": {"s1": {"command": "node"}}}
cfgpath.write_text(json.dumps({"mcpServers": {"s2": {"command": "python"}}}))
changed_keys, server_diff = c.external_change_summary(original, cfgpath)
assert "mcpServers" in changed_keys
assert "s1" in server_diff or "s2" in server_diff
def test_external_change_summary_server_diff_masks_secrets(tmp_path):
cfgpath = tmp_path / "cfg.json"
original = {"mcpServers": {"a": {"command": "node", "env": {"SECRET_KEY": "s3cr3t"}}}}
# command changes (visible in diff) AND secret value changes (both sides masked →
# MASK appears in context lines, raw secret never appears)
cfgpath.write_text(
json.dumps(
{"mcpServers": {"a": {"command": "python", "env": {"SECRET_KEY": "n3w_s3cr3t"}}}}
)
)
changed_keys, server_diff = c.external_change_summary(original, cfgpath)
assert "mcpServers" in changed_keys
assert "s3cr3t" not in server_diff
assert "n3w_s3cr3t" not in server_diff
assert c.MASK in server_diff # appears in context for the masked env value
def test_external_change_summary_empty_when_unchanged(tmp_path):
cfgpath = tmp_path / "cfg.json"
cfg = {"mcpServers": {"s": {"command": "node"}}}
cfgpath.write_text(json.dumps(cfg))
changed_keys, server_diff = c.external_change_summary(cfg, cfgpath)
assert changed_keys == []
assert server_diff == ""
def test_merge_and_reapply_preserves_both_edits(tmp_path):
"""AC test: external non-server change + in-memory server edit both survive merge."""
cfgpath = tmp_path / "cfg.json"
original = {"numStartups": 1, "mcpServers": {"s1": {"command": "node"}}}
cfgpath.write_text(json.dumps(original, indent=2))
user_servers = [c.ServerEntry("s2", {"command": "python"}, True)]
# External process bumps numStartups without touching servers
disk_changed = dict(original)
disk_changed["numStartups"] = 42
cfgpath.write_text(json.dumps(disk_changed, indent=2))
# Merge: load fresh from disk, apply user's server edits, write back
fresh = c.load_config(cfgpath)
c.apply_servers(fresh, user_servers)
c.write_config(cfgpath, fresh)
result = json.loads(cfgpath.read_text())
assert result["numStartups"] == 42 # external change preserved
assert "s2" in result["mcpServers"] # user's server edit preserved
assert "s1" not in result["mcpServers"] # old server replaced
# --------------------------------------------------------------------------- #
# 10 — Secret-in-args warning
# --------------------------------------------------------------------------- #
def test_args_secret_warning_embedded_url_creds():
"""AC case: postgres://user:pass@host triggers a warning."""
data = {"args": ["--connection", "postgres://user:pass@host/db"]}
assert c.args_secret_warning(data) is not None
def test_args_secret_warning_token_prefix_positional():
"""A bare ghp_… token as a positional arg triggers a warning."""
data = {"args": ["ghp_abc123def456"]}
assert c.args_secret_warning(data) is not None
def test_args_secret_warning_secret_flag_value_pair():
"""Value following a secret-named flag triggers a warning."""
data = {"args": ["--token", "supersecret"]}
assert c.args_secret_warning(data) is not None
def test_args_secret_warning_benign_url_no_password():
"""https:// URL without user:pass should NOT warn."""
data = {"args": ["--endpoint", "https://mcp.example.com/sse"]}
assert c.args_secret_warning(data) is None
def test_args_secret_warning_user_at_host_no_password():
"""ssh://user@host without a password should NOT warn."""
data = {"args": ["ssh://git@github.com"]}
assert c.args_secret_warning(data) is None
def test_args_secret_warning_inline_flag_value_skipped():
"""--flag=value inline pairs are self-documenting and should NOT warn."""
data = {"args": ["--api-key=sk-abc123"]}
assert c.args_secret_warning(data) is None
def test_args_secret_warning_env_not_triggered():
"""Secrets in env (not args) must not trigger args warning, even with benign args present."""
data = {"args": ["--verbose"], "env": {"DATABASE_URL": "postgres://user:pass@host/db"}}
assert c.args_secret_warning(data) is None
def test_args_secret_warning_empty():
assert c.args_secret_warning({}) is None
assert c.args_secret_warning({"args": []}) is None
# --------------------------------------------------------------------------- #
# Named server sets (issue #52)
# --------------------------------------------------------------------------- #
def _three_servers():
return [
c.ServerEntry("alpha", {"command": "npx"}, True),
c.ServerEntry("beta", {"command": "uvx"}, True),
c.ServerEntry("gamma", {"url": "https://x.example/mcp"}, False),
]
def test_save_and_list_server_sets_roundtrip():
cfg: dict = {}
members = c.save_server_set(cfg, "webdev", _three_servers())
assert members == ["alpha", "beta"] # only the enabled ones
assert c.list_server_sets(cfg) == {"webdev": ["alpha", "beta"]}
def test_apply_server_set_enables_exactly_the_members():
servers = _three_servers()
missing = c.apply_server_set(servers, ["gamma"])
assert missing == []
assert [s.enabled for s in servers] == [False, False, True]
def test_apply_server_set_reports_missing_members():
servers = _three_servers()
missing = c.apply_server_set(servers, ["alpha", "vanished", "gone"])
assert missing == ["gone", "vanished"]
assert [s.enabled for s in servers] == [True, False, False] # rest still applied
def test_delete_server_set_drops_empty_key():
cfg: dict = {}
c.save_server_set(cfg, "only", _three_servers())
assert c.delete_server_set(cfg, "only") is True
assert c.SETS_KEY not in cfg # no empty bcc key left behind
assert c.delete_server_set(cfg, "only") is False
def test_server_sets_survive_write_and_reload(tmp_path):
cfgpath = tmp_path / "cfg.json"
cfg = {"mcpServers": {"alpha": {"command": "npx"}}}
c.save_server_set(cfg, "webdev", [c.ServerEntry("alpha", {"command": "npx"}, True)])
c.write_config(cfgpath, cfg)
reloaded = c.load_config(cfgpath)
assert c.list_server_sets(reloaded) == {"webdev": ["alpha"]}
def test_apply_servers_preserves_sets_key():
cfg: dict = {"mcpServers": {}}
c.save_server_set(cfg, "s", [c.ServerEntry("alpha", {"command": "npx"}, True)])
c.apply_servers(cfg, _three_servers())
assert c.SETS_KEY in cfg # the server writer never touches sets
def test_list_server_sets_skips_malformed_entries():
cfg = {c.SETS_KEY: {"good": ["a"], "bad-not-list": "a", "bad-items": ["a", 3]}}
assert c.list_server_sets(cfg) == {"good": ["a"]}
assert c.list_server_sets({c.SETS_KEY: "junk"}) == {}
# --------------------------------------------------------------------------- #
# resolve_name_collision (paste/import duplicate-name handling — issue #8)
# --------------------------------------------------------------------------- #
def test_resolve_name_collision_no_conflict_returns_unchanged():
assert c.resolve_name_collision("brave-search", {"other"}) == "brave-search"
def test_resolve_name_collision_single_conflict_appends_dash_two():
assert c.resolve_name_collision("brave-search", {"brave-search"}) == "brave-search-2"
def test_resolve_name_collision_skips_taken_suffixes():
existing = {"brave-search", "brave-search-2", "brave-search-3"}
assert c.resolve_name_collision("brave-search", existing) == "brave-search-4"
def test_resolve_name_collision_empty_existing_set():
assert c.resolve_name_collision("brave-search", set()) == "brave-search"
def test_resolve_name_collision_repeated_calls_stay_unique():
"""Simulates dropping the same file twice in a row: each resolved name
must be fed back into `existing` before resolving the next one, or two
servers could end up sharing a name (and silently collapse into one
when apply_servers() rebuilds its dict at save time)."""
existing = {"brave-search"}
first = c.resolve_name_collision("brave-search", existing)
existing.add(first)
second = c.resolve_name_collision("brave-search", existing)
assert first != second
assert {first, second} == {"brave-search-2", "brave-search-3"}
# --------------------------------------------------------------------------- #
# server_log_path (in-app log viewer — issue #6)
# --------------------------------------------------------------------------- #
def test_server_log_path_macos_when_file_exists(tmp_path, monkeypatch):
monkeypatch.setattr(sys, "platform", "darwin")
monkeypatch.setattr(c.Path, "home", lambda: tmp_path)
log_dir = tmp_path / "Library" / "Logs" / "Claude"
log_dir.mkdir(parents=True)
(log_dir / "mcp-server-brave-search.log").write_text("hello")
result = c.server_log_path("brave-search")
assert result == log_dir / "mcp-server-brave-search.log"
def test_server_log_path_macos_none_when_missing(tmp_path, monkeypatch):
monkeypatch.setattr(sys, "platform", "darwin")
monkeypatch.setattr(c.Path, "home", lambda: tmp_path)
assert c.server_log_path("brave-search") is None
def test_server_log_path_windows_when_file_exists(tmp_path, monkeypatch):
# server_log_path branches on sys.platform only (never os.name), so the
# test doesn't have to touch os.name -- mutating that globally mid-test
# would also flip which concrete Path subclass pathlib hands out.
monkeypatch.setattr(sys, "platform", "win32")
monkeypatch.setenv("APPDATA", str(tmp_path))
log_dir = tmp_path / "Claude" / "logs"
log_dir.mkdir(parents=True)
(log_dir / "mcp.log").write_text("hello")
result = c.server_log_path("brave-search")
assert result == log_dir / "mcp.log"
def test_server_log_path_windows_none_when_missing(tmp_path, monkeypatch):
monkeypatch.setattr(sys, "platform", "win32")
monkeypatch.setenv("APPDATA", str(tmp_path))
assert c.server_log_path("brave-search") is None
def test_server_log_path_unsupported_platform_returns_none(tmp_path, monkeypatch):
monkeypatch.setattr(sys, "platform", "linux")
monkeypatch.setattr(c.Path, "home", lambda: tmp_path)
assert c.server_log_path("brave-search") is None
# --------------------------------------------------------------------------- #
# MSIX-virtualized Claude Desktop config detection (issue #7)
# --------------------------------------------------------------------------- #
def _make_msix_pkg(localappdata, pkg_name="AnthropicClaude_abc123xyz", with_config=True):
cfg_dir = localappdata / "Packages" / pkg_name / "LocalCache" / "Roaming" / "Claude"
cfg_dir.mkdir(parents=True)
cfg_path = cfg_dir / c.CONFIG_FILENAME
if with_config:
cfg_path.write_text('{"mcpServers": {}}')
return cfg_path
def test_msix_config_paths_finds_package_config(tmp_path):
local = tmp_path / "Local"
expected = _make_msix_pkg(local)
assert c.msix_config_paths(local) == [expected]
def test_msix_config_paths_ignores_non_claude_packages(tmp_path):
local = tmp_path / "Local"
(local / "Packages" / "SomeOtherApp_xyz" / "LocalCache" / "Roaming").mkdir(parents=True)
assert c.msix_config_paths(local) == []
def test_msix_config_paths_empty_when_package_dir_has_no_config_yet(tmp_path):
local = tmp_path / "Local"
_make_msix_pkg(local, with_config=False)
assert c.msix_config_paths(local) == []
def test_msix_config_paths_empty_when_no_packages_dir(tmp_path):
assert c.msix_config_paths(tmp_path / "Local") == []
def test_msix_config_paths_uses_localappdata_env_by_default(tmp_path, monkeypatch):
local = tmp_path / "Local"
expected = _make_msix_pkg(local)
monkeypatch.setenv("LOCALAPPDATA", str(local))
assert c.msix_config_paths() == [expected]
def test_detect_msix_claude_returns_none_on_non_windows(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "darwin")
local = tmp_path / "Local"
_make_msix_pkg(local)
assert c.detect_msix_claude(localappdata=local) is None
def test_detect_msix_claude_finds_virtualized_config_on_windows(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "win32")
local = tmp_path / "Local"
expected = _make_msix_pkg(local)
assert c.detect_msix_claude(localappdata=local) == expected
def test_detect_msix_claude_none_when_nothing_present_on_windows(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "win32")
local = tmp_path / "Local"
assert c.detect_msix_claude(localappdata=local) is None
def test_msix_warning_text_none_on_non_windows(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "darwin")
local = tmp_path / "Local"
_make_msix_pkg(local)
assert c.msix_warning_text(localappdata=local) is None
def test_msix_warning_text_none_when_nothing_detected(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "win32")
local = tmp_path / "Local"
assert c.msix_warning_text(localappdata=local) is None
def test_msix_warning_text_warns_when_virtualized_path_differs(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "win32")
roaming = tmp_path / "Roaming"
local = tmp_path / "Local"
real = _make_msix_pkg(local)
warning = c.msix_warning_text(appdata=roaming, localappdata=local)
assert warning is not None
assert "Microsoft Store" in warning
assert str(real) in warning
assert str(roaming / "Claude" / c.CONFIG_FILENAME) in warning
def test_msix_warning_text_none_when_plain_and_virtualized_paths_match(tmp_path, monkeypatch):
# Degenerate case: if the "plain" APPDATA base is pointed at the exact
# same tree the MSIX scan found (e.g. a symlink setup), there's nothing
# surprising to warn about.
monkeypatch.setattr(c.sys, "platform", "win32")
local = tmp_path / "Local"
pkg_claude_dir = local / "Packages" / "AnthropicClaude_abc123xyz" / "LocalCache" / "Roaming"
pkg_claude_dir.mkdir(parents=True)
(pkg_claude_dir / "Claude").mkdir()
(pkg_claude_dir / "Claude" / c.CONFIG_FILENAME).write_text("{}")
warning = c.msix_warning_text(appdata=pkg_claude_dir, localappdata=local)
assert warning is None
def test_discover_profiles_adds_msix_profile_on_windows(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "win32")
base = tmp_path / "AppSupport"
base.mkdir()
monkeypatch.setattr(c, "app_support_base", lambda: base)
monkeypatch.setattr(c.Path, "home", lambda: tmp_path / "home")
(tmp_path / "home" / ".claude").mkdir(parents=True)
local = tmp_path / "Local"
real = _make_msix_pkg(local)
monkeypatch.setenv("LOCALAPPDATA", str(local))
profs = c.discover_profiles()
msix = [p for p in profs if "MSIX" in p.label]
assert len(msix) == 1
assert msix[0].path == real
assert msix[0].config_exists
def test_discover_profiles_no_msix_profile_when_nothing_detected(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "win32")
base = tmp_path / "AppSupport"
base.mkdir()
monkeypatch.setattr(c, "app_support_base", lambda: base)
monkeypatch.setattr(c.Path, "home", lambda: tmp_path / "home")
(tmp_path / "home" / ".claude").mkdir(parents=True)
monkeypatch.setenv("LOCALAPPDATA", str(tmp_path / "Local"))
profs = c.discover_profiles()
assert not any("MSIX" in p.label for p in profs)
def test_discover_profiles_no_msix_profile_on_non_windows(tmp_path, monkeypatch):
monkeypatch.setattr(c.sys, "platform", "darwin")
base = tmp_path / "AppSupport"
base.mkdir()
monkeypatch.setattr(c, "app_support_base", lambda: base)
monkeypatch.setattr(c.Path, "home", lambda: tmp_path / "home")
(tmp_path / "home" / ".claude").mkdir(parents=True)
local = tmp_path / "Local"
_make_msix_pkg(local)
monkeypatch.setenv("LOCALAPPDATA", str(local))
profs = c.discover_profiles()
assert not any("MSIX" in p.label for p in profs)
# --------------------------------------------------------------------------- #
# restart_claude_desktop / profile_targets_claude_desktop (issue #9)
# --------------------------------------------------------------------------- #
def test_profile_targets_claude_desktop_true_for_desktop_config():
p = c.Profile(label="Claude", path="/x/Claude/claude_desktop_config.json", config_exists=True)
assert c.profile_targets_claude_desktop(p)
def test_profile_targets_claude_desktop_false_for_claude_code():
p = c.Profile(label="Claude Code", path="/home/me/.claude.json", config_exists=True)
assert not c.profile_targets_claude_desktop(p)
def test_profile_targets_claude_desktop_false_for_legacy_settings():
p = c.Profile(
label="Claude Code (legacy settings.json)",
path="/home/me/.claude/settings.json",
config_exists=True,
)
assert not c.profile_targets_claude_desktop(p)
class _FakeCompletedProcess:
def __init__(self, returncode=0, stdout="", stderr=""):
self.returncode = returncode
self.stdout = stdout
self.stderr = stderr
def test_restart_claude_desktop_macos_commands(monkeypatch):
"""macOS: pkill -x "Claude", wait for exit (pgrep), then open -a Claude."""
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
if cmd[0] == "pgrep":
return _FakeCompletedProcess(returncode=1) # already exited
return _FakeCompletedProcess(returncode=0)
monkeypatch.setattr(c.sys, "platform", "darwin")
monkeypatch.setattr(c.subprocess, "run", fake_run)
result = c.restart_claude_desktop()
assert result.success
assert calls[0] == ["pkill", "-x", "Claude"]
assert calls[-1] == ["open", "-a", "Claude"]
def test_restart_claude_desktop_macos_pkill_not_running_is_fine(monkeypatch):
"""pkill exiting non-zero (nothing to kill) must NOT be treated as failure."""
def fake_run(cmd, **kwargs):
if cmd[0] in ("pkill", "pgrep"):
return _FakeCompletedProcess(returncode=1) # no matching process
return _FakeCompletedProcess(returncode=0)
monkeypatch.setattr(c.sys, "platform", "darwin")
monkeypatch.setattr(c.subprocess, "run", fake_run)
result = c.restart_claude_desktop()
assert result.success
def test_restart_claude_desktop_macos_relaunch_failure_reported(monkeypatch):
def fake_run(cmd, **kwargs):
if cmd[0] == "open":
return _FakeCompletedProcess(returncode=1, stderr="Unable to find application")
return _FakeCompletedProcess(returncode=1)
monkeypatch.setattr(c.sys, "platform", "darwin")
monkeypatch.setattr(c.subprocess, "run", fake_run)
result = c.restart_claude_desktop()
assert not result.success
assert "Unable to find application" in result.detail
def test_restart_claude_desktop_macos_pkill_binary_missing_does_not_raise(monkeypatch):
"""pkill/pgrep raising OSError (binary missing) must be swallowed, not
propagated -- relaunch is still attempted."""
def fake_run(cmd, **kwargs):
if cmd[0] in ("pkill", "pgrep"):
raise OSError("binary not found")
return _FakeCompletedProcess(returncode=0)
monkeypatch.setattr(c.sys, "platform", "darwin")
monkeypatch.setattr(c.subprocess, "run", fake_run)
result = c.restart_claude_desktop()
assert result.success
def test_restart_claude_desktop_macos_gives_up_if_app_wont_quit(monkeypatch):
monkeypatch.setattr(c.sys, "platform", "darwin")
monkeypatch.setattr(c, "_run_quiet", lambda cmd: None)
monkeypatch.setattr(c, "_macos_claude_running", lambda: True) # never exits
monkeypatch.setattr(c, "_MACOS_QUIT_WAIT_S", 0.2)
res = c.restart_claude_desktop()
assert res.success is False
assert "quit" in res.detail.lower()
def test_restart_claude_desktop_macos_waits_for_exit_then_relaunches(monkeypatch):
monkeypatch.setattr(c.sys, "platform", "darwin")
monkeypatch.setattr(c, "_run_quiet", lambda cmd: None)
alive = iter([True, True, False]) # exits on the third poll
monkeypatch.setattr(c, "_macos_claude_running", lambda: next(alive))
launched = []
def fake_run(cmd, **kwargs):
launched.append(cmd)
return _FakeCompletedProcess(returncode=0)
monkeypatch.setattr(c.subprocess, "run", fake_run)
res = c.restart_claude_desktop()
assert res.success is True
assert launched == [["open", "-a", "Claude"]]
@pytest.fixture
def windows_shortcut(tmp_path, monkeypatch):
"""A fake %APPDATA% containing the Claude Start-menu shortcut."""
monkeypatch.setenv("APPDATA", str(tmp_path))
lnk = tmp_path / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Claude.lnk"
lnk.parent.mkdir(parents=True)
lnk.write_bytes(b"")
return lnk
def test_restart_claude_desktop_windows_commands(monkeypatch, windows_shortcut):
"""Windows: taskkill the process, then relaunch via the Start-menu shortcut."""
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return _FakeCompletedProcess(returncode=0)
monkeypatch.setattr(c.sys, "platform", "win32")
monkeypatch.setattr(c.subprocess, "run", fake_run)
result = c.restart_claude_desktop()
assert result.success
assert calls[0] == ["taskkill", "/IM", "Claude.exe", "/F"]
assert calls[1][:3] == ["cmd", "/c", "start"]
assert calls[1][-1].endswith("Claude.lnk")
def test_restart_claude_desktop_windows_relaunch_failure_reported(monkeypatch, windows_shortcut):
def fake_run(cmd, **kwargs):
if cmd[0] == "cmd":
return _FakeCompletedProcess(returncode=1, stderr="not found")
return _FakeCompletedProcess(returncode=0)
monkeypatch.setattr(c.sys, "platform", "win32")
monkeypatch.setattr(c.subprocess, "run", fake_run)
result = c.restart_claude_desktop()
assert not result.success
assert "not found" in result.detail
def test_restart_claude_desktop_windows_missing_shortcut_refuses_before_killing(
monkeypatch, tmp_path
):
killed = []
monkeypatch.setattr(c, "_run_quiet", lambda cmd: killed.append(cmd))
monkeypatch.setattr(c.sys, "platform", "win32")
monkeypatch.setenv("APPDATA", str(tmp_path)) # no Claude.lnk under here (MSIX case)
res = c.restart_claude_desktop()
assert res.success is False
assert "shortcut" in res.detail.lower()
assert killed == [] # Claude must NOT be killed when it can't be relaunched
def test_restart_supported_only_on_desktop_platforms(monkeypatch):
monkeypatch.setattr(c.sys, "platform", "darwin")
assert c.restart_supported()
monkeypatch.setattr(c.sys, "platform", "win32")
assert c.restart_supported()
monkeypatch.setattr(c.sys, "platform", "linux")
assert not c.restart_supported()
def test_restart_claude_desktop_linux_refuses_without_touching_processes(monkeypatch):
"""There is no Claude Desktop on Linux; 'pkill claude' would substring-match
running Claude Code CLI sessions. The restart must refuse outright."""
killed = []
monkeypatch.setattr(c, "_run_quiet", lambda cmd: killed.append(cmd))
monkeypatch.setattr(c.sys, "platform", "linux")
res = c.restart_claude_desktop()
assert res.success is False
assert killed == [] # nothing killed, nothing spawned
# --------------------------------------------------------------------------- #
# Version / update checking
# --------------------------------------------------------------------------- #
def test_dunder_version_matches_pyproject():
# Guards against the version drifting out of sync between the two places
# a human might bump it. Reads pyproject.toml rather than hard-coding a
# version here (which would make this a third place to bump).
text = (Path(__file__).parent.parent / "pyproject.toml").read_text(encoding="utf-8")
m = re.search(r'^version\s*=\s*"([^"]+)"', text, re.MULTILINE)
assert m, "no [project] version found in pyproject.toml"
assert c.__version__ == m.group(1)
def test_parse_version_basic():
assert c.parse_version("1.2.3") == (1, 2, 3)
def test_parse_version_strips_v_prefix():
assert c.parse_version("v1.2.3") == (1, 2, 3)
def test_parse_version_strips_prerelease_suffix():
assert c.parse_version("1.2.3-beta.1") == (1, 2, 3)
assert c.parse_version("1.2.3+build5") == (1, 2, 3)
def test_parse_version_malformed_returns_empty_tuple():
assert c.parse_version("not-a-version") == ()
assert c.parse_version("") == ()
assert c.parse_version(None) == ()
def test_parse_version_stops_at_first_non_numeric_component():
assert c.parse_version("1.2.rc1") == (1, 2)
def test_is_newer_version_true_when_candidate_is_newer():
assert c.is_newer_version("1.1.0", "1.2.0") is True
def test_is_newer_version_false_when_candidate_is_older():
assert c.is_newer_version("1.2.0", "1.1.0") is False
def test_is_newer_version_false_when_equal():
assert c.is_newer_version("1.1.0", "1.1.0") is False
def test_is_newer_version_equal_with_v_prefix():
assert c.is_newer_version("1.1.0", "v1.1.0") is False
def test_is_newer_version_numeric_not_lexical():
# A lexical/string compare would say "10.0.0" < "2.0.0" ("1" < "2").
assert c.is_newer_version("2.0.0", "10.0.0") is True
assert c.is_newer_version("10.0.0", "2.0.0") is False
def test_is_newer_version_handles_differing_length():
assert c.is_newer_version("1.2", "1.2.0") is False
assert c.is_newer_version("1.2.0", "1.2") is False
assert c.is_newer_version("1.2", "1.3") is True
def test_is_newer_version_malformed_candidate_is_never_newer():
assert c.is_newer_version("1.1.0", "not-a-version") is False
assert c.is_newer_version("1.1.0", "") is False
def test_is_newer_version_malformed_current_does_not_crash():
# Shouldn't raise; a valid candidate is reported as newer than "unknown".
assert c.is_newer_version("garbage", "1.0.0") is True
assert c.is_newer_version("garbage", "not-a-version-either") is False
class _FakeHTTPResponse:
def __init__(self, data: bytes):
self._data = data
def read(self):
return self._data
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def test_fetch_latest_release_ok(monkeypatch):
payload = json.dumps(
{
"tag_name": "v1.2.0",
"html_url": "https://git.avezzano.io/the_og/better-claude-config/releases/tag/v1.2.0",
}
).encode("utf-8")
monkeypatch.setattr(
urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(payload)
)
result = c.fetch_latest_release()
assert result == {
"version": "v1.2.0",
"url": "https://git.avezzano.io/the_og/better-claude-config/releases/tag/v1.2.0",
}
def test_fetch_latest_release_falls_back_to_releases_url_when_no_html_url(monkeypatch):
payload = json.dumps({"tag_name": "v1.2.0"}).encode("utf-8")
monkeypatch.setattr(
urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(payload)
)
result = c.fetch_latest_release()
assert result == {"version": "v1.2.0", "url": c.RELEASES_URL}
def test_fetch_latest_release_network_failure_returns_none(monkeypatch):
def boom(req, timeout=None):
raise urllib.error.URLError("no network")
monkeypatch.setattr(urllib.request, "urlopen", boom)
assert c.fetch_latest_release() is None
def test_fetch_latest_release_timeout_returns_none(monkeypatch):
def boom(req, timeout=None):
raise TimeoutError("timed out")
monkeypatch.setattr(urllib.request, "urlopen", boom)
assert c.fetch_latest_release() is None
def test_fetch_latest_release_missing_tag_returns_none(monkeypatch):
payload = json.dumps({"html_url": "https://example.com"}).encode("utf-8")
monkeypatch.setattr(
urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(payload)
)
assert c.fetch_latest_release() is None
def test_fetch_latest_release_malformed_json_returns_none(monkeypatch):
monkeypatch.setattr(
urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(b"not json")
)
assert c.fetch_latest_release() is None
# --------------------------------------------------------------------------- #
# Server search / filter (#27)
# --------------------------------------------------------------------------- #
def test_filter_matches_name():
e = c.ServerEntry("brave-search", {"command": "npx", "args": ["-y", "@x/brave"]}, True)
assert c.server_matches_filter(e, "brave")
def test_filter_matches_stdio_command():
e = c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True)
assert c.server_matches_filter(e, "uvx")
def test_filter_matches_remote_url():
e = c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True)
assert c.server_matches_filter(e, "example.com")
def test_filter_is_case_insensitive():
e = c.ServerEntry("BraveSearch", {"command": "NPX", "args": []}, True)
assert c.server_matches_filter(e, "bravesearch")
assert c.server_matches_filter(e, "npx")
def test_filter_empty_query_matches_everything():
e = c.ServerEntry("anything", {"command": "node", "args": []}, True)
assert c.server_matches_filter(e, "")
assert c.server_matches_filter(e, " ")
def test_filter_no_match_returns_false():
e = c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True)
assert not c.server_matches_filter(e, "nonexistent")
def test_filter_remote_query_does_not_match_stdio_command_field():
e = c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True)
assert not c.server_matches_filter(e, "npx")
def test_filter_servers_returns_only_matches():
servers = [
c.ServerEntry("brave-search", {"command": "npx", "args": []}, True),
c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True),
c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True),
]
assert [s.name for s in c.filter_servers(servers, "brave")] == ["brave-search"]
assert [s.name for s in c.filter_servers(servers, "")] == [s.name for s in servers]
assert c.filter_servers(servers, "zzz-nope") == []
# --------------------------------------------------------------------------- #
# Health status mapping (#28)
# --------------------------------------------------------------------------- #
def test_health_from_spawn_result_ok():
result = {
"outcome": "ok",
"returncode": None,
"stderr": "",
"detail": "still running after 3s — server started successfully",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.OK
assert "started" in summary
def test_health_from_spawn_result_not_applicable_is_untested():
result = {
"outcome": "not_applicable",
"returncode": None,
"stderr": "",
"detail": "remote server",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.UNTESTED
assert summary == "remote server"
def test_health_from_spawn_result_crashed_is_failed():
result = {
"outcome": "crashed",
"returncode": 1,
"stderr": "Traceback: boom\nsecond line",
"detail": "process exited with code 1",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.FAILED
assert "process exited with code 1" in summary
assert "Traceback: boom" in summary
def test_health_from_spawn_result_exited_is_failed():
result = {
"outcome": "exited",
"returncode": 0,
"stderr": "",
"detail": "process exited cleanly (code 0) — unusual",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.FAILED
assert "exited cleanly" in summary
def test_health_from_spawn_result_not_found_is_failed():
result = {
"outcome": "not_found",
"returncode": None,
"stderr": "",
"detail": "command not found: totallybogus",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.FAILED
assert "not found" in summary
def test_health_from_spawn_result_failed_without_stderr_has_no_dash():
result = {
"outcome": "not_found",
"returncode": None,
"stderr": "",
"detail": "command not found: x",
}
_, summary = c.health_from_spawn_result(result)
assert "—" not in summary
# --------------------------------------------------------------------------- #
# App icon assets present (issue #20 — icons must exist to be bundled/loaded)
# --------------------------------------------------------------------------- #
def test_app_icon_assets_present():
from pathlib import Path
root = Path(c.__file__).resolve().parent
rounded = root / "icons" / "twin-gears" / "rounded"
# The window-icon builder in bcc.py loads these sizes; keep them present.
for size in (16, 32, 128, 256):
assert (rounded / f"icon-{size}.png").is_file(), f"missing icon-{size}.png"
assert (root / "icons" / "app.ico").is_file()
assert (root / "icons" / "app.icns").is_file()
# --------------------------------------------------------------------------- #
# MCP server catalog (issue #10 / #61)
# --------------------------------------------------------------------------- #
def _minimal_catalog(version: int = 1) -> dict:
return {
"schema": 1,
"version": version,
"updated": "2026-07-12",
"servers": [
{
"id": "widget",
"display": "Widget",
"description": "A test widget server.",
"category": "dev",
"homepage": "https://example.com/widget",
"stars": 10,
"official": True,
"setup": "basic",
"config": {
# Pinned on purpose (issue #68 finding 3): an earlier
# version of this fixture used an unpinned package and
# asserted it validated clean, which enshrined the bug
# instead of catching it.
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
},
"placeholders": {},
"env_required": {},
"docs_url": "https://example.com/widget/docs",
"notes": "",
}
],
}
def _catalog_with(server_overrides: dict) -> dict:
data = _minimal_catalog()
data["servers"][0].update(server_overrides)
return data
def _sign(raw: bytes, priv: Ed25519PrivateKey) -> bytes:
# Independent of bcc_core's domain-separation constant on purpose: this
# is the literal wire format the design calls for, hardcoded here so a
# change to the constant would be caught as a real behaviour change.
return priv.sign(b"bcc-catalog-v1|" + raw)
def _signed(data: dict, priv: Ed25519PrivateKey) -> tuple[bytes, bytes]:
raw = json.dumps(data).encode("utf-8")
return raw, _sign(raw, priv)
# --- load_catalog / validate_catalog: valid round trip ------------------- #
def test_load_catalog_valid_round_trip():
data = _minimal_catalog()
raw = json.dumps(data).encode("utf-8")
loaded = c.load_catalog(raw)
assert loaded == data
assert c.validate_catalog(loaded) == []
assert c.catalog_version(loaded) == 1
def test_load_catalog_accepts_str_too():
data = _minimal_catalog()
text = json.dumps(data)
assert c.load_catalog(text) == data
def test_load_catalog_malformed_raises_json_decode_error():
# load_catalog is strict json.loads ONLY -- it must never silently
# "repair" malformed bytes into something that parses.
with pytest.raises(json.JSONDecodeError):
c.load_catalog(b"{not valid json")
def test_shipped_catalog_json_passes_validation():
"""Regression test: the real data/catalog.json bundled with the app."""
root = Path(c.__file__).resolve().parent
raw = (root / "data" / "catalog.json").read_bytes()
data = c.load_catalog(raw)
problems = c.validate_catalog(data)
assert problems == [], problems
assert c.catalog_version(data) >= 1
# --- verify_catalog_signature --------------------------------------------- #
def test_verify_catalog_signature_valid():
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
raw = json.dumps(_minimal_catalog()).encode("utf-8")
sig = _sign(raw, priv)
assert c.verify_catalog_signature(raw, sig, [pub]) is True
def test_verify_catalog_signature_tampered_byte_fails():
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
raw = json.dumps(_minimal_catalog()).encode("utf-8")
sig = _sign(raw, priv)
tampered = bytearray(raw)
tampered[0] ^= 0xFF # flip exactly one byte
assert c.verify_catalog_signature(bytes(tampered), sig, [pub]) is False
def test_verify_catalog_signature_wrong_key_fails():
priv = Ed25519PrivateKey.generate()
other_pub = Ed25519PrivateKey.generate().public_key().public_bytes_raw()
raw = json.dumps(_minimal_catalog()).encode("utf-8")
sig = _sign(raw, priv)
assert c.verify_catalog_signature(raw, sig, [other_pub]) is False
def test_verify_catalog_signature_matches_any_key_in_list():
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
other_pub = Ed25519PrivateKey.generate().public_key().public_bytes_raw()
raw = json.dumps(_minimal_catalog()).encode("utf-8")
sig = _sign(raw, priv)
# signing key is second in the list -- rotation support
assert c.verify_catalog_signature(raw, sig, [other_pub, pub]) is True
def test_verify_catalog_signature_garbage_sig_fails():
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
raw = json.dumps(_minimal_catalog()).encode("utf-8")
assert c.verify_catalog_signature(raw, b"not-a-real-signature", [pub]) is False
assert c.verify_catalog_signature(raw, b"", [pub]) is False
def test_verify_catalog_signature_missing_signature_returns_false():
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
raw = json.dumps(_minimal_catalog()).encode("utf-8")
assert c.verify_catalog_signature(raw, None, [pub]) is False
def test_verify_catalog_signature_never_raises_on_garbage_inputs():
assert c.verify_catalog_signature(b"", b"", []) is False
assert c.verify_catalog_signature(b"x", b"y", [b"too-short"]) is False
assert c.verify_catalog_signature("not-bytes", b"y", [b"\x00" * 32]) is False
assert c.verify_catalog_signature(b"x", b"y", None) is False
# --- validate_catalog: per-rule rejections --------------------------------- #
def test_validate_catalog_rejects_non_dict_root():
assert c.validate_catalog(["not", "a", "dict"]) != []
def test_validate_catalog_rejects_bad_schema_and_version():
data = _minimal_catalog()
data["schema"] = 0
data["version"] = -1
problems = c.validate_catalog(data)
assert any("schema" in p for p in problems)
assert any("version" in p for p in problems)
def test_validate_catalog_basic_requires_config():
data = _catalog_with({"config": None})
problems = c.validate_catalog(data)
assert any("config" in p for p in problems)
def test_validate_catalog_link_only_forbids_config():
data = _minimal_catalog()
data["servers"][0] = {
"id": "hosted",
"display": "Hosted",
"description": "A hosted connector.",
"category": "dev",
"homepage": "https://example.com/hosted",
"official": True,
"setup": "link-only",
"env_required": {},
"docs_url": "https://example.com/hosted/docs",
"notes": "",
"config": {"command": "npx", "args": ["-y", "should-not-be-here"]},
}
problems = c.validate_catalog(data)
assert any("must not have a 'config'" in p for p in problems)
def test_validate_catalog_rejects_disallowed_command():
data = _catalog_with({"config": {"command": "bash", "args": ["-c", "echo hi"]}})
problems = c.validate_catalog(data)
assert any("allowlist" in p for p in problems)
def test_validate_catalog_rejects_node_eval_flag():
data = _catalog_with({"config": {"command": "node", "args": ["-e", "require('fs')"]}})
problems = c.validate_catalog(data)
assert any("-e/--eval/-c" in p for p in problems)
def test_validate_catalog_rejects_python_c_flag():
data = _catalog_with({"config": {"command": "python3", "args": ["-c", "import os"]}})
problems = c.validate_catalog(data)
assert any("-e/--eval/-c" in p for p in problems)
def test_validate_catalog_rejects_docker_privileged():
data = _catalog_with(
{"config": {"command": "docker", "args": ["run", "--privileged", "some/image"]}}
)
problems = c.validate_catalog(data)
assert any("--privileged" in p for p in problems)
def test_validate_catalog_rejects_docker_root_volume_mount():
data = _catalog_with(
{"config": {"command": "docker", "args": ["run", "-v", "/:/host", "some/image"]}}
)
problems = c.validate_catalog(data)
assert any("mounts" in p for p in problems)
def test_validate_catalog_rejects_docker_home_volume_mount():
data = _catalog_with(
{"config": {"command": "docker", "args": ["run", "--volume=$HOME:/host", "some/image"]}}
)
problems = c.validate_catalog(data)
assert any("mounts" in p for p in problems)
def test_validate_catalog_rejects_nonempty_env_required():
data = _catalog_with({"env_required": {"API_TOKEN": "sk-shouldnotbehere"}})
problems = c.validate_catalog(data)
assert any("env_required" in p for p in problems)
def test_validate_catalog_rejects_secret_looking_arg():
data = _catalog_with(
{"config": {"command": "npx", "args": ["-y", "widget-mcp", "--api-key=sk-abcdef123"]}}
)
problems = c.validate_catalog(data)
assert any("secret-looking" in p for p in problems)
def test_validate_catalog_rejects_token_prefix_positional_arg():
data = _catalog_with(
{"config": {"command": "npx", "args": ["-y", "widget-mcp", "ghp_abcdef123456"]}}
)
problems = c.validate_catalog(data)
assert any("secret-looking" in p for p in problems)
def test_validate_catalog_rejects_http_url():
data = _catalog_with({"homepage": "http://example.com/widget"})
problems = c.validate_catalog(data)
assert any("homepage" in p for p in problems)
def test_validate_catalog_rejects_file_url():
data = _catalog_with({"docs_url": "file:///etc/passwd"})
problems = c.validate_catalog(data)
assert any("docs_url" in p for p in problems)
def test_validate_catalog_rejects_non_ascii_id():
data = _catalog_with({"id": "wídget"})
problems = c.validate_catalog(data)
assert any("ASCII" in p for p in problems)
def test_validate_catalog_rejects_non_ascii_command():
data = _catalog_with({"config": {"command": "npxé", "args": ["-y", "widget-mcp"]}})
problems = c.validate_catalog(data)
assert any("ASCII" in p for p in problems)
def test_validate_catalog_rejects_non_ascii_arg():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "wídget-mcp"]}})
problems = c.validate_catalog(data)
assert any("non-ASCII" in p for p in problems)
def test_validate_catalog_rejects_duplicate_ids():
data = _minimal_catalog()
data["servers"].append(dict(data["servers"][0]))
problems = c.validate_catalog(data)
assert any("duplicate id" in p for p in problems)
# --- validate_catalog: config.env (issue #68 finding 2) -------------------- #
def test_validate_catalog_rejects_each_denied_env_key():
for key in sorted(c.CATALOG_DENIED_ENV_KEYS):
data = _catalog_with(
{"config": {"command": "npx", "args": ["-y", "widget-mcp@1.0.0"], "env": {key: ""}}}
)
problems = c.validate_catalog(data)
assert any("deny-list" in p for p in problems), (key, problems)
def test_validate_catalog_rejects_denied_env_key_case_insensitively():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"node_options": ""},
}
}
)
problems = c.validate_catalog(data)
assert any("deny-list" in p for p in problems)
def test_validate_catalog_rejects_nonempty_nonplaceholder_env_value():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FOO_URL": "https://example.com"},
}
}
)
problems = c.validate_catalog(data)
assert any("empty string or a single <PLACEHOLDER>" in p for p in problems)
def test_validate_catalog_accepts_empty_and_placeholder_env_values():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FOO": "", "BAR_URL": "<BAR_URL>"},
}
}
)
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_non_ascii_env_key():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FÖO": ""},
}
}
)
problems = c.validate_catalog(data)
assert any("config.env key" in p and "ASCII" in p for p in problems)
def test_validate_catalog_rejects_non_ascii_env_value():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FOO": "<Bäd>"},
}
}
)
problems = c.validate_catalog(data)
assert any("config.env value" in p and "ASCII" in p for p in problems)
def test_validate_catalog_rejects_secret_looking_env_value():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"SOME_TOKEN": "ghp_abcdef1234567890"},
}
}
)
problems = c.validate_catalog(data)
assert any("real secret value" in p for p in problems)
def test_validate_catalog_rejects_node_options_env_walking_past_allowlist():
# The exact reproduction from issue #68 finding 2: an allowlisted
# `npx` command carrying NODE_OPTIONS in env, which previously passed
# validation and would have flowed straight into the executed
# subprocess via catalog_entry_to_paste_json().
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"NODE_OPTIONS": "--require /tmp/payload.js"},
}
}
)
problems = c.validate_catalog(data)
assert problems != []
# --- validate_catalog: version pinning (issue #68 finding 3) --------------- #
def test_validate_catalog_rejects_unpinned_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "widget-mcp"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_rejects_unpinned_scoped_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "@scope/pkg"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_accepts_pinned_scoped_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_unpinned_uvx_package():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_accepts_uvx_at_version_pin():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool@1.0.0"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_accepts_uvx_double_equals_pin():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool==1.0.0"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_docker_latest_tag():
data = _catalog_with({"config": {"command": "docker", "args": ["run", "some/image:latest"]}})
problems = c.validate_catalog(data)
assert any("'latest'" in p for p in problems)
def test_validate_catalog_rejects_docker_untagged_image():
data = _catalog_with({"config": {"command": "docker", "args": ["run", "some/image"]}})
problems = c.validate_catalog(data)
assert any("no explicit tag" in p for p in problems)
def test_validate_catalog_accepts_pinned_docker_image_with_flags():
data = _catalog_with(
{
"config": {
"command": "docker",
"args": ["run", "-i", "--rm", "-e", "SOME_TOKEN", "some/image:1.2.3"],
}
}
)
assert c.validate_catalog(data) == []
def test_validate_catalog_does_not_pin_check_placeholders_flags_or_subcommand():
# A pinned uvx spec followed by flags and a <PLACEHOLDER> positional
# must not itself get mistaken for an unpinned package.
data = _catalog_with(
{
"config": {
"command": "uvx",
"args": ["mcp-server-git@2026.7.10", "--repository", "<REPO_PATH>"],
}
}
)
assert c.validate_catalog(data) == []
# --- validate_catalog: id constraint (issue #68 finding 7) ----------------- #
def test_validate_catalog_rejects_id_with_markup():
data = _catalog_with({"id": "<b>Verified</b>"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_rejects_id_with_uppercase():
data = _catalog_with({"id": "Widget"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_rejects_id_starting_with_dash():
data = _catalog_with({"id": "-widget"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_accepts_valid_slug_id():
data = _catalog_with({"id": "widget-2.thing-ok"})
assert c.validate_catalog(data) == []
# --- resolve_catalog -------------------------------------------------------- #
def test_resolve_catalog_nothing_available_returns_empty_dict():
assert c.resolve_catalog(None, None, None) == {}
def test_resolve_catalog_prefers_highest_verified_version(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
bundled = _signed(_minimal_catalog(version=1), priv)
cached = _signed(_minimal_catalog(version=2), priv)
remote = _signed(_minimal_catalog(version=3), priv)
result = c.resolve_catalog(bundled, cached, remote)
assert c.catalog_version(result) == 3
def test_resolve_catalog_rejects_unsigned_bundled_catalog(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
# Bundled claims a very high version but is NOT signed by a trusted key
# -- it must get no implicit trust just for being the local copy.
malicious_raw = json.dumps(_minimal_catalog(version=100)).encode("utf-8")
bundled = (malicious_raw, b"totally-not-a-signature")
remote = _signed(_minimal_catalog(version=3), priv)
result = c.resolve_catalog(bundled, None, remote)
assert c.catalog_version(result) == 3
def test_resolve_catalog_rejects_rolled_back_version(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
cached = _signed(_minimal_catalog(version=5), priv)
rolled_back_remote = _signed(_minimal_catalog(version=2), priv)
result = c.resolve_catalog(None, cached, rolled_back_remote)
assert c.catalog_version(result) == 5
def test_resolve_catalog_rejects_absurd_version_jump(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
# The freeze attempt goes FIRST (as `cached`), the legitimate catalog
# SECOND (as `remote`) -- on purpose. Putting the good catalog first
# (as an earlier version of this test did) never exercises the
# vulnerable path: the old implementation only guarded a candidate
# against "the best accepted so far," so whichever candidate was
# evaluated FIRST got in unconditionally, uncapped. Ordering the freeze
# attempt first is what actually proves the cap holds regardless of
# evaluation order.
freeze_attempt = _signed(_minimal_catalog(version=999999), priv)
good = _signed(_minimal_catalog(version=5), priv)
result = c.resolve_catalog(None, freeze_attempt, good)
assert c.catalog_version(result) == 5
def test_resolve_catalog_caps_first_and_only_candidate(monkeypatch):
# issue #68 finding 6: with no bundled catalog to anchor against, a
# signed catalog claiming an absurd version must still be capped even
# when it is the ONLY candidate resolve_catalog() ever sees -- there is
# no "best so far" for it to be compared against, so the cap has to
# apply unconditionally, not "once something else has already landed."
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
freeze_attempt = _signed(_minimal_catalog(version=999999999), priv)
result = c.resolve_catalog(None, None, freeze_attempt)
assert result == {}
def test_resolve_catalog_anchors_cap_to_bundled_not_a_chained_best(monkeypatch):
# Anti-freeze must be measured against the BUNDLED version specifically,
# not against "whatever the best-so-far happens to be after each
# candidate is accepted" -- a chained anchor lets each accepted
# candidate ratchet the allowed ceiling upward, so a legitimate
# moderate bump (cached) plus a second, much larger jump (remote) can
# each individually look "within _CATALOG_MAX_VERSION_JUMP of the
# previous one" while remote is nowhere near bundled's version.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
bundled = _signed(_minimal_catalog(version=2), priv)
cached = _signed(_minimal_catalog(version=1000), priv) # within 1000 of bundled
remote = _signed(_minimal_catalog(version=1900), priv) # within 1000 of cached,
# NOT of bundled
result = c.resolve_catalog(bundled, cached, remote)
assert c.catalog_version(result) == 1000
def test_resolve_catalog_prefers_bundled_on_version_tie(monkeypatch):
# issue #68 finding 6: on a tie the LAST candidate evaluated used to
# win, so remote silently beat bundled at equal version. Bundled --
# the copy frozen into the binary -- must win ties.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
bundled_data = _minimal_catalog(version=5)
bundled = _signed(bundled_data, priv)
remote_data = _minimal_catalog(version=5)
remote_data["servers"][0]["display"] = "Remote Impostor"
remote = _signed(remote_data, priv)
result = c.resolve_catalog(bundled, None, remote)
assert c.catalog_version(result) == 5
assert result["servers"][0]["display"] == "Widget"
def test_resolve_catalog_floor_rejects_below_persisted_version(monkeypatch):
# `floor` is a pure parameter: the caller (eventually the GUI, from
# persisted storage) can pass a previously-accepted version, and
# nothing below it may be accepted even with no bundled catalog to
# anchor against.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
stale = _signed(_minimal_catalog(version=3), priv)
result = c.resolve_catalog(None, None, stale, floor=10)
assert result == {}
def test_resolve_catalog_malformed_candidate_does_not_raise(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
malformed_raw = b"{not valid json"
malformed_sig = _sign(malformed_raw, priv)
good = _signed(_minimal_catalog(version=1), priv)
result = c.resolve_catalog((malformed_raw, malformed_sig), None, good)
assert c.catalog_version(result) == 1
def test_resolve_catalog_invalid_but_signed_candidate_is_skipped(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
invalid = _signed(_catalog_with({"config": {"command": "bash", "args": []}}), priv)
good = _signed(_minimal_catalog(version=1), priv)
result = c.resolve_catalog(invalid, None, good)
assert c.catalog_version(result) == 1
# --- catalog_entry_to_paste_json / config_has_unfilled_placeholders ------- #
def test_catalog_entry_to_paste_json_basic_shape():
entry = _minimal_catalog()["servers"][0]
result = c.catalog_entry_to_paste_json(entry)
assert result == {"widget": {"command": "npx", "args": ["-y", "widget-mcp@1.0.0"]}}
def test_catalog_entry_to_paste_json_includes_env_when_present():
# NOTE: catalog_entry_to_paste_json() is a pure shape-converter for an
# entry that has ALREADY passed validate_catalog() -- it is correct for
# it to carry env through verbatim. The bug (issue #68 finding 2) was
# never in this function; it was that validate_catalog() let entries
# with dangerous/non-placeholder env values reach this function in the
# first place. This test now proves that boundary explicitly: a
# validation-legal env value (a <PLACEHOLDER> token) survives the
# conversion, and a value validate_catalog() would have rejected is
# confirmed rejected before it ever gets here.
entry = _minimal_catalog()["servers"][0]
entry["config"]["env"] = {"GRAFANA_URL": "<GRAFANA_URL>"}
result = c.catalog_entry_to_paste_json(entry)
assert result["widget"]["env"] == {"GRAFANA_URL": "<GRAFANA_URL>"}
catalog = _minimal_catalog()
catalog["servers"][0]["config"]["env"] = {"GRAFANA_URL": "<GRAFANA_URL>"}
assert c.validate_catalog(catalog) == []
malicious = _minimal_catalog()
malicious["servers"][0]["config"]["env"] = {"NODE_OPTIONS": "--require /tmp/payload.js"}
assert c.validate_catalog(malicious) != []
def test_config_has_unfilled_placeholders_true_for_token():
cfg = {"command": "npx", "args": ["-y", "server", "<ALLOWED_DIR>"]}
assert c.config_has_unfilled_placeholders(cfg) is True
def test_config_has_unfilled_placeholders_false_after_fill():
cfg = {"command": "npx", "args": ["-y", "server", "/Users/me/project"]}
assert c.config_has_unfilled_placeholders(cfg) is False
def test_config_has_unfilled_placeholders_checks_env_too():
cfg = {"command": "uvx", "args": ["mcp-grafana"], "env": {"GRAFANA_URL": "<GRAFANA_URL>"}}
assert c.config_has_unfilled_placeholders(cfg) is True
# --------------------------------------------------------------------------- #
# #72 -- a server value that isn't a JSON object must not take the load down
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("bad", ["not-a-dict", 123, ["a", "b"], None, True, 1.5])
def test_extract_servers_survives_non_dict_server_value(bad):
entries = c.extract_servers({"mcpServers": {"foo": bad}})
assert len(entries) == 1
assert entries[0].name == "foo"
assert entries[0].data == {}
assert entries[0].malformed is True
assert entries[0].raw == bad
def test_extract_servers_marks_only_the_bad_entry():
cfg = {"mcpServers": {"good": {"command": "npx"}, "bad": "oops"}}
by_name = {e.name: e for e in c.extract_servers(cfg)}
assert by_name["good"].malformed is False
assert by_name["good"].data == {"command": "npx"}
assert by_name["bad"].malformed is True
def test_extract_servers_handles_malformed_disabled_entry():
entries = c.extract_servers({c.DISABLED_KEY: {"parked": ["nope"]}})
assert entries[0].enabled is False
assert entries[0].malformed is True
def test_malformed_entry_round_trips_through_save_unchanged():
"""The cardinal rule: never silently delete what the user had on disk."""
cfg = {"mcpServers": {"good": {"command": "npx"}, "bad": "oops"}}
servers = c.extract_servers(cfg)
out = c.apply_servers(dict(cfg), servers)
assert out["mcpServers"]["bad"] == "oops"
assert out["mcpServers"]["good"] == {"command": "npx"}
def test_editing_a_malformed_entry_retires_the_raw_value():
entry = c.extract_servers({"mcpServers": {"bad": "oops"}})[0]
entry.set_data({"command": "npx"})
assert entry.malformed is False
assert entry.config_value() == {"command": "npx"}
assert c.apply_servers({}, [entry])["mcpServers"]["bad"] == {"command": "npx"}
def test_lint_reports_the_malformed_entry_by_name():
servers = c.extract_servers({"mcpServers": {"bad": "oops"}})
warnings = c.lint_servers(servers)
assert len(warnings) == 1
assert "'bad'" in warnings[0]
assert "not an object" in warnings[0]
assert "str" in warnings[0]
def test_lint_still_reports_normal_warnings_alongside_malformed():
cfg = {"mcpServers": {"bad": "oops", "sloppy": {"command": "npx", "args": "one two"}}}
warnings = c.lint_servers(c.extract_servers(cfg))
assert any("not an object" in w for w in warnings)
assert any("'args' should be a list" in w for w in warnings)
# --------------------------------------------------------------------------- #
# #73 -- the stale-file merge must not discard BCC-authored keys
# --------------------------------------------------------------------------- #
def test_carry_owned_keys_moves_sets_onto_the_reloaded_config():
local = {"mcpServers": {}, c.SETS_KEY: {"work": ["a", "b"]}}
fresh = {"mcpServers": {"external": {"command": "npx"}}}
contested = c.carry_owned_keys(local, fresh)
assert contested == []
assert fresh[c.SETS_KEY] == {"work": ["a", "b"]}
assert fresh["mcpServers"] == {"external": {"command": "npx"}}
def test_carry_owned_keys_reports_a_genuine_conflict():
local = {c.SETS_KEY: {"work": ["a"]}}
fresh = {c.SETS_KEY: {"work": ["a", "b"]}}
assert c.carry_owned_keys(local, fresh) == [c.SETS_KEY]
assert fresh[c.SETS_KEY] == {"work": ["a"]} # local wins: BCC owns the key
def test_carry_owned_keys_is_quiet_when_both_sides_agree():
local = {c.SETS_KEY: {"work": ["a"]}}
fresh = {c.SETS_KEY: {"work": ["a"]}}
assert c.carry_owned_keys(local, fresh) == []
def test_carry_owned_keys_leaves_disk_alone_when_absent_locally():
"""Can't distinguish 'deleted my last set' from 'never had sets'; keep theirs."""
fresh = {c.SETS_KEY: {"remote": ["a"]}}
assert c.carry_owned_keys({}, fresh) == []
assert fresh[c.SETS_KEY] == {"remote": ["a"]}
def test_carry_owned_keys_deep_copies_so_later_edits_do_not_leak():
local = {c.SETS_KEY: {"work": ["a"]}}
fresh = {}
c.carry_owned_keys(local, fresh)
local[c.SETS_KEY]["work"].append("b")
assert fresh[c.SETS_KEY] == {"work": ["a"]}
def test_merge_flow_preserves_sets_and_external_servers(tmp_path):
"""End-to-end shape of the Merge & save path that lost sets in #73."""
path = tmp_path / "claude.json"
path.write_text(json.dumps({"mcpServers": {"old": {"command": "old"}}}))
# BCC loads, user saves a named set and edits servers in memory.
local = c.load_config(path)
servers = c.extract_servers(local)
c.save_server_set(local, "work", servers)
# Something else rewrites the file underneath us.
path.write_text(json.dumps({"mcpServers": {"external": {"command": "new"}}, "other": 1}))
# Merge & save: reload disk, carry BCC keys, re-apply the user's servers.
fresh = c.load_config(path)
c.carry_owned_keys(local, fresh)
c.apply_servers(fresh, servers)
c.write_config(path, fresh)
saved = c.load_config(path)
assert saved[c.SETS_KEY] == {"work": ["old"]} # the set survived
assert saved["other"] == 1 # unrelated external key preserved
assert "old" in saved["mcpServers"] # user's servers re-applied
def test_null_server_value_is_malformed_not_mistaken_for_absent():
"""`{"mcpServers": {"foo": null}}` is legal JSON and a real malformed case,
so None must not double as the 'nothing here' sentinel."""
entry = c.extract_servers({"mcpServers": {"foo": None}})[0]
assert entry.malformed is True
assert entry.raw is None
assert c.apply_servers({}, [entry])["mcpServers"]["foo"] is None
def test_a_normal_entry_is_not_malformed():
entry = c.extract_servers({"mcpServers": {"foo": {"command": "npx"}}})[0]
assert entry.malformed is False
assert entry.raw is c.NO_RAW
# --------------------------------------------------------------------------- #
# #75 -- theming
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"setting,system_dark,expected",
[
(c.THEME_DARK, False, "dark"),
(c.THEME_DARK, True, "dark"),
(c.THEME_LIGHT, False, "light"),
(c.THEME_LIGHT, True, "light"),
(c.THEME_SYSTEM, True, "dark"),
(c.THEME_SYSTEM, False, "light"),
],
)
def test_resolve_theme_covers_every_setting_and_appearance(setting, system_dark, expected):
assert c.resolve_theme(setting, system_dark) == expected
@pytest.mark.parametrize("junk", ["", "solarized", None, "DARK", 3])
def test_resolve_theme_falls_back_to_following_the_system(junk):
"""A hand-edited or future QSettings value should follow the desktop,
not pin a fixed theme."""
assert c.resolve_theme(junk, True) == "dark"
assert c.resolve_theme(junk, False) == "light"
def test_palette_for_known_names():
assert c.palette_for("dark") is c.DARK_PALETTE
assert c.palette_for("light") is c.LIGHT_PALETTE
def test_palette_for_unknown_name_falls_back_to_dark():
assert c.palette_for("chartreuse") is c.DARK_PALETTE
def test_dark_palette_is_unchanged_from_the_shipped_look():
"""v1.3.0 shipped these exact colours; adding a light theme must not
quietly restyle the dark one."""
p = c.DARK_PALETTE
assert (p.accent, p.bg, p.panel, p.panel_2) == ("#f97316", "#1b1d23", "#23262e", "#2b2f39")
assert (p.text, p.muted, p.border) == ("#e7e9ee", "#9aa0ad", "#3a3f4b")
assert (p.good, p.bad, p.warn, p.remote) == ("#4ade80", "#f87171", "#fbbf24", "#60a5fa")
assert (p.on_accent, p.disabled_bg, p.mono_bg) == ("#1a1205", "#202229", "#16181d")
def test_both_palettes_define_every_slot():
"""A missing slot should fail here rather than render a broken window."""
for pal in (c.DARK_PALETTE, c.LIGHT_PALETTE):
for f in dataclasses.fields(c.Palette):
value = getattr(pal, f.name)
assert value, f"{pal.name}.{f.name} is empty"
if f.name != "name":
assert re.fullmatch(r"#[0-9a-fA-F]{6}", value), f"{pal.name}.{f.name}={value!r}"
@pytest.mark.parametrize("pal_name", ["dark", "light"])
@pytest.mark.parametrize("slot", ["text", "muted", "good", "bad", "warn", "remote", "accent"])
def test_palette_meets_contrast_on_panel(pal_name, slot):
"""Every colour drawn as text/glyph must clear WCAG AA (4.5:1) against the
surface it sits on. The light palette's semantic colours are NOT the dark
ones lightened -- #4ade80 sits near 1.7:1 on white -- so this guards
against someone 'harmonising' them back toward the dark hues."""
pal = c.palette_for(pal_name)
assert c.contrast_ratio(getattr(pal, slot), pal.panel) >= 4.5
@pytest.mark.parametrize("pal_name", ["dark", "light"])
def test_on_accent_is_legible_against_the_accent_fill(pal_name):
"""Primary buttons and selected rows draw on_accent on top of accent."""
pal = c.palette_for(pal_name)
assert c.contrast_ratio(pal.on_accent, pal.accent) >= 4.5
def test_contrast_ratio_endpoints():
assert c.contrast_ratio("#000000", "#ffffff") == pytest.approx(21.0, abs=0.01)
assert c.contrast_ratio("#123456", "#123456") == pytest.approx(1.0, abs=0.001)
assert c.contrast_ratio("#ffffff", "#000000") == pytest.approx(21.0, abs=0.01)
def test_relative_luminance_extremes():
assert c.relative_luminance("#000000") == pytest.approx(0.0)
assert c.relative_luminance("#ffffff") == pytest.approx(1.0)
def test_stylesheet_builder_has_no_hardcoded_colours():
"""Every colour in the QSS must come from the palette.
Three near-black literals used to be inlined here (#1a1205, #202229,
#16181d). Harmless with one theme; with two, they silently render dark
chrome on a light window. Reads the source rather than importing bcc,
which needs PySide6.
"""
src = (Path(__file__).resolve().parent.parent / "bcc.py").read_text(encoding="utf-8")
start = src.index("def build_stylesheet")
body = src[start : src.index("def apply_palette")]
assert re.findall(r"#[0-9a-fA-F]{6}", body) == []
def test_every_palette_slot_is_consumed():
"""A slot added to Palette but never wired up is dead weight.
Checks for `p.<slot>` anywhere in bcc.py, which covers both the QSS and
apply_palette's global bindings -- not every slot belongs in the
stylesheet (`good` and `remote` feed the inline status dots via
STATUS_COLORS/HEALTH_COLORS, never the QSS). This won't catch a slot bound
to a global that nothing then uses; it does catch the common mistake of
extending the dataclass and forgetting to plumb it through.
"""
src = (Path(__file__).resolve().parent.parent / "bcc.py").read_text(encoding="utf-8")
for f in dataclasses.fields(c.Palette):
if f.name == "name":
continue
assert f"p.{f.name}" in src, f"palette slot {f.name!r} is never consumed"
# --------------------------------------------------------------------------- #
# #78/#79 -- update notice: when to show it, and what it says
# --------------------------------------------------------------------------- #
def test_update_notice_when_a_newer_release_exists():
n = c.update_notice("1.2.0", {"version": "v1.3.0", "url": "https://example.test/rel"})
assert n is not None
assert n["version"] == "v1.3.0"
assert n["url"] == "https://example.test/rel"
assert "1.3.0" in n["text"] and "1.2.0" in n["text"]
def test_update_notice_is_silent_when_current():
assert c.update_notice("1.3.0", {"version": "v1.3.0"}) is None
assert c.update_notice("1.4.0", {"version": "v1.3.0"}) is None
@pytest.mark.parametrize("bad", [None, {}, {"version": ""}, {"version": None}, {"version": 3}, []])
def test_update_notice_is_silent_on_a_failed_or_malformed_check(bad):
"""fetch_latest_release returns None on any failure; a half-formed payload
must not produce a notice pointing at nothing."""
assert c.update_notice("1.0.0", bad) is None
def test_update_notice_falls_back_to_the_releases_page_without_a_url():
n = c.update_notice("1.0.0", {"version": "v2.0.0"})
assert n["url"] == c.RELEASES_URL
def test_update_notice_names_no_menu_path():
"""The old status-line text said 'Help > About to view it', which is wrong
on macOS -- Qt moves the About action into the application menu (#79). The
notice carries its own action, so it must not describe a menu path."""
n = c.update_notice("1.0.0", {"version": "v2.0.0"})
lowered = n["text"].lower()
for phrase in ("help", "about", "menu", "▸", ">"):
assert phrase not in lowered, f"notice text should not reference {phrase!r}"
def test_update_notice_handles_the_v_prefix_consistently():
assert c.update_notice("1.2.0", {"version": "1.3.0"}) is not None
assert c.update_notice("v1.2.0", {"version": "v1.3.0"}) is not None
assert c.update_notice("1.3.0", {"version": "v1.3.0"}) is None
def test_update_notice_renders_both_versions_the_same_way():
"""Tags carry a 'v' prefix, __version__ doesn't -- don't show both forms
in one sentence."""
n = c.update_notice("1.2.0", {"version": "v1.3.0"})
assert "v1.3.0" not in n["text"]
assert "1.3.0" in n["text"] and "1.2.0" in n["text"]
# the machine-readable field keeps the real tag
assert n["version"] == "v1.3.0"
# --------------------------------------------------------------------------- #
# #76 -- ${VAR} references. Semantics mirror Claude Code's documented
# behaviour: ${VAR} and ${VAR:-default}, expanded in command/args/env/url/
# headers, and an unset variable with no default left as literal text.
# --------------------------------------------------------------------------- #
def test_find_env_refs_plain_and_defaulted():
refs = c.find_env_refs("${A} and ${B:-fallback}")
assert [(r.name, r.default) for r in refs] == [("A", None), ("B", "fallback")]
@pytest.mark.parametrize("text", ["${}", "${1BAD}", "$NOTBRACED", "{NOPE}", "plain", "$${X"])
def test_find_env_refs_ignores_non_references(text):
assert c.find_env_refs(text) == []
def test_find_env_refs_allows_an_empty_default():
"""`${VAR:-}` is a documented way to say 'blank if unset'."""
refs = c.find_env_refs("${A:-}")
assert refs[0].default == ""
assert refs[0].has_default is True
def test_server_env_refs_covers_all_five_documented_fields():
data = {
"command": "${BIN}",
"args": ["--x", "${ARG}"],
"env": {"K": "${ENVV}"},
"url": "${URL}/mcp",
"headers": {"Authorization": "Bearer ${HDR}"},
}
found = {(r.name, r.field) for r in c.server_env_refs(data)}
assert found == {
("BIN", "command"),
("ARG", "args"),
("ENVV", "env"),
("URL", "url"),
("HDR", "headers"),
}
def test_server_env_refs_ignores_unexpanded_fields():
"""Claude Code expands five fields; a ${VAR} elsewhere isn't a reference."""
assert c.server_env_refs({"description": "${NOPE}", "timeout": "${ALSO_NO}"}) == []
def test_expand_env_refs_matches_documented_semantics():
env = {"SET": "value"}
assert c.expand_env_refs("${SET}", env) == "value"
assert c.expand_env_refs("${MISSING:-dflt}", env) == "dflt"
assert c.expand_env_refs("${SET:-dflt}", env) == "value"
# unset with no default: left as literal text, exactly as Claude Code does
assert c.expand_env_refs("${MISSING}", env) == "${MISSING}"
def test_expand_env_refs_handles_several_in_one_string():
assert c.expand_env_refs("${A}/${B:-two}/${C}", {"A": "one"}) == "one/two/${C}"
def test_unresolved_env_refs_only_flags_unset_without_default():
data = {"env": {"A": "${SET}", "B": "${UNSET}", "C": "${OTHER:-has_default}"}}
assert [r.name for r in c.unresolved_env_refs(data, {"SET": "x"})] == ["UNSET"]
# --- the two interactions that were backwards for this feature ------------
def test_placeholder_under_a_secret_key_is_not_masked():
"""A ${VAR} names a secret rather than being one. Masking it would make a
reference indistinguishable from a stored credential."""
assert c.should_mask_value("API_KEY", "${API_KEY}") is False
assert c.should_mask_value("API_KEY", "ghp_realsecret") is True
assert c.should_mask_value("NOT_SECRET", "${API_KEY}") is False
def test_redacted_display_keeps_placeholders_but_masks_real_secrets():
out = c._redact_server_data({"env": {"API_KEY": "${API_KEY}", "TOKEN": "ghp_real"}})
assert out["env"]["API_KEY"] == "${API_KEY}"
assert out["env"]["TOKEN"] == c.MASK
def test_redact_args_keeps_placeholders_visible():
assert c.redact_args(["--token", "${GH_TOKEN}"]) == ["--token", "${GH_TOKEN}"]
assert c.redact_args(["--api-key=${K}"]) == ["--api-key=${K}"]
# real secrets still masked
assert c.redact_args(["--token", "ghp_real"]) == ["--token", c.MASK]
assert c.redact_args(["--api-key=sk-real"]) == [f"--api-key={c.MASK}"]
def test_args_secret_warning_is_silenced_by_a_placeholder():
"""Moving a token into ${VAR} is the recommended fix for this warning --
still warning afterwards would punish the fix."""
assert c.args_secret_warning({"args": ["--token", "ghp_real"]}) is not None
assert c.args_secret_warning({"args": ["--token", "${GH_TOKEN}"]}) is None
def test_args_secret_warning_still_fires_on_the_arg_after_a_placeholder():
"""A placeholder must clear the pending-flag state, not blanket-suppress."""
assert c.args_secret_warning({"args": ["${SAFE}", "--token", "ghp_real"]}) is not None
# --- per-client gating ----------------------------------------------------
def _profile(path):
return c.Profile(label="p", path=Path(path), config_exists=True)
def test_claude_code_profiles_expand_references():
assert c.client_expands_env_refs(_profile(Path.home() / ".claude.json")) is True
assert c.client_expands_env_refs(_profile("/repo/.mcp.json")) is True
def test_claude_desktop_profile_does_not_expand_references():
desktop = _profile(c.app_support_base() / "Claude" / c.CONFIG_FILENAME)
assert c.client_expands_env_refs(desktop) is False
def test_desktop_profile_warns_that_references_are_literal():
data = {"env": {"API_KEY": "${API_KEY}"}}
desktop = _profile(c.app_support_base() / "Claude" / c.CONFIG_FILENAME)
warnings = c.env_ref_warnings(data, desktop, {"API_KEY": "set"})
assert len(warnings) == 1
assert "NOT be expanded" in warnings[0]
assert "${API_KEY}" in warnings[0]
def test_claude_code_profile_warns_only_about_unset_variables():
code = _profile(Path.home() / ".claude.json")
data = {"env": {"A": "${UNSET_ONE}"}}
assert c.env_ref_warnings(data, code, {}) != []
assert c.env_ref_warnings(data, code, {"UNSET_ONE": "x"}) == []
# a default means it always resolves
assert c.env_ref_warnings({"env": {"A": "${X:-d}"}}, code, {}) == []
def test_no_references_means_no_warnings():
assert c.env_ref_warnings({"command": "npx", "args": ["-y", "pkg"]}, None) == []
# --------------------------------------------------------------------------- #
# Client adapters (issue #5 — cross-client support, phase 1)
#
# The refactor's promise is twofold: (1) the two Claude clients behave exactly
# as before, and (2) the ClientSpec seam is real — a client with a different
# servers key and a different per-server shape flows through the same pipeline.
# A synthetic "VS Code-like" spec stands in for the phase-2 client so the
# abstraction is proven now, before anything depends on it.
# --------------------------------------------------------------------------- #
def test_claude_specs_are_registered_and_mcpservers_shaped():
assert c.CLAUDE_DESKTOP.servers_key == "mcpServers"
assert c.CLAUDE_CODE.servers_key == "mcpServers"
assert c.CLAUDE_DESKTOP.disabled_key == c.DISABLED_KEY
assert c.CLAUDE_CODE.disabled_key == c.DISABLED_KEY
# capabilities the old inline filename checks used to compute
assert c.CLAUDE_DESKTOP.expands_env_refs is False
assert c.CLAUDE_CODE.expands_env_refs is True
assert c.CLAUDE_DESKTOP.supports_restart is True
assert c.CLAUDE_CODE.supports_restart is False
assert set(c.CLIENT_SPECS) == {c.CLAUDE_DESKTOP, c.CLAUDE_CODE}
assert c.DEFAULT_CLIENT is c.CLAUDE_DESKTOP
def test_client_by_key_round_trips_and_misses():
assert c.client_by_key("claude_desktop") is c.CLAUDE_DESKTOP
assert c.client_by_key("claude_code") is c.CLAUDE_CODE
assert c.client_by_key("nope") is None
def test_resolve_client_matches_the_old_filename_rule():
assert c.resolve_client("/x/Claude/claude_desktop_config.json") is c.CLAUDE_DESKTOP
assert c.resolve_client(Path.home() / ".claude.json") is c.CLAUDE_CODE
assert c.resolve_client("/repo/.mcp.json") is c.CLAUDE_CODE
assert c.resolve_client(Path.home() / ".claude" / "settings.json") is c.CLAUDE_CODE
def test_profile_auto_resolves_client_from_path():
desktop = c.Profile(
label="Claude", path="/x/Claude/claude_desktop_config.json", config_exists=True
)
code = c.Profile(label="Claude Code", path="/home/me/.claude.json", config_exists=True)
assert desktop.client is c.CLAUDE_DESKTOP
assert code.client is c.CLAUDE_CODE
def test_profile_honours_an_explicit_client():
# An explicit spec is not overridden by the path-based resolver.
p = c.Profile(
label="odd",
path="/somewhere/claude_desktop_config.json",
config_exists=True,
client=c.CLAUDE_CODE,
)
assert p.client is c.CLAUDE_CODE
def test_desktop_gating_and_env_expansion_read_off_the_spec():
desktop = c.Profile(label="d", path="/x/Claude/claude_desktop_config.json", config_exists=True)
code = c.Profile(label="c", path=Path.home() / ".claude.json", config_exists=True)
assert c.profile_targets_claude_desktop(desktop) is True
assert c.profile_targets_claude_desktop(code) is False
assert c.client_expands_env_refs(desktop) is False
assert c.client_expands_env_refs(code) is True
def test_extract_and_apply_default_spec_is_unchanged():
# No spec argument must behave byte-for-byte like the pre-refactor code.
cfg = {"mcpServers": {"a": {"command": "x"}}, "_disabledMcpServers": {"b": {"command": "y"}}}
servers = c.extract_servers(cfg)
assert {(s.name, s.enabled) for s in servers} == {("a", True), ("b", False)}
out = c.apply_servers({}, servers)
assert out == {
"mcpServers": {"a": {"command": "x"}},
"_disabledMcpServers": {"b": {"command": "y"}},
}
# A stand-in for the phase-2 VS Code adapter: different top-level key
# ("servers"), a different disabled key, and a per-server shape that carries a
# `type` field the internal model doesn't. entry_to/from_internal are the only
# things it overrides — proving that's the whole extension point.
class _FakeVSCode(c.ClientSpec):
def entry_to_internal(self, value):
if not isinstance(value, dict):
return value
return {k: v for k, v in value.items() if k != "type"}
def entry_from_internal(self, data):
if not isinstance(data, dict):
return data
return {"type": "stdio", **data}
_VSCODE = _FakeVSCode(
key="vscode_fake",
label="VS Code (test)",
servers_key="servers",
disabled_key="_bccDisabledServers",
)
def test_extract_reads_a_custom_servers_key_and_translates_shape():
cfg = {"servers": {"a": {"type": "stdio", "command": "x", "args": ["-y"]}}}
servers = c.extract_servers(cfg, _VSCODE)
assert len(servers) == 1
# the `type` field was translated out of the internal model
assert servers[0].data == {"command": "x", "args": ["-y"]}
def test_apply_writes_a_custom_key_translates_back_and_keeps_other_keys():
original = {"servers": {"old": {"type": "stdio", "command": "z"}}, "keepMe": {"x": 1}}
servers = c.extract_servers(original, _VSCODE)
out = c.apply_servers(original, servers, _VSCODE)
# round-trips through the custom key with the shape restored
assert out["servers"] == {"old": {"type": "stdio", "command": "z"}}
# the cardinal rule generalises: mcpServers is never introduced, and every
# unrelated key survives verbatim
assert "mcpServers" not in out
assert out["keepMe"] == {"x": 1}
def test_apply_uses_the_custom_disabled_key():
servers = [
c.ServerEntry("on", {"command": "a"}, True),
c.ServerEntry("off", {"command": "b"}, False),
]
out = c.apply_servers({}, servers, _VSCODE)
assert out["servers"] == {"on": {"type": "stdio", "command": "a"}}
assert out["_bccDisabledServers"] == {"off": {"type": "stdio", "command": "b"}}
assert c.DISABLED_KEY not in out
def test_spec_with_no_disabled_key_drops_disabled_and_never_parks():
no_park = c.ClientSpec(
key="nopark", label="No Park", servers_key="mcpServers", disabled_key=None
)
servers = [
c.ServerEntry("on", {"command": "a"}, True),
c.ServerEntry("off", {"command": "b"}, False),
]
out = c.apply_servers({}, servers, no_park)
assert out == {"mcpServers": {"on": {"command": "a"}}}
assert c.DISABLED_KEY not in out
assert no_park.section_keys() == ("mcpServers",)
def test_section_keys_reports_both_when_a_disabled_key_exists():
assert c.CLAUDE_DESKTOP.section_keys() == ("mcpServers", c.DISABLED_KEY)
assert _VSCODE.section_keys() == ("servers", "_bccDisabledServers")
def test_malformed_entry_round_trips_through_the_default_spec():
# #72's non-object server value must still be preserved verbatim on save.
cfg = {"mcpServers": {"bad": "oops", "good": {"command": "x"}}}
servers = c.extract_servers(cfg)
assert any(s.malformed and s.name == "bad" for s in servers)
out = c.apply_servers({}, servers)
assert out["mcpServers"]["bad"] == "oops"
def test_server_sections_and_change_summary_follow_a_custom_key(tmp_path):
loaded = {"servers": {"a": {"type": "stdio", "command": "x", "env": {"API_KEY": "sekret"}}}}
sections = c._server_sections(loaded, _VSCODE)
assert "servers" in sections
assert "mcpServers" not in sections
# secret masking still applies through the custom key
assert sections["servers"]["a"]["env"]["API_KEY"] == c.MASK
disk = {"servers": {"a": {"type": "stdio", "command": "CHANGED"}}}
p = tmp_path / "vscode.json"
p.write_text(json.dumps(disk), encoding="utf-8")
changed_keys, diff = c.external_change_summary(loaded, p, _VSCODE)
assert "servers" in changed_keys
assert diff # a server-section change under the custom key is diffed
# --------------------------------------------------------------------------- #
# Project profile label disambiguation (issue #74)
# --------------------------------------------------------------------------- #
def test_disambiguate_labels_no_collision_uses_basename():
dirs = [Path("/home/me/work/api"), Path("/home/me/work/web")]
labels = c.disambiguate_project_labels(dirs)
assert labels[Path("/home/me/work/api")] == "Project: api"
assert labels[Path("/home/me/work/web")] == "Project: web"
def test_disambiguate_labels_widens_only_colliding_basenames():
dirs = [
Path("/home/me/work/app"),
Path("/home/me/personal/app"),
Path("/home/me/notes"),
]
labels = c.disambiguate_project_labels(dirs)
# the two "app"s widen by one parent; the unique "notes" stays plain
assert labels[Path("/home/me/work/app")] == "Project: work/app"
assert labels[Path("/home/me/personal/app")] == "Project: personal/app"
assert labels[Path("/home/me/notes")] == "Project: notes"
def test_disambiguate_labels_widens_further_when_parent_also_collides():
dirs = [Path("/a/x/app"), Path("/b/x/app")]
labels = c.disambiguate_project_labels(dirs)
assert labels[Path("/a/x/app")] == "Project: a/x/app"
assert labels[Path("/b/x/app")] == "Project: b/x/app"
def _write_project(tmp_path, name, mcp_content):
d = tmp_path / name
d.mkdir(parents=True)
if mcp_content is not None:
(d / ".mcp.json").write_text(mcp_content, encoding="utf-8")
return d
def _claude_json_with_projects(tmp_path, dirs):
cj = tmp_path / ".claude.json"
cj.write_text(json.dumps({"projects": {str(d): {} for d in dirs}}), encoding="utf-8")
return cj
def test_discover_project_configs_disambiguates_same_basename(tmp_path):
d1 = _write_project(tmp_path / "work", "app", '{"mcpServers": {}}')
d2 = _write_project(tmp_path / "personal", "app", '{"mcpServers": {}}')
cj = _claude_json_with_projects(tmp_path, [d1, d2])
profiles = c.discover_project_configs(cj)
labels = sorted(p.label for p in profiles)
assert labels == ["Project: personal/app", "Project: work/app"]
def test_discover_project_configs_skips_non_object_and_garbage(tmp_path):
good = _write_project(tmp_path, "good", '{"mcpServers": {}}')
array = _write_project(tmp_path, "arr", "[1, 2, 3]")
garbage = _write_project(tmp_path, "junk", "not json at all")
missing = tmp_path / "nofile"
missing.mkdir()
cj = _claude_json_with_projects(tmp_path, [good, array, garbage, missing])
profiles = c.discover_project_configs(cj)
paths = {str(p.path) for p in profiles}
assert str(good / ".mcp.json") in paths
assert str(array / ".mcp.json") not in paths
assert str(garbage / ".mcp.json") not in paths
assert str(missing / ".mcp.json") not in paths
# --------------------------------------------------------------------------- #
# Removed-flag -> env migration (ssh-mcp v2 and the general mechanism)
# --------------------------------------------------------------------------- #
def test_detect_migratable_package_via_npx_args():
data = {"command": "npx", "args": ["-y", "ssh-mcp", "--host=h", "--password=p"]}
assert c.detect_migratable_package(data) == "ssh-mcp"
def test_detect_migratable_package_direct_command_and_path_and_version():
assert c.detect_migratable_package({"command": "ssh-mcp", "args": []}) == "ssh-mcp"
assert (
c.detect_migratable_package({"command": "/usr/local/bin/ssh-mcp", "args": []}) == "ssh-mcp"
)
assert (
c.detect_migratable_package({"command": "npx", "args": ["-y", "ssh-mcp@2.0.1"]})
== "ssh-mcp"
)
def test_detect_migratable_package_unknown_returns_none():
assert c.detect_migratable_package({"command": "npx", "args": ["some-other"]}) is None
assert c.detect_migratable_package({}) is None
assert c.detect_migratable_package({"url": "https://x"}) is None
def test_migrate_removed_flags_inline_form():
data = {
"command": "npx",
"args": ["-y", "ssh-mcp", "--", "--host=1.2.3.4", "--password=hunter2"],
}
new, notes = c.migrate_removed_flags(data)
assert new["args"] == ["-y", "ssh-mcp", "--", "--host=1.2.3.4"]
assert new["env"] == {"SSH_MCP_PASSWORD": "hunter2"}
assert any("SSH_MCP_PASSWORD" in n for n in notes)
# Original untouched (pure function).
assert "env" not in data
def test_migrate_removed_flags_separate_form():
data = {
"command": "npx",
"args": ["-y", "ssh-mcp", "--user", "root", "--password", "s3cret"],
}
new, _ = c.migrate_removed_flags(data)
assert new["args"] == ["-y", "ssh-mcp", "--user", "root"]
assert new["env"] == {"SSH_MCP_PASSWORD": "s3cret"}
def test_migrate_removed_flags_preserves_existing_env_and_merges():
data = {
"command": "npx",
"args": ["ssh-mcp", "--password=p"],
"env": {"OTHER": "keep"},
}
new, _ = c.migrate_removed_flags(data)
assert new["env"] == {"OTHER": "keep", "SSH_MCP_PASSWORD": "p"}
def test_migrate_removed_flags_does_not_clobber_existing_secret():
data = {
"command": "npx",
"args": ["ssh-mcp", "--password=fromargs"],
"env": {"SSH_MCP_PASSWORD": "fromenv"},
}
new, notes = c.migrate_removed_flags(data)
# env value wins; the redundant flag is still stripped so v2 can start.
assert new["env"] == {"SSH_MCP_PASSWORD": "fromenv"}
assert "--password=fromargs" not in new["args"]
assert any("already set" in n for n in notes)
def test_migrate_removed_flags_moves_env_ref_verbatim():
data = {"command": "npx", "args": ["ssh-mcp", "--password", "${MY_PW}"]}
new, _ = c.migrate_removed_flags(data)
assert new["env"] == {"SSH_MCP_PASSWORD": "${MY_PW}"}
assert "${MY_PW}" not in new["args"]
def test_migrate_removed_flags_noop_returns_same_object():
data = {"command": "npx", "args": ["ssh-mcp", "--host=h", "--user=u"]}
new, notes = c.migrate_removed_flags(data)
assert new is data
assert notes == []
def test_migrate_removed_flags_unknown_package_noop():
data = {"command": "npx", "args": ["mystery", "--password=p"]}
new, notes = c.migrate_removed_flags(data)
assert new is data and notes == []
def test_migrate_removed_flags_only_args_left_drops_args_key():
data = {"command": "ssh-mcp", "args": ["--password=p"]}
new, _ = c.migrate_removed_flags(data)
assert "args" not in new
assert new["env"] == {"SSH_MCP_PASSWORD": "p"}
def test_removed_flag_warnings_covers_migratable_and_manual():
data = {
"command": "npx",
"args": ["ssh-mcp", "--password=p", "--sudoPassword=s", "--disableSudo"],
}
warnings = c.removed_flag_warnings(data)
text = " ".join(warnings)
assert "--password" in text
assert "sudoPassword" in text
assert "disableSudo" in text
# #90: --password AND --sudoPassword are now both auto-migratable (the
# verified ssh-mcp facts map sudo/su → SSH_MCP_SUDO_PASSWORD). The migration
# LOGIC is unchanged — only the registry data grew — so both move into env.
# --disableSudo has no env replacement (sudo is now a role/policy) → warn-only.
new, _ = c.migrate_removed_flags(data)
assert new["env"] == {"SSH_MCP_PASSWORD": "p", "SSH_MCP_SUDO_PASSWORD": "s"}
assert "--sudoPassword=s" not in new["args"]
assert "--disableSudo" in new["args"]
def test_migrate_su_password_maps_to_sudo_env():
# --suPassword is the second flag that maps to the same SSH_MCP_SUDO_PASSWORD var.
data = {"command": "ssh-mcp", "args": ["--suPassword", "rootpw"]}
new, notes = c.migrate_removed_flags(data)
assert new["env"] == {"SSH_MCP_SUDO_PASSWORD": "rootpw"}
assert "args" not in new
assert notes
def test_migrate_sudo_and_su_two_flags_one_var_no_clobber():
# Both sudo flags map to one var; the no-clobber path keeps the first, drops
# the second (different value) with a note rather than silently overwriting.
data = {
"command": "ssh-mcp",
"args": ["--sudoPassword=first", "--suPassword=second"],
}
new, notes = c.migrate_removed_flags(data)
assert new["env"] == {"SSH_MCP_SUDO_PASSWORD": "first"}
assert "args" not in new
assert any("already set" in n for n in notes)
def test_removed_flag_warnings_empty_when_clean():
assert c.removed_flag_warnings({"command": "ssh-mcp", "args": ["--host=h"]}) == []
assert c.removed_flag_warning({"command": "ssh-mcp", "args": ["--host=h"]}) is None
def test_ssh_membermatters_end_to_end():
# The exact shape from AJ's failing server.
data = {
"command": "npx",
"args": [
"-y",
"ssh-mcp",
"--",
"--host=member.example.io",
"--user=deploy",
"--port=22",
"--password=topsecret",
],
}
new, notes = c.migrate_removed_flags(data)
assert new["args"] == [
"-y",
"ssh-mcp",
"--",
"--host=member.example.io",
"--user=deploy",
"--port=22",
]
assert new["env"] == {"SSH_MCP_PASSWORD": "topsecret"}
assert notes
# --------------------------------------------------------------------------- #
# ServerSpec spine (issue #90)
# --------------------------------------------------------------------------- #
def test_server_spec_registry_seeds_ssh_mcp():
spec = c.SERVER_SPECS["ssh-mcp"]
assert spec.package == "ssh-mcp"
# sudo/su both map to the one sudo env var (verified facts).
assert spec.env_flags["--sudoPassword"] == "SSH_MCP_SUDO_PASSWORD"
assert spec.env_flags["--suPassword"] == "SSH_MCP_SUDO_PASSWORD"
assert spec.env_flags["--password"] == "SSH_MCP_PASSWORD"
# disableSudo stays warn-only (no env replacement).
assert "--disableSudo" in spec.removed_flags
assert "--disableSudo" not in spec.env_flags
# Verified sidecar paths seeded per platform (README path differs → doc_path).
assert "ssh-mcp/config.toml" in spec.sidecar_paths["darwin"]
assert "Application Support" in spec.sidecar_paths["darwin"]
assert "APPDATA" in spec.sidecar_paths["win32"]
assert spec.sidecar_doc_path == "~/.config/ssh-mcp/config.toml"
# zod enums seeded as data (pick-lists later — #7).
assert spec.schema["auth"] == ["agent", "key", "password", "keychain"]
assert spec.schema["approvalMode"] == ["auto", "ask-destructive", "ask-all", "deny"]
assert spec.schema["role"] == ["viewer", "operator", "admin"]
assert spec.schema["port"] == {"min": 1, "max": 65535}
def test_flag_env_migrations_is_derived_from_server_specs():
# The legacy constant is now a derived view; it must mirror the registry.
assert set(c.FLAG_ENV_MIGRATIONS) == set(c.SERVER_SPECS)
assert c.FLAG_ENV_MIGRATIONS["ssh-mcp"]["env"] == c.SERVER_SPECS["ssh-mcp"].env_flags
assert c.FLAG_ENV_MIGRATIONS["ssh-mcp"]["removed"] == c.SERVER_SPECS["ssh-mcp"].removed_flags
def test_resolve_server_spec_matches_detect_migratable_package():
for data in (
{"command": "npx", "args": ["-y", "ssh-mcp"]},
{"command": "ssh-mcp", "args": []},
{"command": "/usr/local/bin/ssh-mcp", "args": []},
{"command": "npx", "args": ["some-other"]},
{},
{"url": "https://x"},
):
spec = c.resolve_server_spec(data)
pkg = c.detect_migratable_package(data)
assert (spec.package if spec else None) == pkg
def test_drift_warning_maxchars_none_inline_and_separate():
inline = {"command": "ssh-mcp", "args": ["--maxChars=none"]}
separate = {"command": "ssh-mcp", "args": ["--maxChars", "none"]}
for data in (inline, separate):
warnings = c.drift_warnings(data)
assert len(warnings) == 1
assert "--maxChars=none" in warnings[0]
assert "5000" in warnings[0]
def test_drift_warning_quiet_for_other_values_and_unknown_pkg():
# A real numeric cap is fine; only "none" drifts.
assert c.drift_warnings({"command": "ssh-mcp", "args": ["--maxChars=8000"]}) == []
assert c.drift_warnings({"command": "ssh-mcp", "args": ["--host=h"]}) == []
# Unknown package: nothing to say.
assert c.drift_warnings({"command": "npx", "args": ["other", "--maxChars=none"]}) == []
# --------------------------------------------------------------------------- #
# Sidecar config detection + precedence (issue #91)
# --------------------------------------------------------------------------- #
_SSH = {"command": "npx", "args": ["-y", "ssh-mcp", "--host=h", "--user=u"]}
_HOME = "/Users/tester"
def test_sidecar_path_verified_per_platform():
# NOTE: assert on .as_posix() — the CI matrix includes a Windows runner where
# str(Path("/Users/…")) would render with backslashes. as_posix() normalises
# separators so these platform-parameterised checks are portable.
spec = c.SERVER_SPECS["ssh-mcp"]
mac = c.sidecar_path(spec, platform="darwin", environ={}, home=_HOME)
assert mac.as_posix() == "/Users/tester/Library/Application Support/ssh-mcp/config.toml"
# Windows resolves under %APPDATA%, not ~/.config.
win = c.sidecar_path(
spec, platform="win32", environ={"APPDATA": "C:/Users/t/AppData/Roaming"}, home=_HOME
)
assert "ssh-mcp/config.toml" in win.as_posix()
assert "AppData/Roaming" in win.as_posix()
# POSIX honours XDG_CONFIG_HOME, else ~/.config.
xdg = c.sidecar_path(spec, platform="linux", environ={"XDG_CONFIG_HOME": "/cfg"}, home=_HOME)
assert xdg.as_posix() == "/cfg/ssh-mcp/config.toml"
default = c.sidecar_path(spec, platform="linux", environ={}, home=_HOME)
assert default.as_posix() == "/Users/tester/.config/ssh-mcp/config.toml"
def test_sidecar_path_none_for_non_sidecar_package():
assert c.sidecar_path(None) is None
assert c.sidecar_path(c.ServerSpec(package="nope")) is None
def test_sidecar_args_inert_when_file_exists():
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home=_HOME)
st = c.sidecar_status(
_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: p == real
)
assert st["exists"] is True
assert st["has_managed_args"] is True
assert st["args_inert"] is True
assert st["wrong_path"] is False
warns = c.sidecar_warnings(
_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: p == real
)
assert warns and "inert" in warns[0]
assert str(real) in warns[0]
def test_sidecar_args_live_when_file_absent():
st = c.sidecar_status(_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: False)
assert st["exists"] is False
assert st["args_inert"] is False
assert (
c.sidecar_warnings(_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: False)
== []
)
def test_sidecar_wrong_path_flag_on_macos():
# A TOML written at the README's ~/.config path is never read on macOS.
doc = c.sidecar_doc_path(c.SERVER_SPECS["ssh-mcp"], home=_HOME)
assert doc.as_posix() == "/Users/tester/.config/ssh-mcp/config.toml"
st = c.sidecar_status(
_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: p == doc
)
assert st["wrong_path"] is True
assert st["args_inert"] is False # the real file doesn't exist, so args still apply
warns = c.sidecar_warnings(
_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: p == doc
)
assert warns and "never loaded" in warns[0]
def test_sidecar_no_wrong_path_on_linux_where_paths_coincide():
# On Linux the real path and the doc path are the same, so a file there is
# correctly loaded — no wrong-path warning.
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="linux", environ={}, home=_HOME)
st = c.sidecar_status(
_SSH, platform="linux", environ={}, home=_HOME, exists=lambda p: p == real
)
assert st["wrong_path"] is False
assert st["args_inert"] is True
def test_count_toml_profiles():
assert c.count_toml_profiles("") == 0
assert c.count_toml_profiles("[server]\nhost='h'\n") == 1
text = "# comment\n[[hosts]]\nname='a'\n[[hosts]]\nname='b'\n[settings]\nx=1\n"
# two [[hosts]] share a top-level name -> one profile group; [settings] -> another.
assert c.count_toml_profiles(text) == 2
multi = "[prod]\nhost='p'\n[prod.auth]\nkey='k'\n[staging]\nhost='s'\n"
assert c.count_toml_profiles(multi) == 2
def test_unscoped_credential_warning_11():
data = dict(_SSH, env={"SSH_MCP_PASSWORD": "shared"})
# Single profile: no sharing concern.
assert c.unscoped_credential_warning(data, profile_count=1) is None
# Unknown count: claim nothing.
assert c.unscoped_credential_warning(data, profile_count=None) is None
# Multiple profiles + bare credential: warn.
w = c.unscoped_credential_warning(data, profile_count=3)
assert w and "every one of the 3 profiles" in w
# No bare credential: nothing to warn.
assert c.unscoped_credential_warning(_SSH, profile_count=3) is None
def test_sidecar_warnings_includes_credential_scoping():
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home=_HOME)
data = dict(_SSH, env={"SSH_MCP_PASSWORD": "shared"})
toml = "[prod]\nhost='p'\n[staging]\nhost='s'\n"
warns = c.sidecar_warnings(
data,
platform="darwin",
environ={},
home=_HOME,
exists=lambda p: p == real,
read_text=lambda p: toml,
)
assert any("profiles" in w for w in warns)
def test_sidecar_status_none_for_unknown_package():
assert c.sidecar_status({"command": "npx", "args": ["other"]}) is None
assert c.sidecar_warnings({"command": "npx", "args": ["other"]}) == []
# --------------------------------------------------------------------------- #
# Move to environment variable (issue #83)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"raw,expected",
[
("API_KEY", "API_KEY"),
("api-key", "API_KEY"),
("x.y z", "X_Y_Z"),
("2fa", "_2FA"),
("", "VAR"),
("***", "VAR"),
("clé", "CL_"), # non-ASCII becomes _
],
)
def test_sanitize_env_var_name(raw, expected):
assert c.sanitize_env_var_name(raw) == expected
def test_shell_export_lines_quote_safely():
lines = c.shell_export_lines("TOKEN", "ab'cd")
assert lines["posix"] == "export TOKEN='ab'\\''cd'"
assert lines["windows"] == 'setx TOKEN "ab\'cd"'
def test_can_move_gate_requires_secret_and_expanding_client():
desktop = c.Profile(label="d", path="/x/Claude/claude_desktop_config.json", config_exists=True)
code = c.Profile(label="c", path=Path.home() / ".claude.json", config_exists=True)
# real secret on an expanding client -> offer
assert c.can_move_value_to_env_ref("API_KEY", "ghp_abc", code) is True
assert c.can_move_value_to_env_ref("API_KEY", "ghp_abc", None) is True
# non-secret key -> no
assert c.can_move_value_to_env_ref("REGION", "us-east-1", code) is False
# already a reference -> no
assert c.can_move_value_to_env_ref("API_KEY", "${API_KEY}", code) is False
# non-expanding client (Claude Desktop) -> refuse even a real secret
assert c.can_move_value_to_env_ref("API_KEY", "ghp_abc", desktop) is False
def test_move_env_value_replaces_with_reference_and_returns_secret():
data = {"command": "x", "env": {"API_KEY": "ghp_secret", "REGION": "us"}}
conv = c.move_value_to_env_ref(data, field="env", key="API_KEY")
assert conv is not None
assert conv.var_name == "API_KEY"
assert conv.reference == "${API_KEY}"
assert conv.secret == "ghp_secret"
assert conv.data["env"]["API_KEY"] == "${API_KEY}"
# non-secret row untouched
assert conv.data["env"]["REGION"] == "us"
# input never mutated
assert data["env"]["API_KEY"] == "ghp_secret"
def test_move_derives_and_sanitises_var_name_from_key():
data = {"headers": {"x-api-key": "sekret"}}
conv = c.move_value_to_env_ref(data, field="headers", key="x-api-key")
assert conv.var_name == "X_API_KEY"
assert conv.data["headers"]["x-api-key"] == "${X_API_KEY}"
def test_move_honours_explicit_var_name():
data = {"env": {"tok": "sekret"}}
conv = c.move_value_to_env_ref(data, field="env", key="tok", var_name="GITHUB_TOKEN")
assert conv.reference == "${GITHUB_TOKEN}"
assert conv.data["env"]["tok"] == "${GITHUB_TOKEN}"
def test_move_args_by_index():
data = {"command": "x", "args": ["--token", "ghp_secret"]}
conv = c.move_value_to_env_ref(data, field="args", index=1, var_name="GH_TOKEN")
assert conv.secret == "ghp_secret"
assert conv.data["args"] == ["--token", "${GH_TOKEN}"]
def test_move_returns_none_on_missing_or_nonstring_or_already_ref():
data = {"env": {"API_KEY": "${API_KEY}", "N": 5}}
assert c.move_value_to_env_ref(data, field="env", key="ABSENT") is None
assert c.move_value_to_env_ref(data, field="env", key="N") is None # not a string
assert c.move_value_to_env_ref(data, field="env", key="API_KEY") is None # already a ref
assert c.move_value_to_env_ref({}, field="bogus") is None
assert c.move_value_to_env_ref({"args": ["a"]}, field="args", index=9) is None
def test_is_env_var_set():
assert c.is_env_var_set("FOO", {"FOO": "x"}) is True
assert c.is_env_var_set("FOO", {"FOO": ""}) is False
assert c.is_env_var_set("FOO", {}) is False
# --------------------------------------------------------------------------- #
# Args secret indices + suggested var name (issue #83, args surface)
# --------------------------------------------------------------------------- #
def test_secret_arg_indices_flags_token_and_flag_value():
args = ["--port", "8080", "ghp_deadbeef", "--token", "sk-abc", "--flag=val"]
idxs = c.secret_arg_indices(args)
assert 2 in idxs # ghp_ token prefix
assert 4 in idxs # value following --token
assert 1 not in idxs # 8080
assert 5 not in idxs # --flag=val inline pair
def test_secret_arg_indices_flags_embedded_url_credentials():
args = ["postgres://user:pass@host/db"]
assert c.secret_arg_indices(args) == [0]
def test_secret_arg_indices_excludes_existing_references():
args = ["--token", "${GH_TOKEN}"]
assert c.secret_arg_indices(args) == []
def test_suggested_env_var_for_arg_uses_preceding_flag():
args = ["--api-key", "sk-secret"]
assert c.suggested_env_var_for_arg(args, 1) == "API_KEY"
def test_suggested_env_var_for_arg_falls_back_when_no_flag():
args = ["ghp_secret"]
assert c.suggested_env_var_for_arg(args, 0) == "SECRET"
# --------------------------------------------------------------------------- #
# Referenced-variables readout + args->env relocation (issue #83, "both")
# --------------------------------------------------------------------------- #
def test_referenced_env_vars_dedupes_and_reports_status():
data = {
"command": "npx",
"args": ["--token", "${GH_TOKEN}", "${GH_TOKEN}"],
"env": {"API_KEY": "${API_KEY}", "REGION": "${REGION:-us-east-1}"},
}
usages = c.referenced_env_vars(data, environ={"GH_TOKEN": "x"})
by = {u.name: u for u in usages}
assert set(by) == {"GH_TOKEN", "API_KEY", "REGION"}
# GH_TOKEN appears only in args, deduped to one entry, set in env -> resolved
assert by["GH_TOKEN"].fields == ("args",)
assert by["GH_TOKEN"].resolved is True
# API_KEY not set, no default -> unresolved
assert by["API_KEY"].resolved is False
# REGION has a default -> resolved regardless of environment
assert by["REGION"].has_default is True
assert by["REGION"].resolved is True
# sorted by name
assert [u.name for u in usages] == sorted(u.name for u in usages)
def test_referenced_env_vars_empty_when_no_refs():
assert c.referenced_env_vars({"command": "npx", "args": ["-y", "pkg"]}) == []
def test_move_arg_to_env_block_removes_flag_and_value():
data = {"command": "x", "args": ["--api-key", "sk-secret", "run"], "env": {"KEEP": "1"}}
out = c.move_arg_to_env_block(data, 1)
assert out["args"] == ["run"] # flag + value both gone
assert out["env"]["API_KEY"] == "sk-secret"
assert out["env"]["KEEP"] == "1" # existing env preserved
# input not mutated
assert data["args"] == ["--api-key", "sk-secret", "run"]
def test_move_arg_to_env_block_bare_positional_keeps_no_flag():
data = {"command": "x", "args": ["ghp_secret", "serve"]}
out = c.move_arg_to_env_block(data, 0, var_name="GITHUB_TOKEN")
assert out["args"] == ["serve"]
assert out["env"] == {"GITHUB_TOKEN": "ghp_secret"}
def test_move_arg_to_env_block_none_on_bad_target():
assert c.move_arg_to_env_block({"args": ["a"]}, 5) is None
assert c.move_arg_to_env_block({"args": ["${REF}"]}, 0) is None # already a ref
assert c.move_arg_to_env_block({}, 0) is None
Reference in New Issue View Git Blame Copy Permalink