FabledGrimoire/tests/test_usage_attach_seam.py

"""One seam attaches `usage`, and every door that shows it uses that seam.

WHY. `usage_for_notes` is named for notes and works on every note row, and
seven doors show its result — two REST lists, two REST details, two MCP
lists, one MCP detail. Seven copies of the same few lines each read perfectly
well on their own, and nobody compares them, so "which doors attach usage?"
has no answer anywhere in the code. That is the same failure
`test_system_tagging_door_parity.py` records for System tagging: whichever
door nobody exercises for a kind is the one that
never grows the feature, and a human reviewer does not reliably catch it
because each door is only ever read alone.

So this file asserts the PROPERTY, not the behaviour of one route: the attach
logic exists once, and no door re-implements it. A kind added next month either
goes through the seam or fails here.
"""
from __future__ import annotations

import ast
import pathlib
from unittest.mock import AsyncMock, patch

import pytest

from grimoire.services.note_usage import attach_usage, empty_usage

ROOT = pathlib.Path(__file__).resolve().parents[1] / "src" / "grimoire"

# The aggregate the seam is built around. Calling it from a door is the shape
# this file exists to prevent — not because the call is wrong, but because
# seven of them drift.
AGGREGATE = "usage_for_notes"


def _counts(surfaced: int = 5, pulled: int = 0) -> dict:
    u = empty_usage()
    u["surfaced_count"] = surfaced
    u["pull_count"] = pulled
    return u


def _aggregate_returns(mapping: dict[int, dict]) -> AsyncMock:
    return patch(
        "grimoire.services.note_usage.usage_for_notes",
        AsyncMock(return_value=mapping),
    )


# ── the seam itself ───────────────────────────────────────────────────────


async def test_every_row_gets_the_key_even_with_no_events() -> None:
    """Zero-filled, never absent. The UI must not have to tell "no events"
    from "no field" — and `UsageBadge` renders nothing below one surfacing, so
    an un-surfaced record is quiet without the caller doing anything."""
    rows = [{"id": 1}, {"id": 2}]
    with _aggregate_returns({1: _counts(surfaced=3)}):
        await attach_usage(rows)
    assert rows[0]["usage"]["surfaced_count"] == 3
    assert rows[1]["usage"] == empty_usage()


async def test_one_aggregate_for_the_whole_page() -> None:
    """The N+1 guard. A per-row lookup here would be N+1 by construction, which
    is the one shape a list route must not have — and it is invisible in
    review, because the per-row version reads more naturally."""
    rows = [{"id": n} for n in range(25)]
    mock = AsyncMock(return_value={})
    with patch("grimoire.services.note_usage.usage_for_notes", mock):
        await attach_usage(rows)
    assert mock.await_count == 1, "usage must be read once per page, not per row"
    assert sorted(mock.await_args.args[0]) == list(range(25))


async def test_a_detail_payload_is_just_a_one_row_list() -> None:
    """The single-record doors share the seam rather than keeping a second
    shape beside it. Two shapes for one job is how the seven copies started."""
    data = {"id": 7, "title": "x"}
    with _aggregate_returns({7: _counts(surfaced=9, pulled=2)}):
        await attach_usage([data])
    assert data["usage"]["pull_count"] == 2


async def test_a_row_with_no_id_is_skipped_rather_than_failing_the_list() -> None:
    """An unusable id is not a reason to 500 a page of otherwise fine rows."""
    rows = [{"id": 1}, {"title": "no id here"}]
    with _aggregate_returns({1: _counts()}):
        await attach_usage(rows)
    assert "usage" in rows[0]
    assert "usage" not in rows[1]


async def test_a_boolean_is_not_an_id() -> None:
    """`int(True)` is 1, so a row carrying a flag under the key would silently
    be credited with note #1's counts. A wrong chip is worse than no chip: it
    reads as a measurement."""
    rows = [{"id": True}]
    with _aggregate_returns({1: _counts(surfaced=40)}):
        await attach_usage(rows)
    assert "usage" not in rows[0]


