better-claude-config/tests/test_catalog_review.py

"""Tests for catalog_review.py -- semantic diff, risk predicates, review
session (acknowledge-gating + TOCTOU blob pinning), key encryption, and
registry-lookup logic for the Catalog Console (issue #62)."""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import bcc_core as c
import catalog_review as r


def _entry(**overrides):
    base = {
        "id": "filesystem",
        "display": "Filesystem",
        "description": "desc",
        "category": "files",
        "homepage": "https://github.com/modelcontextprotocol/servers",
        "stars": 100,
        "official": True,
        "setup": "basic",
        "config": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem@1.0.0"],
        },
        "env_required": {},
        "docs_url": "https://github.com/modelcontextprotocol/servers",
        "notes": "",
        "last_release": "2026-01-01",
    }
    base.update(overrides)
    return base


def _catalog(*entries):
    return {"schema": 1, "version": 1, "servers": list(entries)}


# --------------------------------------------------------------------------- #
# diff_catalogs
# --------------------------------------------------------------------------- #
def test_diff_detects_added_entry():
    old = _catalog()
    new = _catalog(_entry())
    changes = r.diff_catalogs(old, new)
    assert len(changes) == 1
    assert changes[0].status == "added"
    assert changes[0].entry_id == "filesystem"
    assert changes[0].old is None


def test_diff_detects_removed_entry():
    old = _catalog(_entry())
    new = _catalog()
    changes = r.diff_catalogs(old, new)
    assert len(changes) == 1
    assert changes[0].status == "removed"
    assert changes[0].new is None


def test_diff_detects_no_change():
    e = _entry()
    old = _catalog(e)
    new = _catalog(dict(e))
    assert r.diff_catalogs(old, new) == []


def test_diff_detects_changed_command_and_args():
    old = _catalog(_entry())
    new = _catalog(_entry(config={"command": "uvx", "args": ["other-pkg@2.0.0"]}))
    changes = r.diff_catalogs(old, new)
    assert len(changes) == 1
    ch = changes[0]
    assert ch.status == "changed"
    fields = {fc.field for fc in ch.field_changes}
    assert "config.command" in fields
    assert "config.args" in fields


def test_diff_detects_description_change():
    old = _catalog(_entry())
    new = _catalog(_entry(description="new description"))
    changes = r.diff_catalogs(old, new)
    assert changes[0].field_changes == (r.FieldChange("description", "desc", "new description"),)


def test_diff_ignores_entries_without_id():
    old = _catalog()
    new = _catalog({"display": "no id"})
    assert r.diff_catalogs(old, new) == []


def test_diff_multiple_entries_sorted_by_id():
    old = _catalog(_entry(id="zeta"), _entry(id="alpha"))
    new = _catalog(
        _entry(id="zeta", description="changed"), _entry(id="alpha", description="changed")
    )
    changes = r.diff_catalogs(old, new)
    assert [c_.entry_id for c_ in changes] == ["alpha", "zeta"]


# --------------------------------------------------------------------------- #
# risk_env_required
# --------------------------------------------------------------------------- #
def test_risk_env_required_blocking_on_nonempty_value():
    change = r.EntryChange("x", "changed", None, _entry(env_required={"API_KEY": "sk-real-value"}))
    findings = r.risk_env_required(change)
    assert len(findings) == 1
    assert findings[0].severity == "blocking"
    assert findings[0].code == "env_required_value"


def test_risk_env_required_clean_on_empty_value():
    change = r.EntryChange("x", "changed", None, _entry(env_required={"API_KEY": ""}))
    assert r.risk_env_required(change) == []


# --------------------------------------------------------------------------- #
# risk_command_allowlist
# --------------------------------------------------------------------------- #
def test_risk_command_allowlist_blocks_disallowed_command():
    change = r.EntryChange(
        "x", "changed", None, _entry(config={"command": "bash", "args": ["-c", "evil"]})
    )
    findings = r.risk_command_allowlist(change)
    assert len(findings) == 1
    assert findings[0].severity == "blocking"


def test_risk_command_allowlist_allows_listed_command():
    for cmd in sorted(c.CATALOG_ALLOWED_COMMANDS):
        change = r.EntryChange("x", "changed", None, _entry(config={"command": cmd, "args": []}))
        assert r.risk_command_allowlist(change) == []


