better-claude-config/tests/test_core.py

"""Pytest port of the original test_core.py script (same 23 behaviours, now
proper test functions with tmp_path/monkeypatch fixtures)."""

import json
import sys

import pytest

import bcc_core as c


# --------------------------------------------------------------------------- #
# 1. Discovery
# --------------------------------------------------------------------------- #
@pytest.fixture
def fake_base(tmp_path, monkeypatch):
    base = tmp_path / "AppSupport"
    for d in ("Claude", "Claude-Work", "NotClaude"):
        (base / d).mkdir(parents=True)
    (base / "Claude" / c.CONFIG_FILENAME).write_text("{}")
    monkeypatch.setattr(c, "app_support_base", lambda: base)
    return base


def test_discovers_claude_dirs_not_others(fake_base):
    labels = sorted(p.label for p in c.discover_profiles())
    assert "Claude" in labels
    assert "Claude-Work" in labels
    assert "NotClaude" not in labels


def test_flags_missing_config_file(fake_base):
    profs = c.discover_profiles()
    assert any(p.label == "Claude-Work" and not p.config_exists for p in profs)


@pytest.fixture
def fake_home(tmp_path, monkeypatch, fake_base):
    home = tmp_path / "home"
    (home / ".claude").mkdir(parents=True)
    monkeypatch.setattr(c.Path, "home", lambda: home)
    return home


def test_claude_code_uses_dot_claude_json(fake_home):
    (fake_home / ".claude.json").write_text('{"mcpServers": {"x": {"command": "npx"}}}')
    profs = c.discover_profiles()
    cc = [p for p in profs if p.label == "Claude Code"]
    assert len(cc) == 1
    assert cc[0].path == fake_home / ".claude.json"
    assert cc[0].config_exists


def test_legacy_settings_json_surfaced_only_with_servers(fake_home):
    # settings.json without mcpServers -> not listed
    (fake_home / ".claude" / "settings.json").write_text('{"permissions": {}}')
    assert not any("legacy" in p.label for p in c.discover_profiles())
    # settings.json WITH parked mcpServers -> listed so the user can migrate
    (fake_home / ".claude" / "settings.json").write_text(
        '{"mcpServers": {"old": {"command": "npx"}}}'
    )
    assert any("legacy" in p.label for p in c.discover_profiles())


# --------------------------------------------------------------------------- #
# 2. Write pipeline: preserves other keys + order, only touches mcpServers
# --------------------------------------------------------------------------- #
@pytest.fixture
def written_config(tmp_path):
    cfgpath = tmp_path / "real.json"
    original = {
        "globalShortcut": "Cmd+Shift+Space",
        "mcpServers": {"old": {"command": "node", "args": ["x.js"]}},
        "someOtherTool": {"keep": True},
    }
    cfgpath.write_text(json.dumps(original, indent=2))
    cfg = c.load_config(cfgpath)
    servers = c.extract_servers(cfg)
    servers.append(c.ServerEntry("brave", {"command": "npx", "args": ["-y", "@x/brave"]}, True))
    servers[0].enabled = False
    c.apply_servers(cfg, servers)
    c.write_config(cfgpath, cfg)
    return cfgpath, json.loads(cfgpath.read_text())


def test_preserves_unrelated_keys(written_config):
    _, written = written_config
    assert written.get("globalShortcut") == "Cmd+Shift+Space"
    assert written.get("someOtherTool") == {"keep": True}


def test_disabled_server_parked(written_config):
    _, written = written_config
    assert "old" in written.get(c.DISABLED_KEY, {})
    assert "old" not in written.get("mcpServers", {})


def test_enabled_server_active(written_config):
    _, written = written_config
    assert "brave" in written.get("mcpServers", {})


def test_key_order_kept(written_config):
    _, written = written_config
    keys = list(written.keys())
    assert keys.index("globalShortcut") < keys.index("mcpServers")


def test_backup_created(written_config):
    cfgpath, _ = written_config
    bdir = cfgpath.parent / c.BACKUP_DIRNAME
    assert bdir.is_dir()
    assert any(bdir.iterdir())


# --------------------------------------------------------------------------- #
# 3. Paste parser — all accepted shapes
# --------------------------------------------------------------------------- #
def test_parse_full_config_shape():
    full = '{"mcpServers": {"a": {"command": "node", "args": ["a.js"]}}}'
    assert list(c.parse_pasted_json(full)) == ["a"]


def test_parse_inner_block_shape():
    inner = '{"b": {"command": "uvx", "args": ["mcp-server-git"]}}'
    assert list(c.parse_pasted_json(inner)) == ["b"]