async def test_a_string_id_still_resolves() -> None:
    """Payload rows come from several serialisers; one of them handing back a
    stringified id should not silently drop the chip."""
    rows = [{"id": "12"}]
    with _aggregate_returns({12: _counts(surfaced=4)}):
        await attach_usage(rows)
    assert rows[0]["usage"]["surfaced_count"] == 4


@pytest.mark.parametrize("key", ["note_id", "record_id"])
async def test_the_key_can_be_named(key: str) -> None:
    rows = [{key: 3}]
    with _aggregate_returns({3: _counts()}):
        await attach_usage(rows, key=key)
    assert "usage" in rows[0]


async def test_an_empty_page_asks_nothing_and_breaks_nothing() -> None:
    mock = AsyncMock(return_value={})
    with patch("grimoire.services.note_usage.usage_for_notes", mock):
        await attach_usage([])
    assert mock.await_args.args[0] == []


# ── the property: one seam, and every door uses it ────────────────────────


def _calls(tree: ast.Module) -> set[str]:
    out = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            fn = node.func
            name = fn.attr if isinstance(fn, ast.Attribute) else getattr(fn, "id", None)
            if name:
                out.add(name)
    return out


def _door_modules() -> list[pathlib.Path]:
    return sorted(
        [*(ROOT / "routes").glob("*.py"), *(ROOT / "mcp" / "tools").glob("*.py")]
    )


def test_no_door_calls_the_aggregate_directly() -> None:
    """THE GUARD. Seven doors each called `usage_for_notes` and zero-filled by
    hand; the eighth would have been `/api/knowledge`, and the chip would have
    kept reaching some kinds and not others.

    Keyed on the CALL, not on the text, so a module that merely names the
    function in a comment explaining the seam is not a false positive — and a
    hand-kept skip list, which would itself go stale, is not needed.
    """
    offenders = []
    for path in _door_modules():
        if AGGREGATE in _calls(ast.parse(path.read_text())):
            offenders.append(str(path.relative_to(ROOT.parent.parent)))
    assert not offenders, (
        f"these doors call {AGGREGATE}() themselves instead of attach_usage(); "
        f"that is how the chip came to reach two record kinds out of four: "
        f"{offenders}"
    )


# (module, the functions that return note-bearing payloads)
#
# Not a list of everything that COULD attach usage — a list of the doors that
# demonstrably show it today. A door dropping its call silently is the exact
# regression this pins.
DOORS = [
    ("routes/lessons.py", "list_lessons_route or get_lesson_route"),
    ("routes/snippets.py", "list/get snippet routes"),
    ("routes/knowledge.py", "list_knowledge — the only note & lesson list in the UI"),
    ("mcp/tools/lessons.py", "list_lessons / get_lesson"),
    ("mcp/tools/snippets.py", "list_snippets"),
]


@pytest.mark.parametrize(("module", "why"), DOORS)
def test_every_door_that_shows_usage_goes_through_the_seam(module: str, why: str) -> None:
    assert "attach_usage" in _calls(ast.parse((ROOT / module).read_text())), (
        f"{module} no longer attaches usage ({why}). If that is deliberate, "
        f"remove it from DOORS and say why; a door that silently stops "
        f"attaching looks exactly like a corpus nobody uses."
    )


def test_the_knowledge_browse_is_covered_because_it_is_the_only_note_list() -> None:
    """Pinned on its own, with the reason, because it is the non-obvious one.

    `/api/lessons` already attached usage and it did not help: no view calls
    it. `KnowledgeView` is the only list in the UI that renders notes and
    lessons, so `/api/knowledge` is the only route through which those two
    kinds can show the counter at all. Deleting this line would restore the
    original bug while every other test here still passed.
    """
    assert any(m == "routes/knowledge.py" for m, _ in DOORS)
    assert "attach_usage" in _calls(ast.parse((ROOT / "routes" / "knowledge.py").read_text()))