AJ
6ab4789a70
The diff preview in RestoreDialog was serializing the full config dict,
exposing secret env values and token args in cleartext on a pasteable surface.
- Add _redact_server_data / _redact_servers_block helpers that apply
redact_args to args and mask env values for is_secret_key() keys
- Rewrite backup_diff to compare only {mcpServers, _disabledMcpServers}
sections (sanitized), not the whole file — also avoids double-serializing
multi-MB ~/.claude.json for a servers-only diff
- Add clarifying comment in _restore_from_backup about why full_config
is the right base after confirm-discard
- Add test: backup_diff with secret args/env → MASK in output, raw values absent
- Add test: restore_backup restores _disabledMcpServers correctly
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
503 lines
17 KiB
Python
503 lines
17 KiB
Python
"""Pytest port of the original test_core.py script (same 23 behaviours, now
|
|
proper test functions with tmp_path/monkeypatch fixtures)."""
|
|
|
|
import json
|
|
import sys
|
|
|
|
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
|
|
r = c.spawn_test(
|
|
{"command": sys.executable, "args": ["-c", "raise SystemExit(1)"]}, timeout=0.3
|
|
)
|
|
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)
|
|
r = c.spawn_test({"command": sys.executable, "args": ["-c", "pass"]}, timeout=0.5)
|
|
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=0.5,
|
|
)
|
|
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", {})
|