# --------------------------------------------------------------------------- #
# risk_non_ascii
# --------------------------------------------------------------------------- #
def test_risk_non_ascii_flags_homoglyph_id():
    # Cyrillic 'а' (U+0430) instead of Latin 'a' -- classic homoglyph swap.
    evil_id = "filаsystem"
    change = r.EntryChange(evil_id, "changed", None, _entry(id=evil_id))
    findings = r.risk_non_ascii(change)
    assert any(f.code == "non_ascii_id" for f in findings)
    assert findings[0].severity == "blocking"
    # the offending string must be rendered with escapes, not raw
    assert "\\u0430" in findings[0].message


def test_risk_non_ascii_flags_arg():
    change = r.EntryChange(
        "x", "changed", None, _entry(config={"command": "npx", "args": ["pаckage@1.0.0"]})
    )
    findings = r.risk_non_ascii(change)
    assert any(f.code == "non_ascii_arg" for f in findings)


def test_risk_non_ascii_clean_for_ascii_entry():
    change = r.EntryChange("x", "changed", None, _entry())
    assert r.risk_non_ascii(change) == []


# --------------------------------------------------------------------------- #
# risk_unpinned_package
# --------------------------------------------------------------------------- #
def test_risk_unpinned_npm_package_no_version():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(config={"command": "npx", "args": ["-y", "@scope/pkg"]}),
    )
    findings = r.risk_unpinned_package(change)
    assert len(findings) == 1
    assert findings[0].code == "unpinned_npm_package"
    assert findings[0].severity == "blocking"


def test_risk_pinned_npm_package_is_clean():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(config={"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]}),
    )
    assert r.risk_unpinned_package(change) == []


def test_risk_unpinned_unscoped_npm_package():
    change = r.EntryChange(
        "x", "changed", None, _entry(config={"command": "npx", "args": ["-y", "somepkg"]})
    )
    findings = r.risk_unpinned_package(change)
    assert len(findings) == 1


def test_risk_unpinned_docker_latest_tag():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(config={"command": "docker", "args": ["run", "-i", "--rm", "myimage:latest"]}),
    )
    findings = r.risk_unpinned_package(change)
    assert len(findings) == 1
    assert findings[0].code == "unpinned_docker_image"


def test_risk_unpinned_docker_no_tag():
    change = r.EntryChange(
        "x", "changed", None, _entry(config={"command": "docker", "args": ["run", "myimage"]})
    )
    findings = r.risk_unpinned_package(change)
    assert len(findings) == 1


def test_risk_pinned_docker_image_is_clean():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(config={"command": "docker", "args": ["run", "-i", "--rm", "myimage:1.2.3"]}),
    )
    assert r.risk_unpinned_package(change) == []


def test_risk_docker_digest_pin_is_clean():
    change = r.EntryChange(
        "x",
        "changed",
        None,
        _entry(
            config={
                "command": "docker",
                "args": ["run", "myimage@sha256:" + "a" * 64],
            }
        ),
    )
    assert r.risk_unpinned_package(change) == []


# --------------------------------------------------------------------------- #
# risk_url_domain_change
# --------------------------------------------------------------------------- #
def test_risk_url_domain_change_warns_on_lookalike_swap():
    old_entry = _entry(homepage="https://github.com/foo/bar")
    new_entry = _entry(homepage="https://githu6.com/foo/bar")
    change = r.EntryChange("x", "changed", old_entry, new_entry)
    findings = r.risk_url_domain_change(change)
    assert any(f.code == "domain_changed" for f in findings)
    domain_finding = next(f for f in findings if f.code == "domain_changed")
    assert "github.com" in domain_finding.message
    assert "githu6.com" in domain_finding.message
    assert domain_finding.severity == "warning"


def test_risk_url_domain_change_clean_when_domain_unchanged():
    old_entry = _entry(homepage="https://github.com/foo/bar")
    new_entry = _entry(homepage="https://github.com/foo/bar-renamed")
    change = r.EntryChange("x", "changed", old_entry, new_entry)
    assert r.risk_url_domain_change(change) == []