def test_parse_bare_server_suggests_name():
    bare = '{"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]}'
    assert "filesystem" in c.parse_pasted_json(bare)


def test_parse_remote_suggests_host_name():
    remote = '{"url": "https://mcp.asana.com/sse"}'
    assert "mcp" in c.parse_pasted_json(remote)


def test_rejects_junk():
    with pytest.raises(ValueError):
        c.parse_pasted_json('{"junk": 5}')


# --------------------------------------------------------------------------- #
# 4. Validation
# --------------------------------------------------------------------------- #
@pytest.fixture
def problems():
    bad = [
        c.ServerEntry("", {"command": "x"}, True),
        c.ServerEntry("dup", {"command": "x"}, True),
        c.ServerEntry("dup", {"command": "x"}, True),
        c.ServerEntry("r", {"url": ""}, True),
        c.ServerEntry("nocmd", {"command": ""}, True),
    ]
    return c.validate_servers(bad)


def test_validation_catches_empty_name(problems):
    assert any("empty name" in p for p in problems)


def test_validation_catches_duplicate(problems):
    assert any("Duplicate" in p for p in problems)


def test_validation_catches_missing_url(problems):
    assert any("no URL" in p for p in problems)


def test_validation_catches_missing_command(problems):
    assert any("no command" in p for p in problems)


def test_valid_set_passes_clean():
    assert c.validate_servers([c.ServerEntry("good", {"command": "node"}, True)]) == []


# --------------------------------------------------------------------------- #
# 5. Secrets: detection + redaction
# --------------------------------------------------------------------------- #
def test_is_secret_key_positives():
    for k in (
        "BRAVE_API_KEY",
        "BEARER_TOKEN",
        "password",
        "GH_TOKEN",
        "Authorization",
        "client_secret",
        "PRIVATE_KEY",
        "aws_access_key_id",
    ):
        assert c.is_secret_key(k), k


def test_is_secret_key_negatives():
    for k in ("JOOMLA_BASE_URL", "PORT", "HOME", "PATH", "debug", "TIMEOUT"):
        assert not c.is_secret_key(k), k


def test_redact_args_flag_value():
    assert c.redact_args(["--token", "abc123", "run"]) == ["--token", c.MASK, "run"]


def test_redact_args_inline_flag():
    assert c.redact_args(["--api-key=abc123"]) == [f"--api-key={c.MASK}"]


def test_redact_args_known_prefix():
    assert c.redact_args(["ghp_abcdef123456"]) == [c.MASK]


def test_redact_args_leaves_normal_args():
    args = ["--directory", "/path/to/server", "run", "main.py", "-y"]
    assert c.redact_args(args) == args


def test_diagnostics_redacts_token_args():
    txt = c.diagnostics_text("t", {"command": "npx", "args": ["--token", "s3cret-value"]})
    assert "s3cret-value" not in txt
    assert c.MASK in txt


# --------------------------------------------------------------------------- #
# 6. Dependency / PATH checking
# --------------------------------------------------------------------------- #
def test_dep_check_finds_python3():
    assert c.check_dependency({"command": "python3"})["status"] == "ok"


def test_dep_check_flags_missing():
    assert c.check_dependency({"command": "definitely-not-real-binary-xyz"})["status"] == "missing"


def test_dep_check_marks_remote():
    assert c.check_dependency({"url": "https://example.com/mcp"})["status"] == "remote"


def test_dep_check_sees_through_shell_wrapper():
    r = c.check_dependency({"command": "cmd", "args": ["/c", "definitely-not-real-xyz", "foo"]})
    assert "definitely-not-real-xyz" in r["label"]


# --------------------------------------------------------------------------- #
# 7. Spawn test
# --------------------------------------------------------------------------- #


def test_spawn_test_ok_still_running():
    # A process that sleeps longer than the timeout → "ok" (server started)
    r = c.spawn_test(
        {"command": sys.executable, "args": ["-c", "import time; time.sleep(60)"]}, timeout=0.3
    )
    assert r["outcome"] == "ok"
    assert r["returncode"] is None


def test_spawn_test_crashed():
    # A process that exits with a non-zero code immediately
    r = c.spawn_test(
        {"command": sys.executable, "args": ["-c", "raise SystemExit(1)"]}, timeout=0.3
    )
    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)
    r = c.spawn_test({"command": sys.executable, "args": ["-c", "pass"]}, timeout=0.5)
    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=0.5,
    )
    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