the_og
f9752211a2
Includes the required regression case: rewrite a file with different content, force the original mtime back via os.utime, and assert the fingerprint still differs (because size changed).
678 lines
24 KiB
Python
678 lines
24 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 os
|
|
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.
|
|
# 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
|