def test_risk_url_non_https_warns():
    new_entry = _entry(homepage="http://example.com")
    change = r.EntryChange("x", "changed", None, new_entry)
    findings = r.risk_url_domain_change(change)
    assert any(f.code == "non_https_url" for f in findings)


# --------------------------------------------------------------------------- #
# risk_new_entry / entry_risk_findings / has_blocking_risk
# --------------------------------------------------------------------------- #
def test_risk_new_entry_flags_added():
    change = r.EntryChange("x", "added", None, _entry())
    findings = r.risk_new_entry(change)
    assert len(findings) == 1
    assert findings[0].severity == "info"


def test_risk_new_entry_silent_for_changed():
    change = r.EntryChange("x", "changed", _entry(), _entry(description="x"))
    assert r.risk_new_entry(change) == []


def test_has_blocking_risk_true_for_disallowed_command():
    change = r.EntryChange("x", "changed", None, _entry(config={"command": "bash", "args": []}))
    assert r.has_blocking_risk(change) is True


def test_has_blocking_risk_false_for_clean_entry():
    change = r.EntryChange("x", "changed", _entry(), _entry(description="new"))
    assert r.has_blocking_risk(change) is False


# --------------------------------------------------------------------------- #
# ReviewSession: acknowledge gating
# --------------------------------------------------------------------------- #
def test_start_review_computes_diff():
    old = _catalog()
    new = _catalog(_entry())
    session = r.start_review("sha1", old, new)
    assert len(session.changes) == 1


def test_all_entries_acknowledged_false_initially():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    assert r.all_entries_acknowledged(session) is False


def test_acknowledge_entry_marks_acknowledged():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    r.acknowledge_entry(session, "filesystem")
    assert r.all_entries_acknowledged(session) is True


def test_acknowledge_unknown_entry_raises():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    with pytest.raises(ValueError):
        r.acknowledge_entry(session, "not-in-diff")


def test_acknowledge_gating_requires_every_entry():
    session = r.start_review("sha1", _catalog(), _catalog(_entry(id="a"), _entry(id="b")))
    r.acknowledge_entry(session, "a")
    assert r.all_entries_acknowledged(session) is False
    r.acknowledge_entry(session, "b")
    assert r.all_entries_acknowledged(session) is True


def test_no_acknowledge_all_shortcut_and_gate_is_real():
    """Two things, both load-bearing (issue #68: the original version of
    this test asserted ONLY the first half, and passed the entire time the
    gate below it was vacuously satisfiable -- 'no function named
    acknowledge_all' is worthless if signing doesn't actually require
    acknowledgement in practice).

    1. No bulk-acknowledge shortcut exists (see the comment in
       catalog_review.py above SignDecision -- deliberate friction).
    2. The gate that friction protects is actually enforced: with entries
       still unacknowledged, can_sign() must refuse, not just "some GUI
       checkbox happens to be unticked".
    """
    names = [n for n in dir(r) if "acknowledge" in n.lower()]
    assert "acknowledge_all" not in names
    assert "acknowledge_all_entries" not in names

    session = r.start_review("sha1", _catalog(), _catalog(_entry(id="a"), _entry(id="b")))
    r.acknowledge_entry(session, "a")  # only one of two -- not a bulk call
    decision = r.can_sign(session, "sha1")
    assert decision.ok is False
    assert "acknowledged" in decision.reason.lower()


# --------------------------------------------------------------------------- #
# can_sign: TOCTOU blob pinning + acknowledge gating combined
# --------------------------------------------------------------------------- #
def test_can_sign_false_when_not_all_acknowledged():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    decision = r.can_sign(session, "sha1")
    assert decision.ok is False
    assert "acknowledged" in decision.reason


def test_can_sign_true_when_acknowledged_and_blob_matches():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    r.acknowledge_entry(session, "filesystem")
    decision = r.can_sign(session, "sha1")
    assert decision.ok is True
    assert decision.reason is None


def test_can_sign_refuses_on_blob_mismatch_even_if_acknowledged():
    """The core TOCTOU fix: acknowledging everything is not enough if the
    bytes on the remote changed underneath the review."""
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    r.acknowledge_entry(session, "filesystem")
    decision = r.can_sign(session, "sha2-a-new-commit-landed")
    assert decision.ok is False
    assert "changed" in decision.reason.lower() or "mismatch" in decision.reason.lower()


