the_og/better-claude-config
1
0
Fork 0
Code Issues 8 Pull Requests 2 Actions Packages Projects Releases 5 Wiki Activity
Files
694439b6f3935c15e05b011dc39bf34c34dc2af9
better-claude-config/tests/test_core.py
T
the_ogandClaude Opus 4.8 694439b6f3
CI / Lint (ruff) (pull_request) Successful in 11s
Details
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 30s
Details
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 19s
Details
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 20s
Details
CI / Catalog signature (pull_request) Successful in 13s
Details
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 18s
Details
feat(#83): variables readout, two clear move actions, args->env relocation 
Field testing showed the feature was doing the right thing but describing it
wrong, and left the moved variable invisible. Reworked per that feedback:

Naming. "Move to environment variable" read like "move into the Environment
variables table"; it actually creates a ${VAR} reference to the shell/OS
environment. Renamed the action to "Replace with a ${VAR} reference (out of
file)…" everywhere, and the dialog now says plainly the secret goes to your
shell environment -- not this file, not the table below.

Visibility (the real gap). A lone ${SECRET_KEY} with nothing saying whether
it's wired up isn't much better than a mystery. New Variables… button opens
ReferencedVarsDialog: every ${VAR} the loaded server references, each with
✓ set / ✓ default / ✗ not set and the exact export/setx line to set it.
Backed by pure core.referenced_env_vars (dedupes across fields, resolves
against a given environment or a default).

Two actions on an args secret, because the user may want either:
  * "Replace with a ${VAR} reference (out of file)…" -- secret leaves the
    file (needs a client that expands refs; disabled with reason on Desktop).
  * "Move into Environment variables (kept in this config)…" -- relocates the
    arg into the env block where it's visible and editable. Works on any
    client (no ${VAR} needed). core.move_arg_to_env_block drops the flag+value
    and sets env[VAR]; the dialog warns it changes how the server launches.

Both args actions run through ServerEditor (moving into env touches args AND
the env table), which reloads the form from the transformed data.

Tests: +10 core (referenced_env_vars dedupe/status/default; move_arg_to_env_block
flag+value removal, bare positional, None on bad target). 488 passed, ruff
clean. New dialogs/menus are GUI, untestable in CI as before.

Refs #83

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EKwBecy6N83jnqQmw8ezwE
2026-08-04 04:25:28 +00:00

3234 lines
123 KiB
Python
Raw Blame History

