llm-ingestion-okf/tests/test_mcp_gate.py

"""The MCP gate's own tests (`tools/okf_mcp_gate.py`).

Two halves, and the split is the point.

THE CAPABILITY TESTS ARE RED UNTIL THE SERVER EXISTS, and each one is red on
an ASSERT about behaviour -- never on an import, never on an AttributeError.
A test that dies with `ModuleNotFoundError` proves that a name is missing; a
test that dies with `0 == 7` proves that seven required tools did not answer
over a real protocol, which is the claim the order asked to be able to fell.

THE INSTRUMENT TESTS ARE GREEN NOW, and every denominator they pin is counted
a SECOND time inside the test. A guard that read the gate's own total would
agree with any total the gate happened to hold -- the defect this repository
has already met twice, most recently in the retrieval gate's rows 2 and 3,
where a row counted what the run produced and went green by producing less.

THE SYNTHETIC CORPUS IS THE ROW-2 MACHINERY'S KNOWN-POSITIVE, and the test
below proves the quotes really are in the bundles ON DISK, without asking any
server. Without it, `0 of 4` could mean the fixture never carried the text.
"""

from __future__ import annotations

import importlib.util
import json
import sys
import tempfile
from pathlib import Path

import pytest

TOOLS = Path(__file__).resolve().parents[1] / "tools"
sys.path.insert(0, str(TOOLS))

import okf_mcp_gate as gate  # noqa: E402


@pytest.fixture()
def scratch() -> Path:
    with tempfile.TemporaryDirectory(prefix="okf-mcp-gate-test-") as directory:
        yield Path(directory)


# --- the instrument ---------------------------------------------------------


def test_the_required_tool_roster_is_pinned_and_not_taken_from_a_run() -> None:
    """Counted again here, by name. A roster the gate could shrink is a
    denominator the capability could satisfy by offering less."""
    one_to_one = ("okf_describe", "okf_ask", "okf_fetch")
    one_to_many = ("okf_list", "okf_describe", "okf_ask", "okf_fetch")
    assert gate.REQUIRED_TOOLS["one-to-one"] == one_to_one
    assert gate.REQUIRED_TOOLS["one-to-many"] == one_to_many
    assert sum(len(names) for names in gate.REQUIRED_TOOLS.values()) == len(one_to_one) + len(
        one_to_many
    )


def test_every_pinned_denominator_is_recounted_here() -> None:
    """Four artefact classes, three discovery checks over three bundles, three
    cross-bundle checks, six hostile cases -- each listed again."""
    assert gate.DRILL_ARTEFACTS == (
        "mcp-one-to-one",
        "mcp-one-to-many",
        "skill-one-to-one",
        "skill-one-to-many",
    )
    assert gate.DISCOVERY_BUNDLES * len(gate.DISCOVERY_CHECKS) == 3 * 3
    assert len(gate.CROSS_CHECKS) == 3
    assert tuple(gate.HOSTILE_CASES) == (
        "traversal-in-bundle-id",
        "traversal-in-concept-id",
        "symlink-out-of-root",
        "broken-manifest",
        "oversized-concept",
        "unknown-bundle-id",
    )
    # The 10 MB case has exactly one acceptable refusal. Anything else means
    # the ceiling did not run -- the false green this row shipped with once.
    assert gate.HOSTILE_CASES["oversized-concept"] == frozenset({"concept_too_large"})
    assert gate.OVERSIZED_BYTES == 10 * 1024 * 1024


def test_a_row_that_counted_nothing_is_never_green() -> None:
    """`0 of 0` is the shape a green verdict hides behind."""
    assert gate._row(1, "n", 0, 0, "r", []).status == gate.RED
    assert gate._row(1, "n", 3, 3, "r", []).status == gate.GREEN
    assert gate._row(1, "n", 2, 3, "r", []).status == gate.RED