def test_can_sign_blob_mismatch_takes_priority_message():
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    decision = r.can_sign(session, "sha2")
    assert decision.ok is False
    assert "blob" in decision.reason.lower() or "changed" in decision.reason.lower()


def test_can_sign_false_on_empty_changeset():
    """The exact bug behind issue #68 finding 1: 'main' loaded against
    itself diffs to [], and an empty changeset used to leave can_sign()
    with nothing to refuse on (set() <= set() is vacuously True). Commit
    b08cf21 signed 19 entries through precisely this path -- zero of them
    were ever reviewed. An empty diff must mean 'nothing to sign', never
    'sign unlocked'."""
    same_catalog = _catalog(_entry())
    session = r.start_review("sha1", same_catalog, same_catalog)
    assert session.changes == []  # diff_catalogs(x, x) -> []
    assert r.all_entries_acknowledged(session) is True  # vacuously -- this is the trap
    decision = r.can_sign(session, "sha1")  # blob matches, "everything" acknowledged
    assert decision.ok is False
    assert "nothing to sign" in decision.reason.lower()


def test_can_sign_false_with_outstanding_blocking_risk_even_if_acknowledged():
    """can_sign() must itself refuse a blocking risk finding -- today a
    blocking finding only disables the GUI checkbox, so the pure gate must
    not simply trust that the caller never acknowledged a blocking entry.
    Acknowledge it directly here (bypassing any GUI checkbox-disable logic
    entirely) to prove the gate catches it independently of the GUI."""
    session = r.start_review(
        "sha1",
        _catalog(),
        _catalog(_entry(config={"command": "bash", "args": ["-c", "evil"]})),
    )
    r.acknowledge_entry(session, "filesystem")
    assert r.all_entries_acknowledged(session) is True
    decision = r.can_sign(session, "sha1")
    assert decision.ok is False
    assert "blocking" in decision.reason.lower()


# --------------------------------------------------------------------------- #
# sign_precondition: the ref-resolution seam that used to hardcode "main"
# --------------------------------------------------------------------------- #
def test_sign_precondition_resolves_against_loaded_ref_not_hardcoded_main():
    """The regression test for issue #68 finding 1's first bug:
    ReviewWindow._on_sign used to hardcode fetch_ref(repo, "main") as the
    TOCTOU comparison ref. For a PR review, _on_load pins the PR HEAD's
    blob SHA, so comparing against main's SHA differs by definition and
    Sign could never fire on the PR path.

    The fake resolver below returns a DIFFERENT (deliberately wrong) SHA for
    "main" than for the PR ref that was actually loaded. If
    sign_precondition ever resolves against "main" instead of
    session.loaded_ref, this test fails -- both via the recorded `calls`
    list and via decision.ok flipping to False.
    """
    pr_ref = "refs/pull/42/head"
    session = r.start_review("pr-blob-sha", _catalog(), _catalog(_entry()), loaded_ref=pr_ref)
    r.acknowledge_entry(session, "filesystem")

    calls: list[str] = []

    def fake_resolver(ref: str) -> str:
        calls.append(ref)
        return {"main": "main-blob-sha-WRONG", pr_ref: "pr-blob-sha"}[ref]

    decision = r.sign_precondition(session, fake_resolver)
    assert calls == [pr_ref]  # never asked the resolver for "main"
    assert decision.ok is True
    assert decision.reason is None


def test_sign_precondition_refuses_when_loaded_ref_blob_moved():
    """Same seam, the negative case: if the loaded ref's blob SHA has moved
    since review began (a new commit landed on the reviewed PR/branch), the
    resolver reflects that and sign_precondition must refuse -- proving this
    isn't just a hardcoded pass-through."""
    pr_ref = "refs/pull/42/head"
    session = r.start_review("pr-blob-sha", _catalog(), _catalog(_entry()), loaded_ref=pr_ref)
    r.acknowledge_entry(session, "filesystem")

    def fake_resolver(_ref: str) -> str:
        return "pr-blob-sha-AFTER-A-NEW-PUSH"

    decision = r.sign_precondition(session, fake_resolver)
    assert decision.ok is False
    assert "mismatch" in decision.reason.lower() or "changed" in decision.reason.lower()


