"""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 gate.HOSTILE_CASES == ( "traversal-in-bundle-id", "traversal-in-concept-id", "symlink-out-of-root", "broken-manifest", "oversized-concept", "unknown-bundle-id", ) 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["bridge-notes"] & words["kitchen-notes"] assert "spennvidde" 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": [ { "bundle": "bridge-notes", "atomer": [{"kilde_anker": "spennvidde", "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 == (("bridge-notes", "spennvidde", "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_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}"