def test_the_synthetic_quotes_are_really_in_the_bundles_on_disk(scratch: Path) -> None:
    """Row 2's known-positive, proved without a server.

    If this is green and row 2's known-positive is 0 of 4, the surface could
    not reach text that is demonstrably there.
    """
    bundles = gate.corpus(scratch)
    for bundle_id, anchor, quote in gate.SYNTHETIC_ANCHORS:
        concept = bundles[bundle_id] / f"{anchor}.md"
        assert concept.is_file(), f"{bundle_id}/{anchor} is not in the fixture"
        assert quote in concept.read_text(encoding="utf-8")


def test_the_two_synthetic_bundles_share_no_question_token(scratch: Path) -> None:
    """Row 5 asks for an answer spanning two bundles. If the bundles shared
    vocabulary, one of them could carry the whole answer and the row would
    pass without ever crossing."""
    from llm_ingestion_okf import consume

    bundles = gate.corpus(scratch)
    words = {
        name: {
            token
            for path in sorted(root.rglob("*.md"))
            if path.name != "index.md"
            for token in consume.normalise(path.read_text(encoding="utf-8"))
        }
        for name, root in bundles.items()
    }
    shared = words["garden-notes"] & words["kitchen-notes"]
    assert "bedbredde" not in shared
    assert "surdeig" not in shared


def test_a_question_set_whose_bytes_moved_is_a_usage_error_not_a_red_row(
    scratch: Path,
) -> None:
    """A gate that measured a moving set would report on nothing. Driven from
    BOTH sides: the matching pair is accepted, the moved one is refused."""
    questions = scratch / "sporsmal.json"
    payload = {
        "sporsmal": [
            {
                "bundles": ["garden-notes"],
                "atomer": [{"kilde_anker": "bedbredde", "kilde_sitat": "24 meter"}],
            }
        ]
    }
    questions.write_text(json.dumps(payload), encoding="utf-8")
    import hashlib

    digest = hashlib.sha256(questions.read_bytes()).hexdigest()
    freeze = scratch / "frys.json"
    freeze.write_text(
        json.dumps({"versjon": 4, "sha256": {"sporsmal.json": digest}}), encoding="utf-8"
    )
    anchors = gate.read_anchor_set(questions, freeze, want_version=4)
    assert anchors.pairs == (("garden-notes", "bedbredde", "24 meter"),)

    questions.write_text(json.dumps(payload) + " ", encoding="utf-8")
    with pytest.raises(gate.GateUsage, match="is not the file"):
        gate.read_anchor_set(questions, freeze, want_version=4)


def test_an_older_freeze_version_is_refused_rather_than_measured(scratch: Path) -> None:
    """The set is versioned by its scoring regime. Measuring version 3 and
    reporting it as the ordered version 4 is how a number outlives its own
    definition."""
    import hashlib

    questions = scratch / "sporsmal.json"
    questions.write_text(json.dumps({"sporsmal": []}), encoding="utf-8")
    freeze = scratch / "frys.json"
    freeze.write_text(
        json.dumps(
            {
                "versjon": 3,
                "sha256": {"sporsmal.json": hashlib.sha256(questions.read_bytes()).hexdigest()},
            }
        ),
        encoding="utf-8",
    )
    with pytest.raises(gate.GateUsage, match="declares version 3"):
        gate.read_anchor_set(questions, freeze, want_version=4)


def test_a_refusal_code_is_read_from_the_message_a_client_sees() -> None:
    """Driven from both sides: a coded refusal yields its code, an uncoded one
    yields the empty string rather than a plausible guess."""
    assert gate.refusal_code(gate.RpcError(-32000, "refused (concept_too_large): 10 MB")) == (
        "concept_too_large"
    )
    assert gate.refusal_code(gate.RpcError(-32000, "something went wrong")) == ""


def test_the_gate_reads_bundle_and_concept_ids_at_any_depth() -> None:
    """An `ask` answer carries one id per excerpt and a `list` answer one per
    bundle. A rule reading the top level only would score the shape."""
    answer = {"excerpts": [{"bundle_id": "a", "concept_id": "x"}, {"bundle_id": "b"}]}
    assert gate._bundle_ids(answer) == {"a", "b"}
    assert gate._source_ids(answer) == {"x"}
    assert gate._bundle_ids({"excerpts": []}) == set()


