BCC Agent
38f14deeff
CI / Lint (ruff) (pull_request) Successful in 7s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 23s
CI / Catalog signature (pull_request) Successful in 7s
Fixes findings 2, 3, 4, 6, 7 from the issue #68 adversarial review. - Finding 2: config.env was type-checked only. Add CATALOG_DENIED_ENV_KEYS (case-insensitive) for interpreter/loader-override keys (NODE_OPTIONS, PYTHONPATH, LD_PRELOAD, ...), apply the ASCII check and the existing secret-value check to env keys/values, and require env values to be empty or a single <PLACEHOLDER> token. - Finding 3: version pinning was only checked by catalog_review.py (which never runs on the signing path per finding 1). Move enforcement into _validate_catalog_config: npm/uvx specs must carry @version or ==version (scoped names handled), docker images must have an explicit non-latest tag. Only the first plausible package-spec token is checked, so flags, <PLACEHOLDER>s, and docker subcommands/flags don't trip it. All 19 real catalog entries still validate clean. - Finding 4: the CI catalog-signature gate imported bcc_core from the PR branch and trusted whatever CATALOG_PUBKEYS said there, so a PR changing both catalog.json and CATALOG_PUBKEYS (with a matching signature) went green. ci.yml now hardcodes the expected base64 pubkey and asserts bcc_core.CATALOG_PUBKEYS matches it before verifying the signature. NOTE: the maintainer is planning to rotate this key -- update EXPECTED_CATALOG_PUBKEY_B64 in ci.yml as its own reviewed change when that happens, never bundled with a catalog content change. - Finding 6: resolve_catalog's anti-rollback/anti-freeze guards sat behind `if best_version >= 0`, so the first verified candidate was accepted unconditionally and the anti-freeze anchor drifted with each accepted candidate instead of staying fixed. The cap is now measured against the bundled catalog's version specifically (the trust anchor baked into the binary), regardless of evaluation order; bundled wins version ties; and a new pure `floor` parameter lets a future caller pass a persisted accepted-version floor. - Finding 7: catalog id is now constrained to ^[a-z0-9][a-z0-9._-]{0,63}$. Tests: fixed _minimal_catalog to use a pinned package (was enshrining finding 3), rewrote the env-passthrough test to prove the validation boundary instead of asserting env passes through unchecked, and reordered test_resolve_catalog_rejects_absurd_version_jump so it actually exercises the first-candidate path. Added positive/negative tests for every new rule. Manually verified each new check by commenting it out and confirming the guarding test goes red, then restoring it.
2373 lines
87 KiB
Python
2373 lines
87 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 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
|