the_og/better-claude-config
1
0
Fork 0
Code Issues 8 Pull Requests 3 Actions Packages Projects Releases 4 Wiki Activity
Files
668fb903d03d89344c1daebe60dc49dfb93208d5
better-claude-config/tests/test_core.py
T
Cowork Supervisor 668fb903d0
CI / Lint (ruff) (pull_request) Successful in 6s
Details
CI / Tests (py3.10) (pull_request) Successful in 8s
Details
CI / Tests (py3.12) (pull_request) Successful in 7s
Details
feat: server search/filter + test-all health column (#27, #28)
2026-07-08 11:53:31 -04:00

1183 lines
42 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 json
import os
import sys
import urllib.error
import urllib.request
import pytest
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())
# --------------------------------------------------------------------------- #
# 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)]) == []
# --------------------------------------------------------------------------- #
# 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_python3():
assert c.check_dependency({"command": "python3"})["status"] == "ok"
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
# --------------------------------------------------------------------------- #
# 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
# --------------------------------------------------------------------------- #
# 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
# --------------------------------------------------------------------------- #
# 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" then open -a Claude, in that order."""
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
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] == "pkill":
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 raising OSError (binary missing) must be swallowed, not propagated --
relaunch is still attempted."""
def fake_run(cmd, **kwargs):
if cmd[0] == "pkill":
raise OSError("pkill 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_windows_commands(monkeypatch):
"""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):
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_linux_commands(monkeypatch):
"""Linux: pkill claude, then relaunch via a detached Popen (no waiting)."""
run_calls = []
popen_calls = []
def fake_run(cmd, **kwargs):
run_calls.append(cmd)
return _FakeCompletedProcess(returncode=0)
class _FakePopen:
def __init__(self, cmd, **kwargs):
popen_calls.append((cmd, kwargs))
monkeypatch.setattr(c.sys, "platform", "linux")
monkeypatch.setattr(c.subprocess, "run", fake_run)
monkeypatch.setattr(c.subprocess, "Popen", _FakePopen)
result = c.restart_claude_desktop()
assert result.success
assert run_calls == [["pkill", "claude"]]
cmd, kwargs = popen_calls[0]
assert cmd == ["claude"]
assert kwargs.get("start_new_session") is True
def test_restart_claude_desktop_linux_popen_failure_reported(monkeypatch):
def fake_run(cmd, **kwargs):
return _FakeCompletedProcess(returncode=0)
def fake_popen(cmd, **kwargs):
raise OSError("no such file or directory")
monkeypatch.setattr(c.sys, "platform", "linux")
monkeypatch.setattr(c.subprocess, "run", fake_run)
monkeypatch.setattr(c.subprocess, "Popen", fake_popen)
result = c.restart_claude_desktop()
assert not result.success
assert "Claude Desktop" in result.detail
# --------------------------------------------------------------------------- #
# 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.
assert c.__version__ == "1.2.0"
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
Reference in New Issue View Git Blame Copy Permalink