the_og
cfe05b4324
CI / Tests (py3.12 / windows-latest) (pull_request) Failing after 23s
CI / Lint (ruff) (pull_request) Successful in 6s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
CI / Catalog signature (pull_request) Successful in 7s
2717 lines
100 KiB
Python
2717 lines
100 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_catalog_entry_to_paste_json_seeds_env_required_keys():
|
|
# Regression: env_required is where the seed data actually keeps its
|
|
# secret VAR NAMES (postgres/github/notion/etc. all declare their secret
|
|
# here with config.env left empty) -- catalog_entry_to_paste_json must
|
|
# surface those names as blank env rows, not silently drop them.
|
|
entry = _minimal_catalog()["servers"][0]
|
|
entry["env_required"] = {"DATABASE_URI": ""}
|
|
result = c.catalog_entry_to_paste_json(entry)
|
|
assert result["widget"]["env"] == {"DATABASE_URI": ""}
|
|
|
|
|
|
def test_catalog_entry_to_paste_json_config_env_wins_over_env_required_default():
|
|
entry = _minimal_catalog()["servers"][0]
|
|
entry["config"]["env"] = {"GRAFANA_URL": "<GRAFANA_URL>"}
|
|
entry["env_required"] = {"GRAFANA_URL": "", "GRAFANA_SERVICE_ACCOUNT_TOKEN": ""}
|
|
result = c.catalog_entry_to_paste_json(entry)
|
|
assert result["widget"]["env"] == {
|
|
"GRAFANA_URL": "<GRAFANA_URL>",
|
|
"GRAFANA_SERVICE_ACCOUNT_TOKEN": "",
|
|
}
|
|
|
|
|
|
def test_catalog_entry_to_paste_json_real_postgres_entry_seeds_database_uri():
|
|
"""End-to-end regression against the actual shipped postgres entry,
|
|
which needs DATABASE_URI via env_required and has no config.env at
|
|
all -- this is exactly the shape that was silently dropping the env
|
|
field before catalog_entry_to_paste_json accounted for env_required."""
|
|
root = Path(__file__).resolve().parent.parent
|
|
raw = (root / "data" / "catalog.json").read_bytes()
|
|
data = c.load_catalog(raw)
|
|
entry = next(s for s in data["servers"] if s["id"] == "postgres")
|
|
result = c.catalog_entry_to_paste_json(entry)
|
|
assert result["postgres"]["env"] == {"DATABASE_URI": ""}
|
|
# And the focus-target helper now has something to point the user at.
|
|
assert c.first_unfilled_focus_target(result["postgres"]) == ("env", "DATABASE_URI")
|
|
|
|
|
|
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
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Browse-catalog dialog helpers (issue #10 phase 2)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
# --- catalog_category_group / CATALOG_CATEGORY_GROUPS --------------------- #
|
|
@pytest.mark.parametrize(
|
|
"category,expected_group",
|
|
[
|
|
("files", "Files & Dev"),
|
|
("dev", "Files & Dev"),
|
|
("code-hosting", "Files & Dev"),
|
|
("browser", "Files & Dev"),
|
|
("database", "Data"),
|
|
("data", "Data"),
|
|
("search", "Search & AI"),
|
|
("ai", "Search & AI"),
|
|
("cloud", "Cloud & Infra"),
|
|
("infra", "Cloud & Infra"),
|
|
("observability", "Cloud & Infra"),
|
|
("productivity", "Work"),
|
|
("communication", "Work"),
|
|
("crm", "Work"),
|
|
("finance", "Work"),
|
|
("design", "Work"),
|
|
("media", "Home & Personal"),
|
|
("smart-home", "Home & Personal"),
|
|
("personal", "Home & Personal"),
|
|
],
|
|
)
|
|
def test_catalog_category_group_maps_every_taxonomy_value(category, expected_group):
|
|
assert c.catalog_category_group(category) == expected_group
|
|
|
|
|
|
def test_catalog_category_group_unknown_falls_back_to_other():
|
|
assert c.catalog_category_group("some-future-category-nobody-has-seen-yet") == "Other"
|
|
assert c.catalog_category_group("") == "Other"
|
|
assert c.catalog_category_group(None) == "Other"
|
|
|
|
|
|
def test_catalog_category_group_is_case_insensitive():
|
|
assert c.catalog_category_group("Files") == "Files & Dev"
|
|
assert c.catalog_category_group("DATABASE") == "Data"
|
|
|
|
|
|
def test_shipped_catalog_categories_all_have_a_known_group():
|
|
"""Regression: every category actually used in data/catalog.json must
|
|
collapse to one of the 7 chips, never silently drop an entry."""
|
|
root = Path(__file__).resolve().parent.parent
|
|
raw = (root / "data" / "catalog.json").read_bytes()
|
|
data = c.load_catalog(raw)
|
|
for entry in data["servers"]:
|
|
group = c.catalog_category_group(entry["category"])
|
|
assert group in c.CATALOG_CATEGORY_CHIPS
|
|
|
|
|
|
# --- catalog_entry_matches_query / filter_catalog_entries ------------------ #
|
|
def _catalog_entries():
|
|
return [
|
|
{
|
|
"id": "filesystem",
|
|
"display": "Filesystem",
|
|
"description": "Read/write access to local directories you choose.",
|
|
"category": "files",
|
|
},
|
|
{
|
|
"id": "postgres",
|
|
"display": "Postgres MCP Pro",
|
|
"description": "Query and inspect a PostgreSQL database.",
|
|
"category": "database",
|
|
},
|
|
{
|
|
"id": "slack",
|
|
"display": "Slack",
|
|
"description": "Search messages and send messages from your assistant.",
|
|
"category": "communication",
|
|
},
|
|
]
|
|
|
|
|
|
def test_catalog_entry_matches_query_empty_matches_everything():
|
|
entries = _catalog_entries()
|
|
assert c.filter_catalog_entries(entries, "") == entries
|
|
assert c.filter_catalog_entries(entries, " ") == entries
|
|
|
|
|
|
def test_catalog_entry_matches_query_matches_id():
|
|
result = c.filter_catalog_entries(_catalog_entries(), "postgres")
|
|
assert [e["id"] for e in result] == ["postgres"]
|
|
|
|
|
|
def test_catalog_entry_matches_query_matches_display_case_insensitive():
|
|
result = c.filter_catalog_entries(_catalog_entries(), "SLACK")
|
|
assert [e["id"] for e in result] == ["slack"]
|
|
|
|
|
|
def test_catalog_entry_matches_query_matches_description():
|
|
result = c.filter_catalog_entries(_catalog_entries(), "PostgreSQL database")
|
|
assert [e["id"] for e in result] == ["postgres"]
|
|
|
|
|
|
def test_catalog_entry_matches_query_matches_category():
|
|
result = c.filter_catalog_entries(_catalog_entries(), "database")
|
|
assert [e["id"] for e in result] == ["postgres"]
|
|
|
|
|
|
def test_catalog_entry_matches_query_no_match_returns_empty():
|
|
assert c.filter_catalog_entries(_catalog_entries(), "kubernetes") == []
|
|
|
|
|
|
def test_catalog_entries_in_group_all_returns_everything():
|
|
entries = _catalog_entries()
|
|
assert c.catalog_entries_in_group(entries, "All") == entries
|
|
assert c.catalog_entries_in_group(entries, "") == entries
|
|
assert c.catalog_entries_in_group(entries, None) == entries
|
|
|
|
|
|
def test_catalog_entries_in_group_filters_by_collapsed_category():
|
|
result = c.catalog_entries_in_group(_catalog_entries(), "Data")
|
|
assert [e["id"] for e in result] == ["postgres"]
|
|
|
|
result = c.catalog_entries_in_group(_catalog_entries(), "Work")
|
|
assert [e["id"] for e in result] == ["slack"]
|
|
|
|
|
|
# --- format_freshness_hint -------------------------------------------------- #
|
|
def test_format_freshness_hint_none_returns_empty_string():
|
|
assert c.format_freshness_hint(None) == ""
|
|
assert c.format_freshness_hint("") == ""
|
|
|
|
|
|
def test_format_freshness_hint_unparseable_returns_empty_string():
|
|
assert c.format_freshness_hint("not-a-date") == ""
|
|
|
|
|
|
def test_format_freshness_hint_this_month():
|
|
assert (
|
|
c.format_freshness_hint("2026-07-01", today=c.date(2026, 7, 12))
|
|
== "Last updated this month"
|
|
)
|
|
|
|
|
|
def test_format_freshness_hint_one_month_singular():
|
|
assert (
|
|
c.format_freshness_hint("2026-06-01", today=c.date(2026, 7, 12))
|
|
== "Last updated 1 month ago"
|
|
)
|
|
|
|
|
|
def test_format_freshness_hint_months_ago():
|
|
# Exactly 14 full months elapsed, no day-of-month remainder to round off.
|
|
assert (
|
|
c.format_freshness_hint("2025-01-15", today=c.date(2026, 3, 15))
|
|
== "Last updated 14 months ago"
|
|
)
|
|
|
|
|
|
def test_format_freshness_hint_rounds_down_partial_month():
|
|
# 2025-05-16 -> 2026-07-12 is 13 full months, not 14: the 14th month
|
|
# would only complete on 2026-07-16.
|
|
assert (
|
|
c.format_freshness_hint("2025-05-16", today=c.date(2026, 7, 12))
|
|
== "Last updated 13 months ago"
|
|
)
|
|
|
|
|
|
def test_format_freshness_hint_years_ago():
|
|
assert (
|
|
c.format_freshness_hint("2024-01-01", today=c.date(2026, 7, 12))
|
|
== "Last updated 2 years ago"
|
|
)
|
|
|
|
|
|
def test_format_freshness_hint_23_months_stays_in_months_not_years():
|
|
# The switch to "N years ago" happens at 24 full months, not 12 -- the
|
|
# whole point of this hint is the granular "14 months ago" phrasing the
|
|
# design comment on #10 asked for, so 13-23 months must stay in months.
|
|
assert (
|
|
c.format_freshness_hint("2024-08-12", today=c.date(2026, 7, 12))
|
|
== "Last updated 23 months ago"
|
|
)
|
|
|
|
|
|
def test_format_freshness_hint_future_date_returns_empty_string():
|
|
# A last_release "in the future" relative to `today` is nonsensical --
|
|
# show nothing rather than a misleading negative offset.
|
|
assert c.format_freshness_hint("2027-01-01", today=c.date(2026, 7, 12)) == ""
|
|
|
|
|
|
# --- first_unfilled_focus_target -------------------------------------------- #
|
|
def test_first_unfilled_focus_target_prefers_placeholder_arg():
|
|
data = {
|
|
"command": "npx",
|
|
"args": ["-y", "server", "<ALLOWED_DIR>"],
|
|
"env": {"API_KEY": ""},
|
|
}
|
|
assert c.first_unfilled_focus_target(data) == ("args", 2)
|
|
|
|
|
|
def test_first_unfilled_focus_target_falls_back_to_first_blank_env():
|
|
data = {"command": "uvx", "args": ["mcp-grafana"], "env": {"GRAFANA_URL": ""}}
|
|
assert c.first_unfilled_focus_target(data) == ("env", "GRAFANA_URL")
|
|
|
|
|
|
def test_first_unfilled_focus_target_none_when_fully_filled():
|
|
data = {"command": "npx", "args": ["-y", "server"], "env": {"API_KEY": "sk-real-value"}}
|
|
assert c.first_unfilled_focus_target(data) is None
|
|
|
|
|
|
def test_first_unfilled_focus_target_none_for_config_with_no_env_or_args():
|
|
assert c.first_unfilled_focus_target({"command": "npx", "args": []}) is None
|
|
|
|
|
|
# --- load_bundled_catalog_entries ------------------------------------------- #
|
|
def test_load_bundled_catalog_entries_valid_signature(tmp_path, monkeypatch):
|
|
priv = Ed25519PrivateKey.generate()
|
|
pub = priv.public_key().public_bytes_raw()
|
|
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
|
|
|
|
raw, sig = _signed(_minimal_catalog(version=1), priv)
|
|
catalog_path = tmp_path / "catalog.json"
|
|
sig_path = tmp_path / "catalog.json.sig"
|
|
catalog_path.write_bytes(raw)
|
|
sig_path.write_bytes(sig)
|
|
|
|
entries = c.load_bundled_catalog_entries(catalog_path, sig_path)
|
|
assert len(entries) == 1
|
|
assert entries[0]["id"] == "widget"
|
|
|
|
|
|
def test_load_bundled_catalog_entries_tampered_payload_returns_empty_list(tmp_path, monkeypatch):
|
|
priv = Ed25519PrivateKey.generate()
|
|
pub = priv.public_key().public_bytes_raw()
|
|
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
|
|
|
|
raw, sig = _signed(_minimal_catalog(version=1), priv)
|
|
tampered = bytearray(raw)
|
|
tampered[-2] ^= 0xFF # flip a byte inside the trailing bytes, still valid-ish JSON shape
|
|
catalog_path = tmp_path / "catalog.json"
|
|
sig_path = tmp_path / "catalog.json.sig"
|
|
catalog_path.write_bytes(bytes(tampered))
|
|
sig_path.write_bytes(sig)
|
|
|
|
assert c.load_bundled_catalog_entries(catalog_path, sig_path) == []
|
|
|
|
|
|
def test_load_bundled_catalog_entries_wrong_key_returns_empty_list(tmp_path, monkeypatch):
|
|
priv = Ed25519PrivateKey.generate()
|
|
other_priv = Ed25519PrivateKey.generate()
|
|
other_pub = other_priv.public_key().public_bytes_raw()
|
|
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [other_pub])
|
|
|
|
raw, sig = _signed(_minimal_catalog(version=1), priv) # signed by the WRONG key
|
|
catalog_path = tmp_path / "catalog.json"
|
|
sig_path = tmp_path / "catalog.json.sig"
|
|
catalog_path.write_bytes(raw)
|
|
sig_path.write_bytes(sig)
|
|
|
|
assert c.load_bundled_catalog_entries(catalog_path, sig_path) == []
|
|
|
|
|
|
def test_load_bundled_catalog_entries_missing_files_returns_empty_list(tmp_path):
|
|
assert c.load_bundled_catalog_entries(tmp_path / "nope.json", tmp_path / "nope.json.sig") == []
|
|
|
|
|
|
def test_load_bundled_catalog_entries_missing_sig_returns_empty_list(tmp_path, monkeypatch):
|
|
priv = Ed25519PrivateKey.generate()
|
|
pub = priv.public_key().public_bytes_raw()
|
|
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
|
|
|
|
raw, _sig = _signed(_minimal_catalog(version=1), priv)
|
|
catalog_path = tmp_path / "catalog.json"
|
|
catalog_path.write_bytes(raw)
|
|
missing_sig_path = tmp_path / "catalog.json.sig" # never written
|
|
|
|
assert c.load_bundled_catalog_entries(catalog_path, missing_sig_path) == []
|
|
|
|
|
|
def test_load_bundled_catalog_entries_invalid_but_signed_returns_empty_list(tmp_path, monkeypatch):
|
|
"""A payload that verifies but fails validate_catalog() (disallowed
|
|
command) must still come back empty -- signing is necessary, not
|
|
sufficient."""
|
|
priv = Ed25519PrivateKey.generate()
|
|
pub = priv.public_key().public_bytes_raw()
|
|
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
|
|
|
|
raw, sig = _signed(_catalog_with({"config": {"command": "bash", "args": []}}), priv)
|
|
catalog_path = tmp_path / "catalog.json"
|
|
sig_path = tmp_path / "catalog.json.sig"
|
|
catalog_path.write_bytes(raw)
|
|
sig_path.write_bytes(sig)
|
|
|
|
assert c.load_bundled_catalog_entries(catalog_path, sig_path) == []
|
|
|
|
|
|
def test_load_bundled_catalog_entries_real_shipped_catalog():
|
|
"""End-to-end regression against the actual bundled data/catalog.json +
|
|
.sig, using the real CATALOG_PUBKEYS (no monkeypatch) -- this is what
|
|
the Browse dialog actually calls on startup."""
|
|
root = Path(__file__).resolve().parent.parent
|
|
entries = c.load_bundled_catalog_entries(
|
|
root / "data" / "catalog.json", root / "data" / "catalog.json.sig"
|
|
)
|
|
assert len(entries) == 19
|
|
assert {e["id"] for e in entries} >= {"filesystem", "github", "slack", "postgres"}
|