def test_sign_precondition_defaults_to_main_when_loaded_ref_unset():
    """start_review()'s loaded_ref defaults to 'main' for source=main
    reviews (and backward-compat with callers that don't pass it)."""
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    assert session.loaded_ref == "main"
    r.acknowledge_entry(session, "filesystem")

    def fake_resolver(ref: str) -> str:
        assert ref == "main"
        return "sha1"

    decision = r.sign_precondition(session, fake_resolver)
    assert decision.ok is True


# --------------------------------------------------------------------------- #
# ReviewSession(reattest=True): key-rotation re-attestation (issue #68
# finding 5 follow-up). After rotating bcc_core.CATALOG_PUBKEYS, the
# existing data/catalog.json.sig no longer verifies under the new key even
# though catalog CONTENT is unchanged -- diff_catalogs(old, new) would be
# empty, and an empty changeset must never unlock Sign (can_sign gate 0).
# Re-attestation mode sidesteps that correctly: instead of diffing against
# the (now-untrustworthy) last-signed content, it treats every entry as
# requiring a fresh acknowledgement, exactly like a brand-new catalog.
# --------------------------------------------------------------------------- #
def test_reattest_mode_with_unchanged_content_yields_one_change_per_entry():
    same = _catalog(_entry(id="a"), _entry(id="b"), _entry(id="c"))
    session = r.start_review("sha1", same, same, reattest=True)
    assert len(session.changes) == 3
    assert {c_.entry_id for c_ in session.changes} == {"a", "b", "c"}
    # Presented "as if newly added" -- old_catalog plays no role here.
    assert all(c_.status == "added" for c_ in session.changes)
    assert all(c_.old is None for c_ in session.changes)


def test_reattest_mode_can_sign_refuses_until_all_acknowledged_then_permits():
    same = _catalog(_entry(id="a"), _entry(id="b"), _entry(id="c"))
    session = r.start_review("sha1", same, same, reattest=True)

    decision = r.can_sign(session, "sha1")
    assert decision.ok is False
    assert "acknowledged" in decision.reason.lower()

    r.acknowledge_entry(session, "a")
    r.acknowledge_entry(session, "b")
    decision = r.can_sign(session, "sha1")
    assert decision.ok is False  # "c" still outstanding
    assert "acknowledged" in decision.reason.lower()

    r.acknowledge_entry(session, "c")
    decision = r.can_sign(session, "sha1")
    assert decision.ok is True
    assert decision.reason is None


def test_normal_mode_with_unchanged_content_still_refuses_empty_diff():
    """The rotation path must NOT become a general bypass of the empty-diff
    guard: reattest=False (the default) against identical old/new catalogs
    must behave exactly as before -- can_sign refuses with 'nothing to
    sign', full stop."""
    same = _catalog(_entry(id="a"), _entry(id="b"))
    session = r.start_review("sha1", same, same)  # reattest defaults False
    assert session.changes == []
    decision = r.can_sign(session, "sha1")
    assert decision.ok is False
    assert "nothing to sign" in decision.reason.lower()


def test_reattest_mode_still_enforces_blocking_risk_check():
    """Re-attestation must not relax the blocking-risk gate: a
    disallowed-command entry blocks Sign even with every entry
    acknowledged."""
    cat = _catalog(
        _entry(id="a"),
        _entry(id="evil", config={"command": "bash", "args": ["-c", "rm -rf /"]}),
    )
    session = r.start_review("sha1", cat, cat, reattest=True)
    assert len(session.changes) == 2
    r.acknowledge_entry(session, "a")
    r.acknowledge_entry(session, "evil")
    assert r.all_entries_acknowledged(session) is True
    decision = r.can_sign(session, "sha1")
    assert decision.ok is False
    assert "blocking" in decision.reason.lower()