"""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 <PLACEHOLDER>" in p for p in problems)
def test_validate_catalog_accepts_empty_and_placeholder_env_values():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FOO": "", "BAR_URL": "<BAR_URL>"},
}
}
)
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_non_ascii_env_key():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FÖO": ""},
}
}
)
problems = c.validate_catalog(data)
assert any("config.env key" in p and "ASCII" in p for p in problems)
def test_validate_catalog_rejects_non_ascii_env_value():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FOO": "<Bäd>"},
}
}
)
problems = c.validate_catalog(data)
assert any("config.env value" in p and "ASCII" in p for p in problems)
def test_validate_catalog_rejects_secret_looking_env_value():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"SOME_TOKEN": "ghp_abcdef1234567890"},
}
}
)
problems = c.validate_catalog(data)
assert any("real secret value" in p for p in problems)
def test_validate_catalog_rejects_node_options_env_walking_past_allowlist():
# The exact reproduction from issue #68 finding 2: an allowlisted
# `npx` command carrying NODE_OPTIONS in env, which previously passed
# validation and would have flowed straight into the executed
# subprocess via catalog_entry_to_paste_json().
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"NODE_OPTIONS": "--require /tmp/payload.js"},
}
}
)
problems = c.validate_catalog(data)
assert problems != []
# --- validate_catalog: version pinning (issue #68 finding 3) --------------- #
def test_validate_catalog_rejects_unpinned_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "widget-mcp"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_rejects_unpinned_scoped_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "@scope/pkg"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_accepts_pinned_scoped_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_unpinned_uvx_package():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_accepts_uvx_at_version_pin():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool@1.0.0"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_accepts_uvx_double_equals_pin():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool==1.0.0"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_docker_latest_tag():
data = _catalog_with({"config": {"command": "docker", "args": ["run", "some/image:latest"]}})
problems = c.validate_catalog(data)
assert any("'latest'" in p for p in problems)
def test_validate_catalog_rejects_docker_untagged_image():
data = _catalog_with({"config": {"command": "docker", "args": ["run", "some/image"]}})
problems = c.validate_catalog(data)
assert any("no explicit tag" in p for p in problems)
def test_validate_catalog_accepts_pinned_docker_image_with_flags():
data = _catalog_with(
{
"config": {
"command": "docker",
"args": ["run", "-i", "--rm", "-e", "SOME_TOKEN", "some/image:1.2.3"],
}
}
)
assert c.validate_catalog(data) == []
def test_validate_catalog_does_not_pin_check_placeholders_flags_or_subcommand():
# A pinned uvx spec followed by flags and a <PLACEHOLDER> positional
# must not itself get mistaken for an unpinned package.
data = _catalog_with(
{
"config": {
"command": "uvx",
"args": ["mcp-server-git@2026.7.10", "--repository", "<REPO_PATH>"],
}
}
)
assert c.validate_catalog(data) == []
# --- validate_catalog: id constraint (issue #68 finding 7) ----------------- #
def test_validate_catalog_rejects_id_with_markup():
data = _catalog_with({"id": "<b>Verified</b>"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_rejects_id_with_uppercase():
data = _catalog_with({"id": "Widget"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_rejects_id_starting_with_dash():
data = _catalog_with({"id": "-widget"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_accepts_valid_slug_id():
data = _catalog_with({"id": "widget-2.thing-ok"})
assert c.validate_catalog(data) == []
# --- resolve_catalog -------------------------------------------------------- #
def test_resolve_catalog_nothing_available_returns_empty_dict():
assert c.resolve_catalog(None, None, None) == {}
def test_resolve_catalog_prefers_highest_verified_version(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
bundled = _signed(_minimal_catalog(version=1), priv)
cached = _signed(_minimal_catalog(version=2), priv)
remote = _signed(_minimal_catalog(version=3), priv)
result = c.resolve_catalog(bundled, cached, remote)
assert c.catalog_version(result) == 3
def test_resolve_catalog_rejects_unsigned_bundled_catalog(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
# Bundled claims a very high version but is NOT signed by a trusted key
# -- it must get no implicit trust just for being the local copy.
malicious_raw = json.dumps(_minimal_catalog(version=100)).encode("utf-8")
bundled = (malicious_raw, b"totally-not-a-signature")
remote = _signed(_minimal_catalog(version=3), priv)
result = c.resolve_catalog(bundled, None, remote)
assert c.catalog_version(result) == 3
def test_resolve_catalog_rejects_rolled_back_version(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
cached = _signed(_minimal_catalog(version=5), priv)
rolled_back_remote = _signed(_minimal_catalog(version=2), priv)
result = c.resolve_catalog(None, cached, rolled_back_remote)
assert c.catalog_version(result) == 5
def test_resolve_catalog_rejects_absurd_version_jump(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
# The freeze attempt goes FIRST (as `cached`), the legitimate catalog
# SECOND (as `remote`) -- on purpose. Putting the good catalog first
# (as an earlier version of this test did) never exercises the
# vulnerable path: the old implementation only guarded a candidate
# against "the best accepted so far," so whichever candidate was
# evaluated FIRST got in unconditionally, uncapped. Ordering the freeze
# attempt first is what actually proves the cap holds regardless of
# evaluation order.
freeze_attempt = _signed(_minimal_catalog(version=999999), priv)
good = _signed(_minimal_catalog(version=5), priv)
result = c.resolve_catalog(None, freeze_attempt, good)
assert c.catalog_version(result) == 5
def test_resolve_catalog_caps_first_and_only_candidate(monkeypatch):
# issue #68 finding 6: with no bundled catalog to anchor against, a
# signed catalog claiming an absurd version must still be capped even
# when it is the ONLY candidate resolve_catalog() ever sees -- there is
# no "best so far" for it to be compared against, so the cap has to
# apply unconditionally, not "once something else has already landed."
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
freeze_attempt = _signed(_minimal_catalog(version=999999999), priv)
result = c.resolve_catalog(None, None, freeze_attempt)
assert result == {}
def test_resolve_catalog_anchors_cap_to_bundled_not_a_chained_best(monkeypatch):
# Anti-freeze must be measured against the BUNDLED version specifically,
# not against "whatever the best-so-far happens to be after each
# candidate is accepted" -- a chained anchor lets each accepted
# candidate ratchet the allowed ceiling upward, so a legitimate
# moderate bump (cached) plus a second, much larger jump (remote) can
# each individually look "within _CATALOG_MAX_VERSION_JUMP of the
# previous one" while remote is nowhere near bundled's version.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
bundled = _signed(_minimal_catalog(version=2), priv)
cached = _signed(_minimal_catalog(version=1000), priv) # within 1000 of bundled
remote = _signed(_minimal_catalog(version=1900), priv) # within 1000 of cached,
# NOT of bundled
result = c.resolve_catalog(bundled, cached, remote)
assert c.catalog_version(result) == 1000
def test_resolve_catalog_prefers_bundled_on_version_tie(monkeypatch):
# issue #68 finding 6: on a tie the LAST candidate evaluated used to
# win, so remote silently beat bundled at equal version. Bundled --
# the copy frozen into the binary -- must win ties.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
bundled_data = _minimal_catalog(version=5)
bundled = _signed(bundled_data, priv)
remote_data = _minimal_catalog(version=5)
remote_data["servers"][0]["display"] = "Remote Impostor"
remote = _signed(remote_data, priv)
result = c.resolve_catalog(bundled, None, remote)
assert c.catalog_version(result) == 5
assert result["servers"][0]["display"] == "Widget"
def test_resolve_catalog_floor_rejects_below_persisted_version(monkeypatch):
# `floor` is a pure parameter: the caller (eventually the GUI, from
# persisted storage) can pass a previously-accepted version, and
# nothing below it may be accepted even with no bundled catalog to
# anchor against.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
stale = _signed(_minimal_catalog(version=3), priv)
result = c.resolve_catalog(None, None, stale, floor=10)
assert result == {}
def test_resolve_catalog_malformed_candidate_does_not_raise(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
malformed_raw = b"{not valid json"
malformed_sig = _sign(malformed_raw, priv)
good = _signed(_minimal_catalog(version=1), priv)
result = c.resolve_catalog((malformed_raw, malformed_sig), None, good)
assert c.catalog_version(result) == 1
def test_resolve_catalog_invalid_but_signed_candidate_is_skipped(monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
invalid = _signed(_catalog_with({"config": {"command": "bash", "args": []}}), priv)
good = _signed(_minimal_catalog(version=1), priv)
result = c.resolve_catalog(invalid, None, good)
assert c.catalog_version(result) == 1
# --- catalog_entry_to_paste_json / config_has_unfilled_placeholders ------- #
def test_catalog_entry_to_paste_json_basic_shape():
entry = _minimal_catalog()["servers"][0]
result = c.catalog_entry_to_paste_json(entry)
assert result == {"widget": {"command": "npx", "args": ["-y", "widget-mcp@1.0.0"]}}
def test_catalog_entry_to_paste_json_includes_env_when_present():
# NOTE: catalog_entry_to_paste_json() is a pure shape-converter for an
# entry that has ALREADY passed validate_catalog() -- it is correct for
# it to carry env through verbatim. The bug (issue #68 finding 2) was
# never in this function; it was that validate_catalog() let entries
# with dangerous/non-placeholder env values reach this function in the
# first place. This test now proves that boundary explicitly: a
# validation-legal env value (a <PLACEHOLDER> token) survives the
# conversion, and a value validate_catalog() would have rejected is
# confirmed rejected before it ever gets here.
entry = _minimal_catalog()["servers"][0]
entry["config"]["env"] = {"GRAFANA_URL": "<GRAFANA_URL>"}
result = c.catalog_entry_to_paste_json(entry)
assert result["widget"]["env"] == {"GRAFANA_URL": "<GRAFANA_URL>"}
catalog = _minimal_catalog()
catalog["servers"][0]["config"]["env"] = {"GRAFANA_URL": "<GRAFANA_URL>"}
assert c.validate_catalog(catalog) == []
malicious = _minimal_catalog()
malicious["servers"][0]["config"]["env"] = {"NODE_OPTIONS": "--require /tmp/payload.js"}
assert c.validate_catalog(malicious) != []
def test_config_has_unfilled_placeholders_true_for_token():
cfg = {"command": "npx", "args": ["-y", "server", "<ALLOWED_DIR>"]}
assert c.config_has_unfilled_placeholders(cfg) is True
def test_config_has_unfilled_placeholders_false_after_fill():
cfg = {"command": "npx", "args": ["-y", "server", "/Users/me/project"]}
assert c.config_has_unfilled_placeholders(cfg) is False
def test_config_has_unfilled_placeholders_checks_env_too():
cfg = {"command": "uvx", "args": ["mcp-grafana"], "env": {"GRAFANA_URL": "<GRAFANA_URL>"}}
assert c.config_has_unfilled_placeholders(cfg) is True
# --------------------------------------------------------------------------- #
# #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.<slot>` 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"
# --------------------------------------------------------------------------- #
# #76 -- ${VAR} references. Semantics mirror Claude Code's documented
# behaviour: ${VAR} and ${VAR:-default}, expanded in command/args/env/url/
# headers, and an unset variable with no default left as literal text.
# --------------------------------------------------------------------------- #
def test_find_env_refs_plain_and_defaulted():
refs = c.find_env_refs("${A} and ${B:-fallback}")
assert [(r.name, r.default) for r in refs] == [("A", None), ("B", "fallback")]
@pytest.mark.parametrize("text", ["${}", "${1BAD}", "$NOTBRACED", "{NOPE}", "plain", "$${X"])
def test_find_env_refs_ignores_non_references(text):
assert c.find_env_refs(text) == []
def test_find_env_refs_allows_an_empty_default():
"""`${VAR:-}` is a documented way to say 'blank if unset'."""
refs = c.find_env_refs("${A:-}")
assert refs[0].default == ""
assert refs[0].has_default is True
def test_server_env_refs_covers_all_five_documented_fields():
data = {
"command": "${BIN}",
"args": ["--x", "${ARG}"],
"env": {"K": "${ENVV}"},
"url": "${URL}/mcp",
"headers": {"Authorization": "Bearer ${HDR}"},
}
found = {(r.name, r.field) for r in c.server_env_refs(data)}
assert found == {
("BIN", "command"),
("ARG", "args"),
("ENVV", "env"),
("URL", "url"),
("HDR", "headers"),
}
def test_server_env_refs_ignores_unexpanded_fields():
"""Claude Code expands five fields; a ${VAR} elsewhere isn't a reference."""
assert c.server_env_refs({"description": "${NOPE}", "timeout": "${ALSO_NO}"}) == []
def test_expand_env_refs_matches_documented_semantics():
env = {"SET": "value"}
assert c.expand_env_refs("${SET}", env) == "value"
assert c.expand_env_refs("${MISSING:-dflt}", env) == "dflt"
assert c.expand_env_refs("${SET:-dflt}", env) == "value"
# unset with no default: left as literal text, exactly as Claude Code does
assert c.expand_env_refs("${MISSING}", env) == "${MISSING}"
def test_expand_env_refs_handles_several_in_one_string():
assert c.expand_env_refs("${A}/${B:-two}/${C}", {"A": "one"}) == "one/two/${C}"
def test_unresolved_env_refs_only_flags_unset_without_default():
data = {"env": {"A": "${SET}", "B": "${UNSET}", "C": "${OTHER:-has_default}"}}
assert [r.name for r in c.unresolved_env_refs(data, {"SET": "x"})] == ["UNSET"]
# --- the two interactions that were backwards for this feature ------------
def test_placeholder_under_a_secret_key_is_not_masked():
"""A ${VAR} names a secret rather than being one. Masking it would make a
reference indistinguishable from a stored credential."""
assert c.should_mask_value("API_KEY", "${API_KEY}") is False
assert c.should_mask_value("API_KEY", "ghp_realsecret") is True
assert c.should_mask_value("NOT_SECRET", "${API_KEY}") is False
def test_redacted_display_keeps_placeholders_but_masks_real_secrets():
out = c._redact_server_data({"env": {"API_KEY": "${API_KEY}", "TOKEN": "ghp_real"}})
assert out["env"]["API_KEY"] == "${API_KEY}"
assert out["env"]["TOKEN"] == c.MASK
def test_redact_args_keeps_placeholders_visible():
assert c.redact_args(["--token", "${GH_TOKEN}"]) == ["--token", "${GH_TOKEN}"]
assert c.redact_args(["--api-key=${K}"]) == ["--api-key=${K}"]
# real secrets still masked
assert c.redact_args(["--token", "ghp_real"]) == ["--token", c.MASK]
assert c.redact_args(["--api-key=sk-real"]) == [f"--api-key={c.MASK}"]
def test_args_secret_warning_is_silenced_by_a_placeholder():
"""Moving a token into ${VAR} is the recommended fix for this warning --
still warning afterwards would punish the fix."""
assert c.args_secret_warning({"args": ["--token", "ghp_real"]}) is not None
assert c.args_secret_warning({"args": ["--token", "${GH_TOKEN}"]}) is None
def test_args_secret_warning_still_fires_on_the_arg_after_a_placeholder():
"""A placeholder must clear the pending-flag state, not blanket-suppress."""
assert c.args_secret_warning({"args": ["${SAFE}", "--token", "ghp_real"]}) is not None
# --- per-client gating ----------------------------------------------------
def _profile(path):
return c.Profile(label="p", path=Path(path), config_exists=True)
def test_claude_code_profiles_expand_references():
assert c.client_expands_env_refs(_profile(Path.home() / ".claude.json")) is True
assert c.client_expands_env_refs(_profile("/repo/.mcp.json")) is True
def test_claude_desktop_profile_does_not_expand_references():
desktop = _profile(c.app_support_base() / "Claude" / c.CONFIG_FILENAME)
assert c.client_expands_env_refs(desktop) is False
def test_desktop_profile_warns_that_references_are_literal():
data = {"env": {"API_KEY": "${API_KEY}"}}
desktop = _profile(c.app_support_base() / "Claude" / c.CONFIG_FILENAME)
warnings = c.env_ref_warnings(data, desktop, {"API_KEY": "set"})
assert len(warnings) == 1
assert "NOT be expanded" in warnings[0]
assert "${API_KEY}" in warnings[0]
def test_claude_code_profile_warns_only_about_unset_variables():
code = _profile(Path.home() / ".claude.json")
data = {"env": {"A": "${UNSET_ONE}"}}
assert c.env_ref_warnings(data, code, {}) != []
assert c.env_ref_warnings(data, code, {"UNSET_ONE": "x"}) == []
# a default means it always resolves
assert c.env_ref_warnings({"env": {"A": "${X:-d}"}}, code, {}) == []
def test_no_references_means_no_warnings():
assert c.env_ref_warnings({"command": "npx", "args": ["-y", "pkg"]}, None) == []
# --------------------------------------------------------------------------- #
# Client adapters (issue #5 — cross-client support, phase 1)
#
# The refactor's promise is twofold: (1) the two Claude clients behave exactly
# as before, and (2) the ClientSpec seam is real — a client with a different
# servers key and a different per-server shape flows through the same pipeline.
# A synthetic "VS Code-like" spec stands in for the phase-2 client so the
# abstraction is proven now, before anything depends on it.
# --------------------------------------------------------------------------- #
def test_claude_specs_are_registered_and_mcpservers_shaped():
assert c.CLAUDE_DESKTOP.servers_key == "mcpServers"
assert c.CLAUDE_CODE.servers_key == "mcpServers"
assert c.CLAUDE_DESKTOP.disabled_key == c.DISABLED_KEY
assert c.CLAUDE_CODE.disabled_key == c.DISABLED_KEY
# capabilities the old inline filename checks used to compute
assert c.CLAUDE_DESKTOP.expands_env_refs is False
assert c.CLAUDE_CODE.expands_env_refs is True
assert c.CLAUDE_DESKTOP.supports_restart is True
assert c.CLAUDE_CODE.supports_restart is False
assert set(c.CLIENT_SPECS) == {c.CLAUDE_DESKTOP, c.CLAUDE_CODE}
assert c.DEFAULT_CLIENT is c.CLAUDE_DESKTOP
def test_client_by_key_round_trips_and_misses():
assert c.client_by_key("claude_desktop") is c.CLAUDE_DESKTOP
assert c.client_by_key("claude_code") is c.CLAUDE_CODE
assert c.client_by_key("nope") is None
def test_resolve_client_matches_the_old_filename_rule():
assert c.resolve_client("/x/Claude/claude_desktop_config.json") is c.CLAUDE_DESKTOP
assert c.resolve_client(Path.home() / ".claude.json") is c.CLAUDE_CODE
assert c.resolve_client("/repo/.mcp.json") is c.CLAUDE_CODE
assert c.resolve_client(Path.home() / ".claude" / "settings.json") is c.CLAUDE_CODE
def test_profile_auto_resolves_client_from_path():
desktop = c.Profile(
label="Claude", path="/x/Claude/claude_desktop_config.json", config_exists=True
)
code = c.Profile(label="Claude Code", path="/home/me/.claude.json", config_exists=True)
assert desktop.client is c.CLAUDE_DESKTOP
assert code.client is c.CLAUDE_CODE
def test_profile_honours_an_explicit_client():
# An explicit spec is not overridden by the path-based resolver.
p = c.Profile(
label="odd",
path="/somewhere/claude_desktop_config.json",
config_exists=True,
client=c.CLAUDE_CODE,
)
assert p.client is c.CLAUDE_CODE
def test_desktop_gating_and_env_expansion_read_off_the_spec():
desktop = c.Profile(label="d", path="/x/Claude/claude_desktop_config.json", config_exists=True)
code = c.Profile(label="c", path=Path.home() / ".claude.json", config_exists=True)
assert c.profile_targets_claude_desktop(desktop) is True
assert c.profile_targets_claude_desktop(code) is False
assert c.client_expands_env_refs(desktop) is False
assert c.client_expands_env_refs(code) is True
def test_extract_and_apply_default_spec_is_unchanged():
# No spec argument must behave byte-for-byte like the pre-refactor code.
cfg = {"mcpServers": {"a": {"command": "x"}}, "_disabledMcpServers": {"b": {"command": "y"}}}
servers = c.extract_servers(cfg)
assert {(s.name, s.enabled) for s in servers} == {("a", True), ("b", False)}
out = c.apply_servers({}, servers)
assert out == {
"mcpServers": {"a": {"command": "x"}},
"_disabledMcpServers": {"b": {"command": "y"}},
}
# A stand-in for the phase-2 VS Code adapter: different top-level key
# ("servers"), a different disabled key, and a per-server shape that carries a
# `type` field the internal model doesn't. entry_to/from_internal are the only
# things it overrides — proving that's the whole extension point.
class _FakeVSCode(c.ClientSpec):
def entry_to_internal(self, value):
if not isinstance(value, dict):
return value
return {k: v for k, v in value.items() if k != "type"}
def entry_from_internal(self, data):
if not isinstance(data, dict):
return data
return {"type": "stdio", **data}
_VSCODE = _FakeVSCode(
key="vscode_fake",
label="VS Code (test)",
servers_key="servers",
disabled_key="_bccDisabledServers",
)
def test_extract_reads_a_custom_servers_key_and_translates_shape():
cfg = {"servers": {"a": {"type": "stdio", "command": "x", "args": ["-y"]}}}
servers = c.extract_servers(cfg, _VSCODE)
assert len(servers) == 1
# the `type` field was translated out of the internal model
assert servers[0].data == {"command": "x", "args": ["-y"]}
def test_apply_writes_a_custom_key_translates_back_and_keeps_other_keys():
original = {"servers": {"old": {"type": "stdio", "command": "z"}}, "keepMe": {"x": 1}}
servers = c.extract_servers(original, _VSCODE)
out = c.apply_servers(original, servers, _VSCODE)
# round-trips through the custom key with the shape restored
assert out["servers"] == {"old": {"type": "stdio", "command": "z"}}
# the cardinal rule generalises: mcpServers is never introduced, and every
# unrelated key survives verbatim
assert "mcpServers" not in out
assert out["keepMe"] == {"x": 1}
def test_apply_uses_the_custom_disabled_key():
servers = [
c.ServerEntry("on", {"command": "a"}, True),
c.ServerEntry("off", {"command": "b"}, False),
]
out = c.apply_servers({}, servers, _VSCODE)
assert out["servers"] == {"on": {"type": "stdio", "command": "a"}}
assert out["_bccDisabledServers"] == {"off": {"type": "stdio", "command": "b"}}
assert c.DISABLED_KEY not in out
def test_spec_with_no_disabled_key_drops_disabled_and_never_parks():
no_park = c.ClientSpec(
key="nopark", label="No Park", servers_key="mcpServers", disabled_key=None
)
servers = [
c.ServerEntry("on", {"command": "a"}, True),
c.ServerEntry("off", {"command": "b"}, False),
]
out = c.apply_servers({}, servers, no_park)
assert out == {"mcpServers": {"on": {"command": "a"}}}
assert c.DISABLED_KEY not in out
assert no_park.section_keys() == ("mcpServers",)
def test_section_keys_reports_both_when_a_disabled_key_exists():
assert c.CLAUDE_DESKTOP.section_keys() == ("mcpServers", c.DISABLED_KEY)
assert _VSCODE.section_keys() == ("servers", "_bccDisabledServers")
def test_malformed_entry_round_trips_through_the_default_spec():
# #72's non-object server value must still be preserved verbatim on save.
cfg = {"mcpServers": {"bad": "oops", "good": {"command": "x"}}}
servers = c.extract_servers(cfg)
assert any(s.malformed and s.name == "bad" for s in servers)
out = c.apply_servers({}, servers)
assert out["mcpServers"]["bad"] == "oops"
def test_server_sections_and_change_summary_follow_a_custom_key(tmp_path):
loaded = {"servers": {"a": {"type": "stdio", "command": "x", "env": {"API_KEY": "sekret"}}}}
sections = c._server_sections(loaded, _VSCODE)
assert "servers" in sections
assert "mcpServers" not in sections
# secret masking still applies through the custom key
assert sections["servers"]["a"]["env"]["API_KEY"] == c.MASK
disk = {"servers": {"a": {"type": "stdio", "command": "CHANGED"}}}
p = tmp_path / "vscode.json"
p.write_text(json.dumps(disk), encoding="utf-8")
changed_keys, diff = c.external_change_summary(loaded, p, _VSCODE)
assert "servers" in changed_keys
assert diff # a server-section change under the custom key is diffed
# --------------------------------------------------------------------------- #
# Project profile label disambiguation (issue #74)
# --------------------------------------------------------------------------- #
def test_disambiguate_labels_no_collision_uses_basename():
dirs = [Path("/home/me/work/api"), Path("/home/me/work/web")]
labels = c.disambiguate_project_labels(dirs)
assert labels[Path("/home/me/work/api")] == "Project: api"
assert labels[Path("/home/me/work/web")] == "Project: web"
def test_disambiguate_labels_widens_only_colliding_basenames():
dirs = [
Path("/home/me/work/app"),
Path("/home/me/personal/app"),
Path("/home/me/notes"),
]
labels = c.disambiguate_project_labels(dirs)
# the two "app"s widen by one parent; the unique "notes" stays plain
assert labels[Path("/home/me/work/app")] == "Project: work/app"
assert labels[Path("/home/me/personal/app")] == "Project: personal/app"
assert labels[Path("/home/me/notes")] == "Project: notes"
def test_disambiguate_labels_widens_further_when_parent_also_collides():
dirs = [Path("/a/x/app"), Path("/b/x/app")]
labels = c.disambiguate_project_labels(dirs)
assert labels[Path("/a/x/app")] == "Project: a/x/app"
assert labels[Path("/b/x/app")] == "Project: b/x/app"
def _write_project(tmp_path, name, mcp_content):
d = tmp_path / name
d.mkdir(parents=True)
if mcp_content is not None:
(d / ".mcp.json").write_text(mcp_content, encoding="utf-8")
return d
def _claude_json_with_projects(tmp_path, dirs):
cj = tmp_path / ".claude.json"
cj.write_text(json.dumps({"projects": {str(d): {} for d in dirs}}), encoding="utf-8")
return cj
def test_discover_project_configs_disambiguates_same_basename(tmp_path):
d1 = _write_project(tmp_path / "work", "app", '{"mcpServers": {}}')
d2 = _write_project(tmp_path / "personal", "app", '{"mcpServers": {}}')
cj = _claude_json_with_projects(tmp_path, [d1, d2])
profiles = c.discover_project_configs(cj)
labels = sorted(p.label for p in profiles)
assert labels == ["Project: personal/app", "Project: work/app"]
def test_discover_project_configs_skips_non_object_and_garbage(tmp_path):
good = _write_project(tmp_path, "good", '{"mcpServers": {}}')
array = _write_project(tmp_path, "arr", "[1, 2, 3]")
garbage = _write_project(tmp_path, "junk", "not json at all")
missing = tmp_path / "nofile"
missing.mkdir()
cj = _claude_json_with_projects(tmp_path, [good, array, garbage, missing])
profiles = c.discover_project_configs(cj)
paths = {str(p.path) for p in profiles}
assert str(good / ".mcp.json") in paths
assert str(array / ".mcp.json") not in paths
assert str(garbage / ".mcp.json") not in paths
assert str(missing / ".mcp.json") not in paths
# --------------------------------------------------------------------------- #
# Move to environment variable (issue #83)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"raw,expected",
[
("API_KEY", "API_KEY"),
("api-key", "API_KEY"),
("x.y z", "X_Y_Z"),
("2fa", "_2FA"),
("", "VAR"),
("***", "VAR"),
("clé", "CL_"), # non-ASCII becomes _
],
)
def test_sanitize_env_var_name(raw, expected):
assert c.sanitize_env_var_name(raw) == expected
def test_shell_export_lines_quote_safely():
lines = c.shell_export_lines("TOKEN", "ab'cd")
assert lines["posix"] == "export TOKEN='ab'\\''cd'"
assert lines["windows"] == 'setx TOKEN "ab\'cd"'
def test_can_move_gate_requires_secret_and_expanding_client():
desktop = c.Profile(label="d", path="/x/Claude/claude_desktop_config.json", config_exists=True)
code = c.Profile(label="c", path=Path.home() / ".claude.json", config_exists=True)
# real secret on an expanding client -> offer
assert c.can_move_value_to_env_ref("API_KEY", "ghp_abc", code) is True
assert c.can_move_value_to_env_ref("API_KEY", "ghp_abc", None) is True
# non-secret key -> no
assert c.can_move_value_to_env_ref("REGION", "us-east-1", code) is False
# already a reference -> no
assert c.can_move_value_to_env_ref("API_KEY", "${API_KEY}", code) is False
# non-expanding client (Claude Desktop) -> refuse even a real secret
assert c.can_move_value_to_env_ref("API_KEY", "ghp_abc", desktop) is False
def test_move_env_value_replaces_with_reference_and_returns_secret():
data = {"command": "x", "env": {"API_KEY": "ghp_secret", "REGION": "us"}}
conv = c.move_value_to_env_ref(data, field="env", key="API_KEY")
assert conv is not None
assert conv.var_name == "API_KEY"
assert conv.reference == "${API_KEY}"
assert conv.secret == "ghp_secret"
assert conv.data["env"]["API_KEY"] == "${API_KEY}"
# non-secret row untouched
assert conv.data["env"]["REGION"] == "us"
# input never mutated
assert data["env"]["API_KEY"] == "ghp_secret"
def test_move_derives_and_sanitises_var_name_from_key():
data = {"headers": {"x-api-key": "sekret"}}
conv = c.move_value_to_env_ref(data, field="headers", key="x-api-key")
assert conv.var_name == "X_API_KEY"
assert conv.data["headers"]["x-api-key"] == "${X_API_KEY}"
def test_move_honours_explicit_var_name():
data = {"env": {"tok": "sekret"}}
conv = c.move_value_to_env_ref(data, field="env", key="tok", var_name="GITHUB_TOKEN")
assert conv.reference == "${GITHUB_TOKEN}"
assert conv.data["env"]["tok"] == "${GITHUB_TOKEN}"
def test_move_args_by_index():
data = {"command": "x", "args": ["--token", "ghp_secret"]}
conv = c.move_value_to_env_ref(data, field="args", index=1, var_name="GH_TOKEN")
assert conv.secret == "ghp_secret"
assert conv.data["args"] == ["--token", "${GH_TOKEN}"]
def test_move_returns_none_on_missing_or_nonstring_or_already_ref():
data = {"env": {"API_KEY": "${API_KEY}", "N": 5}}
assert c.move_value_to_env_ref(data, field="env", key="ABSENT") is None
assert c.move_value_to_env_ref(data, field="env", key="N") is None # not a string
assert c.move_value_to_env_ref(data, field="env", key="API_KEY") is None # already a ref
assert c.move_value_to_env_ref({}, field="bogus") is None
assert c.move_value_to_env_ref({"args": ["a"]}, field="args", index=9) is None
def test_is_env_var_set():
assert c.is_env_var_set("FOO", {"FOO": "x"}) is True
assert c.is_env_var_set("FOO", {"FOO": ""}) is False
assert c.is_env_var_set("FOO", {}) is False
# --------------------------------------------------------------------------- #
# Args secret indices + suggested var name (issue #83, args surface)
# --------------------------------------------------------------------------- #
def test_secret_arg_indices_flags_token_and_flag_value():
args = ["--port", "8080", "ghp_deadbeef", "--token", "sk-abc", "--flag=val"]
idxs = c.secret_arg_indices(args)
assert 2 in idxs # ghp_ token prefix
assert 4 in idxs # value following --token
assert 1 not in idxs # 8080
assert 5 not in idxs # --flag=val inline pair
def test_secret_arg_indices_flags_embedded_url_credentials():
args = ["postgres://user:pass@host/db"]
assert c.secret_arg_indices(args) == [0]
def test_secret_arg_indices_excludes_existing_references():
args = ["--token", "${GH_TOKEN}"]
assert c.secret_arg_indices(args) == []
def test_suggested_env_var_for_arg_uses_preceding_flag():
args = ["--api-key", "sk-secret"]
assert c.suggested_env_var_for_arg(args, 1) == "API_KEY"
def test_suggested_env_var_for_arg_falls_back_when_no_flag():
args = ["ghp_secret"]
assert c.suggested_env_var_for_arg(args, 0) == "SECRET"
# --------------------------------------------------------------------------- #
# Referenced-variables readout + args->env relocation (issue #83, "both")
# --------------------------------------------------------------------------- #
def test_referenced_env_vars_dedupes_and_reports_status():
data = {
"command": "npx",
"args": ["--token", "${GH_TOKEN}", "${GH_TOKEN}"],
"env": {"API_KEY": "${API_KEY}", "REGION": "${REGION:-us-east-1}"},
}
usages = c.referenced_env_vars(data, environ={"GH_TOKEN": "x"})
by = {u.name: u for u in usages}
assert set(by) == {"GH_TOKEN", "API_KEY", "REGION"}
# GH_TOKEN appears only in args, deduped to one entry, set in env -> resolved
assert by["GH_TOKEN"].fields == ("args",)
assert by["GH_TOKEN"].resolved is True
# API_KEY not set, no default -> unresolved
assert by["API_KEY"].resolved is False
# REGION has a default -> resolved regardless of environment
assert by["REGION"].has_default is True
assert by["REGION"].resolved is True
# sorted by name
assert [u.name for u in usages] == sorted(u.name for u in usages)
def test_referenced_env_vars_empty_when_no_refs():
assert c.referenced_env_vars({"command": "npx", "args": ["-y", "pkg"]}) == []
def test_move_arg_to_env_block_removes_flag_and_value():
data = {"command": "x", "args": ["--api-key", "sk-secret", "run"], "env": {"KEEP": "1"}}
out = c.move_arg_to_env_block(data, 1)
assert out["args"] == ["run"] # flag + value both gone
assert out["env"]["API_KEY"] == "sk-secret"
assert out["env"]["KEEP"] == "1" # existing env preserved
# input not mutated
assert data["args"] == ["--api-key", "sk-secret", "run"]
def test_move_arg_to_env_block_bare_positional_keeps_no_flag():
data = {"command": "x", "args": ["ghp_secret", "serve"]}
out = c.move_arg_to_env_block(data, 0, var_name="GITHUB_TOKEN")
assert out["args"] == ["serve"]
assert out["env"] == {"GITHUB_TOKEN": "ghp_secret"}
def test_move_arg_to_env_block_none_on_bad_target():
assert c.move_arg_to_env_block({"args": ["a"]}, 5) is None
assert c.move_arg_to_env_block({"args": ["${REF}"]}, 0) is None # already a ref
assert c.move_arg_to_env_block({}, 0) is None
Reference in New Issue View Git Blame Copy Permalink