"""Pytest port of the original test_core.py script (same 23 behaviours, now proper test functions with tmp_path/monkeypatch fixtures).""" import dataclasses 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 " 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": ""}, } } ) 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": ""}, } } ) 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 positional # must not itself get mistaken for an unpinned package. data = _catalog_with( { "config": { "command": "uvx", "args": ["mcp-server-git@2026.7.10", "--repository", ""], } } ) 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": "Verified"}) 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 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": ""} result = c.catalog_entry_to_paste_json(entry) assert result["widget"]["env"] == {"GRAFANA_URL": ""} catalog = _minimal_catalog() catalog["servers"][0]["config"]["env"] = {"GRAFANA_URL": ""} assert c.validate_catalog(catalog) == [] malicious = _minimal_catalog() malicious["servers"][0]["config"]["env"] = {"NODE_OPTIONS": "--require /tmp/payload.js"} assert c.validate_catalog(malicious) != [] def test_config_has_unfilled_placeholders_true_for_token(): cfg = {"command": "npx", "args": ["-y", "server", ""]} 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": ""}} assert c.config_has_unfilled_placeholders(cfg) is True # --------------------------------------------------------------------------- # # #72 -- a server value that isn't a JSON object must not take the load down # --------------------------------------------------------------------------- # @pytest.mark.parametrize("bad", ["not-a-dict", 123, ["a", "b"], None, True, 1.5]) def test_extract_servers_survives_non_dict_server_value(bad): entries = c.extract_servers({"mcpServers": {"foo": bad}}) assert len(entries) == 1 assert entries[0].name == "foo" assert entries[0].data == {} assert entries[0].malformed is True assert entries[0].raw == bad def test_extract_servers_marks_only_the_bad_entry(): cfg = {"mcpServers": {"good": {"command": "npx"}, "bad": "oops"}} by_name = {e.name: e for e in c.extract_servers(cfg)} assert by_name["good"].malformed is False assert by_name["good"].data == {"command": "npx"} assert by_name["bad"].malformed is True def test_extract_servers_handles_malformed_disabled_entry(): entries = c.extract_servers({c.DISABLED_KEY: {"parked": ["nope"]}}) assert entries[0].enabled is False assert entries[0].malformed is True def test_malformed_entry_round_trips_through_save_unchanged(): """The cardinal rule: never silently delete what the user had on disk.""" cfg = {"mcpServers": {"good": {"command": "npx"}, "bad": "oops"}} servers = c.extract_servers(cfg) out = c.apply_servers(dict(cfg), servers) assert out["mcpServers"]["bad"] == "oops" assert out["mcpServers"]["good"] == {"command": "npx"} def test_editing_a_malformed_entry_retires_the_raw_value(): entry = c.extract_servers({"mcpServers": {"bad": "oops"}})[0] entry.set_data({"command": "npx"}) assert entry.malformed is False assert entry.config_value() == {"command": "npx"} assert c.apply_servers({}, [entry])["mcpServers"]["bad"] == {"command": "npx"} def test_lint_reports_the_malformed_entry_by_name(): servers = c.extract_servers({"mcpServers": {"bad": "oops"}}) warnings = c.lint_servers(servers) assert len(warnings) == 1 assert "'bad'" in warnings[0] assert "not an object" in warnings[0] assert "str" in warnings[0] def test_lint_still_reports_normal_warnings_alongside_malformed(): cfg = {"mcpServers": {"bad": "oops", "sloppy": {"command": "npx", "args": "one two"}}} warnings = c.lint_servers(c.extract_servers(cfg)) assert any("not an object" in w for w in warnings) assert any("'args' should be a list" in w for w in warnings) # --------------------------------------------------------------------------- # # #73 -- the stale-file merge must not discard BCC-authored keys # --------------------------------------------------------------------------- # def test_carry_owned_keys_moves_sets_onto_the_reloaded_config(): local = {"mcpServers": {}, c.SETS_KEY: {"work": ["a", "b"]}} fresh = {"mcpServers": {"external": {"command": "npx"}}} contested = c.carry_owned_keys(local, fresh) assert contested == [] assert fresh[c.SETS_KEY] == {"work": ["a", "b"]} assert fresh["mcpServers"] == {"external": {"command": "npx"}} def test_carry_owned_keys_reports_a_genuine_conflict(): local = {c.SETS_KEY: {"work": ["a"]}} fresh = {c.SETS_KEY: {"work": ["a", "b"]}} assert c.carry_owned_keys(local, fresh) == [c.SETS_KEY] assert fresh[c.SETS_KEY] == {"work": ["a"]} # local wins: BCC owns the key def test_carry_owned_keys_is_quiet_when_both_sides_agree(): local = {c.SETS_KEY: {"work": ["a"]}} fresh = {c.SETS_KEY: {"work": ["a"]}} assert c.carry_owned_keys(local, fresh) == [] def test_carry_owned_keys_leaves_disk_alone_when_absent_locally(): """Can't distinguish 'deleted my last set' from 'never had sets'; keep theirs.""" fresh = {c.SETS_KEY: {"remote": ["a"]}} assert c.carry_owned_keys({}, fresh) == [] assert fresh[c.SETS_KEY] == {"remote": ["a"]} def test_carry_owned_keys_deep_copies_so_later_edits_do_not_leak(): local = {c.SETS_KEY: {"work": ["a"]}} fresh = {} c.carry_owned_keys(local, fresh) local[c.SETS_KEY]["work"].append("b") assert fresh[c.SETS_KEY] == {"work": ["a"]} def test_merge_flow_preserves_sets_and_external_servers(tmp_path): """End-to-end shape of the Merge & save path that lost sets in #73.""" path = tmp_path / "claude.json" path.write_text(json.dumps({"mcpServers": {"old": {"command": "old"}}})) # BCC loads, user saves a named set and edits servers in memory. local = c.load_config(path) servers = c.extract_servers(local) c.save_server_set(local, "work", servers) # Something else rewrites the file underneath us. path.write_text(json.dumps({"mcpServers": {"external": {"command": "new"}}, "other": 1})) # Merge & save: reload disk, carry BCC keys, re-apply the user's servers. fresh = c.load_config(path) c.carry_owned_keys(local, fresh) c.apply_servers(fresh, servers) c.write_config(path, fresh) saved = c.load_config(path) assert saved[c.SETS_KEY] == {"work": ["old"]} # the set survived assert saved["other"] == 1 # unrelated external key preserved assert "old" in saved["mcpServers"] # user's servers re-applied def test_null_server_value_is_malformed_not_mistaken_for_absent(): """`{"mcpServers": {"foo": null}}` is legal JSON and a real malformed case, so None must not double as the 'nothing here' sentinel.""" entry = c.extract_servers({"mcpServers": {"foo": None}})[0] assert entry.malformed is True assert entry.raw is None assert c.apply_servers({}, [entry])["mcpServers"]["foo"] is None def test_a_normal_entry_is_not_malformed(): entry = c.extract_servers({"mcpServers": {"foo": {"command": "npx"}}})[0] assert entry.malformed is False assert entry.raw is c.NO_RAW # --------------------------------------------------------------------------- # # #75 -- theming # --------------------------------------------------------------------------- # @pytest.mark.parametrize( "setting,system_dark,expected", [ (c.THEME_DARK, False, "dark"), (c.THEME_DARK, True, "dark"), (c.THEME_LIGHT, False, "light"), (c.THEME_LIGHT, True, "light"), (c.THEME_SYSTEM, True, "dark"), (c.THEME_SYSTEM, False, "light"), ], ) def test_resolve_theme_covers_every_setting_and_appearance(setting, system_dark, expected): assert c.resolve_theme(setting, system_dark) == expected @pytest.mark.parametrize("junk", ["", "solarized", None, "DARK", 3]) def test_resolve_theme_falls_back_to_following_the_system(junk): """A hand-edited or future QSettings value should follow the desktop, not pin a fixed theme.""" assert c.resolve_theme(junk, True) == "dark" assert c.resolve_theme(junk, False) == "light" def test_palette_for_known_names(): assert c.palette_for("dark") is c.DARK_PALETTE assert c.palette_for("light") is c.LIGHT_PALETTE def test_palette_for_unknown_name_falls_back_to_dark(): assert c.palette_for("chartreuse") is c.DARK_PALETTE def test_dark_palette_is_unchanged_from_the_shipped_look(): """v1.3.0 shipped these exact colours; adding a light theme must not quietly restyle the dark one.""" p = c.DARK_PALETTE assert (p.accent, p.bg, p.panel, p.panel_2) == ("#f97316", "#1b1d23", "#23262e", "#2b2f39") assert (p.text, p.muted, p.border) == ("#e7e9ee", "#9aa0ad", "#3a3f4b") assert (p.good, p.bad, p.warn, p.remote) == ("#4ade80", "#f87171", "#fbbf24", "#60a5fa") assert (p.on_accent, p.disabled_bg, p.mono_bg) == ("#1a1205", "#202229", "#16181d") def test_both_palettes_define_every_slot(): """A missing slot should fail here rather than render a broken window.""" for pal in (c.DARK_PALETTE, c.LIGHT_PALETTE): for f in dataclasses.fields(c.Palette): value = getattr(pal, f.name) assert value, f"{pal.name}.{f.name} is empty" if f.name != "name": assert re.fullmatch(r"#[0-9a-fA-F]{6}", value), f"{pal.name}.{f.name}={value!r}" @pytest.mark.parametrize("pal_name", ["dark", "light"]) @pytest.mark.parametrize("slot", ["text", "muted", "good", "bad", "warn", "remote", "accent"]) def test_palette_meets_contrast_on_panel(pal_name, slot): """Every colour drawn as text/glyph must clear WCAG AA (4.5:1) against the surface it sits on. The light palette's semantic colours are NOT the dark ones lightened -- #4ade80 sits near 1.7:1 on white -- so this guards against someone 'harmonising' them back toward the dark hues.""" pal = c.palette_for(pal_name) assert c.contrast_ratio(getattr(pal, slot), pal.panel) >= 4.5 @pytest.mark.parametrize("pal_name", ["dark", "light"]) def test_on_accent_is_legible_against_the_accent_fill(pal_name): """Primary buttons and selected rows draw on_accent on top of accent.""" pal = c.palette_for(pal_name) assert c.contrast_ratio(pal.on_accent, pal.accent) >= 4.5 def test_contrast_ratio_endpoints(): assert c.contrast_ratio("#000000", "#ffffff") == pytest.approx(21.0, abs=0.01) assert c.contrast_ratio("#123456", "#123456") == pytest.approx(1.0, abs=0.001) assert c.contrast_ratio("#ffffff", "#000000") == pytest.approx(21.0, abs=0.01) def test_relative_luminance_extremes(): assert c.relative_luminance("#000000") == pytest.approx(0.0) assert c.relative_luminance("#ffffff") == pytest.approx(1.0) def test_stylesheet_builder_has_no_hardcoded_colours(): """Every colour in the QSS must come from the palette. Three near-black literals used to be inlined here (#1a1205, #202229, #16181d). Harmless with one theme; with two, they silently render dark chrome on a light window. Reads the source rather than importing bcc, which needs PySide6. """ src = (Path(__file__).resolve().parent.parent / "bcc.py").read_text(encoding="utf-8") start = src.index("def build_stylesheet") body = src[start : src.index("def apply_palette")] assert re.findall(r"#[0-9a-fA-F]{6}", body) == [] def test_every_palette_slot_is_consumed(): """A slot added to Palette but never wired up is dead weight. Checks for `p.` anywhere in bcc.py, which covers both the QSS and apply_palette's global bindings -- not every slot belongs in the stylesheet (`good` and `remote` feed the inline status dots via STATUS_COLORS/HEALTH_COLORS, never the QSS). This won't catch a slot bound to a global that nothing then uses; it does catch the common mistake of extending the dataclass and forgetting to plumb it through. """ src = (Path(__file__).resolve().parent.parent / "bcc.py").read_text(encoding="utf-8") for f in dataclasses.fields(c.Palette): if f.name == "name": continue assert f"p.{f.name}" in src, f"palette slot {f.name!r} is never consumed" # --------------------------------------------------------------------------- # # #78/#79 -- update notice: when to show it, and what it says # --------------------------------------------------------------------------- # def test_update_notice_when_a_newer_release_exists(): n = c.update_notice("1.2.0", {"version": "v1.3.0", "url": "https://example.test/rel"}) assert n is not None assert n["version"] == "v1.3.0" assert n["url"] == "https://example.test/rel" assert "1.3.0" in n["text"] and "1.2.0" in n["text"] def test_update_notice_is_silent_when_current(): assert c.update_notice("1.3.0", {"version": "v1.3.0"}) is None assert c.update_notice("1.4.0", {"version": "v1.3.0"}) is None @pytest.mark.parametrize("bad", [None, {}, {"version": ""}, {"version": None}, {"version": 3}, []]) def test_update_notice_is_silent_on_a_failed_or_malformed_check(bad): """fetch_latest_release returns None on any failure; a half-formed payload must not produce a notice pointing at nothing.""" assert c.update_notice("1.0.0", bad) is None def test_update_notice_falls_back_to_the_releases_page_without_a_url(): n = c.update_notice("1.0.0", {"version": "v2.0.0"}) assert n["url"] == c.RELEASES_URL def test_update_notice_names_no_menu_path(): """The old status-line text said 'Help > About to view it', which is wrong on macOS -- Qt moves the About action into the application menu (#79). The notice carries its own action, so it must not describe a menu path.""" n = c.update_notice("1.0.0", {"version": "v2.0.0"}) lowered = n["text"].lower() for phrase in ("help", "about", "menu", "▸", ">"): assert phrase not in lowered, f"notice text should not reference {phrase!r}" def test_update_notice_handles_the_v_prefix_consistently(): assert c.update_notice("1.2.0", {"version": "1.3.0"}) is not None assert c.update_notice("v1.2.0", {"version": "v1.3.0"}) is not None assert c.update_notice("1.3.0", {"version": "v1.3.0"}) is None def test_update_notice_renders_both_versions_the_same_way(): """Tags carry a 'v' prefix, __version__ doesn't -- don't show both forms in one sentence.""" n = c.update_notice("1.2.0", {"version": "v1.3.0"}) assert "v1.3.0" not in n["text"] assert "1.3.0" in n["text"] and "1.2.0" in n["text"] # the machine-readable field keeps the real tag assert n["version"] == "v1.3.0"