def test_reattest_mode_still_enforces_toctou_pin():
    """Re-attestation must not relax the TOCTOU blob-SHA pin: acknowledging
    everything is not enough if the bytes moved underneath the review."""
    same = _catalog(_entry(id="a"))
    session = r.start_review("sha1", same, same, reattest=True)
    r.acknowledge_entry(session, "a")
    decision = r.can_sign(session, "sha1")
    assert decision.ok is True  # sanity: matches when blob is unchanged

    decision = r.can_sign(session, "sha2-a-new-commit-landed")
    assert decision.ok is False
    assert "mismatch" in decision.reason.lower() or "changed" in decision.reason.lower()


def test_reattest_defaults_to_false():
    """start_review()'s reattest parameter defaults to False -- normal
    (diff-based) review is the default behaviour, never silently entered."""
    session = r.start_review("sha1", _catalog(), _catalog(_entry()))
    assert session.reattest is False


# --------------------------------------------------------------------------- #
# find_last_signed_catalog_raw: what source=main diffs against
# --------------------------------------------------------------------------- #
def test_find_last_signed_catalog_raw_returns_matching_candidate():
    """Simulates walking catalog.json's git history: the CURRENT signature
    covers an OLDER version of the bytes (a later commit changed
    catalog.json without re-signing -- the exact bypass that produced
    commit b08cf21). The first candidate that verifies against that
    signature is 'the last catalog a maintainer actually signed'."""
    seed, pubkey = r.generate_keypair()
    old_raw = b'{"schema":1,"version":1,"servers":[]}'
    new_raw = b'{"schema":1,"version":2,"servers":[]}'
    sig = r.sign_catalog_bytes(old_raw, seed)  # signature covers the OLD bytes
    found = r.find_last_signed_catalog_raw([new_raw, old_raw], sig, [pubkey])
    assert found == old_raw


def test_find_last_signed_catalog_raw_none_when_nothing_verifies():
    seed, _pubkey = r.generate_keypair()
    _other_seed, other_pubkey = r.generate_keypair()
    raw = b'{"schema":1,"version":1,"servers":[]}'
    sig = r.sign_catalog_bytes(raw, seed)
    # Check against a pubkey list that does NOT include the signer's key.
    assert r.find_last_signed_catalog_raw([raw], sig, [other_pubkey]) is None


# --------------------------------------------------------------------------- #
# catalog_signing_message: domain separation must match bcc_core exactly
# --------------------------------------------------------------------------- #
def test_signing_message_uses_bcc_core_domain_prefix():
    raw = b'{"schema":1}'
    msg = r.catalog_signing_message(raw)
    assert msg == c._CATALOG_SIG_DOMAIN + raw
    assert msg.startswith(b"bcc-catalog-v1|")


def test_sign_then_verify_round_trips_with_bcc_core():
    """End-to-end: a signature produced by the Console's sign_catalog_bytes
    must verify with bcc_core.verify_catalog_signature -- proves the two
    modules can never drift on the domain-separation prefix."""
    seed, pubkey = r.generate_keypair()
    raw = b'{"schema":1,"version":2,"servers":[]}'
    sig = r.sign_catalog_bytes(raw, seed)
    assert c.verify_catalog_signature(raw, sig, [pubkey]) is True


def test_sign_tampered_bytes_fails_verify():
    seed, pubkey = r.generate_keypair()
    raw = b'{"schema":1,"version":2,"servers":[]}'
    sig = r.sign_catalog_bytes(raw, seed)
    tampered = raw[:-1] + b"0"
    assert c.verify_catalog_signature(tampered, sig, [pubkey]) is False


# --------------------------------------------------------------------------- #
# Key encryption at rest
# --------------------------------------------------------------------------- #
def test_encrypt_decrypt_round_trip():
    seed, _pub = r.generate_keypair()
    blob = r.encrypt_private_key(seed, "correct horse battery staple")
    decrypted = r.decrypt_private_key(blob, "correct horse battery staple")
    assert decrypted == seed


def test_decrypt_wrong_passphrase_raises():
    seed, _pub = r.generate_keypair()
    blob = r.encrypt_private_key(seed, "right passphrase")
    with pytest.raises(ValueError):
        r.decrypt_private_key(blob, "wrong passphrase")


def test_decrypt_corrupted_blob_raises():
    seed, _pub = r.generate_keypair()
    blob = r.encrypt_private_key(seed, "pass")
    corrupted = blob[:-1] + bytes([blob[-1] ^ 0xFF])
    with pytest.raises(ValueError):
        r.decrypt_private_key(corrupted, "pass")