# --- the capability: red until the server exists -----------------------------


def test_the_server_module_exists() -> None:
    """Red on an ASSERT, never on an import: the claim is that the module is
    reachable, and a `ModuleNotFoundError` would prove only that this file
    spelled a name."""
    assert importlib.util.find_spec(gate.SERVER_MODULE) is not None, (
        f"{gate.SERVER_MODULE} is not importable; the MCP surface has not been built"
    )


def _row(number: int, scratch: Path) -> gate.Row:
    rows = {row.number: row for row in gate.evaluate(scratch)}
    return rows[number]


def test_every_required_tool_answers_over_real_stdio(scratch: Path) -> None:
    row = _row(1, scratch)
    assert (row.k, row.m) == (row.m, 7), f"{row.k} of {row.m}: {row.details}"


def test_a_rebuilt_bundle_never_gets_yesterdays_answer(scratch: Path) -> None:
    """Scenario 1. A silent stale answer is red regardless of every other
    number, so this asserts the whole row rather than the MCP arms alone."""
    row = _row(3, scratch)
    assert row.k == row.m == len(gate.DRILL_ARTEFACTS), f"{row.k} of {row.m}: {row.details}"


def test_a_bundle_added_while_the_server_runs_is_discovered(scratch: Path) -> None:
    row = _row(4, scratch)
    assert row.k == row.m == 9, f"{row.k} of {row.m}: {row.details}"


def test_one_documented_sequence_answers_from_two_bundles(scratch: Path) -> None:
    row = _row(5, scratch)
    assert row.k == row.m == 3, f"{row.k} of {row.m}: {row.details}"


def test_hostile_input_is_refused_out_loud(scratch: Path) -> None:
    row = _row(6, scratch)
    assert row.k == row.m == 6, f"{row.k} of {row.m}: {row.details}"


def test_row_twos_denominator_is_the_sets_and_not_what_the_surface_reached(
    scratch: Path,
) -> None:
    """The defect this repository has already met twice: a row that counts
    against what the run produced closes by producing less.

    Three pairs are declared and one of them names an anchor no bundle carries,
    so a denominator taken from the run would read `2 of 2` and call the gate
    satisfied. Driven from both sides -- the reachable pairs really are
    reachable, so a `0 of 3` would be a different defect.
    """
    import hashlib

    bundles = gate.corpus(scratch / "base")
    # The second reachable pair is the fixture's own second triple, read rather
    # than restated, so the test and the corpus cannot drift apart.
    second_bundle, second_anchor, second_quote = gate.SYNTHETIC_ANCHORS[1]
    assert second_bundle == "garden-notes"
    questions = scratch / "sporsmal.json"
    questions.write_text(
        json.dumps(
            {
                "sporsmal": [
                    {
                        "bundles": ["garden-notes"],
                        "atomer": [
                            {"kilde_anker": "bedbredde", "kilde_sitat": "bedbredde 24 meter"},
                            {"kilde_anker": second_anchor, "kilde_sitat": second_quote},
                            {
                                "kilde_anker": "finnes-ikke",
                                "kilde_sitat": "dette staar ingen steder",
                            },
                        ],
                    }
                ]
            }
        ),
        encoding="utf-8",
    )
    freeze = scratch / "frys.json"
    freeze.write_text(
        json.dumps(
            {
                "versjon": 4,
                "sha256": {"sporsmal.json": hashlib.sha256(questions.read_bytes()).hexdigest()},
            }
        ),
        encoding="utf-8",
    )
    anchors = gate.read_anchor_set(questions, freeze, want_version=4)
    assert len(anchors.pairs) == 3
    rows = {
        row.number: row
        for row in gate.evaluate(
            scratch / "run",
            anchors=anchors,
            real={"garden-notes": bundles["garden-notes"]},
        )
    }
    row = rows[2]
    assert row.m == 3, f"the denominator came from the run: {row.k} of {row.m}"
    assert row.k == 2, f"{row.k} of {row.m}: {row.details}"
    assert row.status == gate.RED