FabledScribe/tests/test_services_rule_usage.py

"""Rule usage telemetry — the parts that need no database (milestone 333 step 1).

The round trip lives in `test_integration_backup_rule_usage_roundtrip.py`.
What is here is the payload building and the zero shape: cheap, and the half
where a mistake is silent rather than loud.
"""
import pytest

from scribe.models.rule_usage import PULLED, SURFACED, RuleUsageEvent
from scribe.services import rule_usage


@pytest.fixture
def captured(monkeypatch):
    """Intercept the scheduler so the payload can be read without a loop.

    Patching `_schedule` rather than `background.spawn` keeps the test on this
    module's own seam: what is under test is which rows get built, not whether
    the shared fire-and-forget machinery works — that has its own home.
    """
    rows: list[list[dict]] = []
    monkeypatch.setattr(rule_usage, "_schedule", rows.append)
    return rows


def test_a_surfacing_records_one_row_per_rule(captured):
    """The arm shows a hint containing several rules at once; each needs its
    own row, because the readout is per rule."""
    rule_usage.record_rule_surfaced(
        user_id=7, rule_ids=[156, 157], source="write_path_rule"
    )
    [batch] = captured
    assert batch == [
        {"user_id": 7, "rule_id": 156, "event": SURFACED, "source": "write_path_rule"},
        {"user_id": 7, "rule_id": 157, "event": SURFACED, "source": "write_path_rule"},
    ]


def test_the_whole_hint_lands_as_one_batch(captured):
    """One scheduled insert for the hint, not one per rule. A hint is a single
    decision and its rows should land together — a partial batch would read as
    a hint that surfaced fewer rules than it did."""
    rule_usage.record_rule_surfaced(
        user_id=7, rule_ids=[1, 2, 3], source="write_path_rule"
    )
    assert len(captured) == 1
    assert len(captured[0]) == 3


def test_a_pull_records_one_row(captured):
    rule_usage.record_rule_pulled(user_id=7, rule_id=156, source="mcp_get_rule")
    assert captured == [
        [{"user_id": 7, "rule_id": 156, "event": PULLED, "source": "mcp_get_rule"}]
    ]


def test_an_actorless_event_is_still_recorded(captured):
    """The arm fires from a hook that may carry no authenticated user. Dropping
    those would silently shrink the denominator the ratio divides by — the
    surfacings would vanish while any later pull still counted."""
    rule_usage.record_rule_surfaced(
        user_id=None, rule_ids=[156], source="write_path_rule"
    )
    assert captured[0][0]["user_id"] is None


def test_an_empty_surfacing_builds_no_rows(captured):
    """The arm can rank everything out — `exclude_rule_ids` drops what the
    session already holds. That is not a surfacing, and the empty batch is
    where `_schedule` returns early rather than opening a session to insert
    nothing."""
    rule_usage.record_rule_surfaced(user_id=7, rule_ids=[], source="write_path_rule")
    assert captured == [[]]


def test_the_real_scheduler_returns_early_on_an_empty_batch():
    """The guard itself, against the REAL `_schedule` the stub above replaces.

    There is no running loop in a unit test, so `spawn` would be harmless
    anyway — but it would build a coroutine only to close it, and the point is
    that an empty batch never gets that far.
    """
    rule_usage._schedule([])  # must not raise


def test_a_bad_rule_id_is_dropped_not_raised(captured):
    """Telemetry must never break the surface it observes. An unconvertible id
    is a bug somewhere upstream, and the right response is to lose the row and
    log it — not to take down the write-path hint."""
    rule_usage.record_rule_pulled(
        user_id=7, rule_id="not-an-int", source="mcp_get_rule"  # type: ignore[arg-type]
    )
    assert captured == []


def test_the_zero_readout_names_every_key():
    """Callers render this shape unconditionally. Every rule in an existing
    install predates the table, so for a while "no events" is the NORMAL state
    — a missing key here would read as a broken readout on almost every row."""
    assert rule_usage.empty_rule_usage() == {
        "surfaced_count": 0,
        "pull_count": 0,
        "last_surfaced_at": None,
        "last_pulled_at": None,
    }


def test_the_model_serialises_the_fields_the_ratio_needs():
    ev = RuleUsageEvent(
        user_id=7, rule_id=156, event=SURFACED, source="write_path_rule"
    )
    row = ev.to_dict()
    assert row["rule_id"] == 156
    assert row["event"] == SURFACED
    assert row["source"] == "write_path_rule"
    # created_at is server-defaulted, so it is None until the row is flushed —
    # `iso()` must tolerate that rather than raising on a fresh instance.
    assert row["created_at"] is None


# ─── the readout (milestone 333 step 5) ──────────────────────────────────────
# Integration: a real GROUP BY over a real table. Step 1 unit-tested the WRITE
# path and the zero shape and left the aggregate uncovered, which only became
# load-bearing when the rule list started rendering it.


@pytest.mark.integration
@pytest.mark.asyncio
async def test_usage_for_rules_aggregates_per_rule(_dispose_engine):
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.rule_usage import RuleUsageEvent

    async with async_session() as s:
        s.add_all([
            RuleUsageEvent(user_id=990020, rule_id=6001,
                           event=SURFACED, source="write_path_rule"),
            RuleUsageEvent(user_id=990020, rule_id=6001,
                           event=SURFACED, source="write_path_rule"),
            RuleUsageEvent(user_id=990020, rule_id=6001,
                           event=PULLED, source="mcp_get_rule"),
            RuleUsageEvent(user_id=990020, rule_id=6002,
                           event=SURFACED, source="write_path_rule"),
        ])
        await s.commit()
    try:
        out = await rule_usage.usage_for_rules([6001, 6002, 6003])

        assert out[6001]["surfaced_count"] == 2
        assert out[6001]["pull_count"] == 1
        assert out[6001]["last_surfaced_at"] is not None
        assert out[6001]["last_pulled_at"] is not None

        # Surfaced twice as often as it was opened — never, in this case.
        assert out[6002]["surfaced_count"] == 1
        assert out[6002]["pull_count"] == 0
        assert out[6002]["last_pulled_at"] is None

        # A rule with NO events still comes back, zero-filled. The caller must
        # never have to tell "no events" from "not in the result" — and on any
        # existing install that is nearly every rule.
        assert out[6003] == rule_usage.empty_rule_usage()
    finally:
        async with async_session() as s:
            await s.execute(
                delete(RuleUsageEvent).where(RuleUsageEvent.user_id == 990020)
            )
            await s.commit()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_usage_for_rules_on_an_empty_id_list_asks_the_database_nothing(
    _dispose_engine,
):
    """The list route calls this with whatever the page holds, which on an
    empty topic is nothing. An unguarded `IN ()` is both a pointless round trip
    and, on some drivers, a syntax error."""
    assert await rule_usage.usage_for_rules([]) == {}