def test_encrypt_private_key_rejects_wrong_length_seed():
    with pytest.raises(ValueError):
        r.encrypt_private_key(b"too-short", "pass")


def test_encrypt_private_key_rejects_empty_passphrase():
    seed, _pub = r.generate_keypair()
    with pytest.raises(ValueError):
        r.encrypt_private_key(seed, "")


def test_encrypted_blob_never_contains_seed_plaintext():
    seed, _pub = r.generate_keypair()
    blob = r.encrypt_private_key(seed, "some passphrase")
    assert seed not in blob


def test_generate_keypair_produces_valid_ed25519_pair():
    seed, pubkey = r.generate_keypair()
    assert len(seed) == 32
    assert len(pubkey) == 32
    raw = b"test payload"
    sig = r.sign_catalog_bytes(raw, seed)
    assert c.verify_catalog_signature(raw, sig, [pubkey]) is True


# --------------------------------------------------------------------------- #
# Registry lookup / near-neighbour edit distance
# --------------------------------------------------------------------------- #
def test_edit_distance_identical():
    assert r.edit_distance("abc", "abc") == 0


def test_edit_distance_one_substitution():
    assert r.edit_distance("firecrawl-mcp", "f1recrawl-mcp") == 1


def test_near_neighbor_ids_finds_close_match():
    others = ["firecrawl-mcp", "unrelated-server", "totally-different"]
    neighbors = r.near_neighbor_ids("firecrawl-mcp2", others, max_distance=2)
    assert "firecrawl-mcp" in neighbors


def test_near_neighbor_ids_excludes_self():
    others = ["filesystem", "other"]
    assert "filesystem" not in r.near_neighbor_ids("filesystem", others)


def test_near_neighbor_ids_excludes_far_matches():
    others = ["completely-unrelated-name"]
    assert r.near_neighbor_ids("filesystem", others, max_distance=2) == []


def test_extract_package_refs_npm():
    entry = _entry(config={"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]})
    refs = r.extract_package_refs(entry)
    assert len(refs) == 1
    assert refs[0].ecosystem == "npm"
    assert refs[0].name == "@scope/pkg"
    assert refs[0].version == "1.2.3"


def test_extract_package_refs_uvx():
    entry = _entry(config={"command": "uvx", "args": ["some-pypi-pkg==1.0.0"]})
    refs = r.extract_package_refs(entry)
    assert len(refs) == 1
    assert refs[0].ecosystem == "pypi"
    assert refs[0].name == "some-pypi-pkg"
    assert refs[0].version == "1.0.0"


def test_extract_package_refs_link_only_entry_returns_empty():
    entry = {"id": "slack", "setup": "link-only", "docs_url": "https://example.com"}
    assert r.extract_package_refs(entry) == []


def test_lookup_registry_info_fails_soft_on_none():
    ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")
    info = r.lookup_registry_info(ref, lambda _ref: None, [])
    assert info.available is False


def test_lookup_registry_info_fails_soft_on_exception():
    def boom(_ref):
        raise RuntimeError("network down")

    ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")
    info = r.lookup_registry_info(ref, boom, [])
    assert info.available is False


def test_lookup_registry_info_populates_fields_when_available():
    ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")

    def fetcher(_ref):
        return {
            "publisher": "hello_sideguide",
            "age_days": 30,
            "last_release": "2026-01-01",
            "downloads": 500,
        }

    info = r.lookup_registry_info(ref, fetcher, [])
    assert info.available is True
    assert info.publisher == "hello_sideguide"
    assert info.downloads == 500


def test_lookup_registry_info_includes_near_neighbors():
    ref = r.PackageRef("x", "npm", "firecrawl-mcp2", "1.0.0")
    info = r.lookup_registry_info(ref, lambda _ref: None, ["firecrawl-mcp"])
    assert "firecrawl-mcp" in info.near_neighbor_ids


# --------------------------------------------------------------------------- #
# contains_non_ascii
# --------------------------------------------------------------------------- #
def test_contains_non_ascii_true():
    assert r.contains_non_ascii("pаckage") is True


def test_contains_non_ascii_false():
    assert r.contains_non_ascii("package") is False