FabledScribe/tests/test_integration_backup_rule_usage_roundtrip.py

"""Real-Postgres round trip for rule_usage_events (milestone 333 step 1).

**This file is the reason the table exists.** `rule_usage_events` could have
been a `rule_id` column on `note_usage_events` — the row carries no
note-specific field and the readout is the same shape, which is the strongest
case for sharing that note #3163 admits. What decided against it is identity at
restore, and that is a claim only a real round trip can support.

The failure it guards is the quiet kind. `note_usage_events`'s importer maps
`note_id` through `note_id_map`; a rule id parked in that column comes back
attached to whatever note happens to hold that number in the target database.
Not dropped — REATTACHED. The restore reports success, the counters are
populated, and every one of them is about the wrong record. Nothing downstream
can detect it, because a usage row has no other field to disagree with.

So the assertions below are about WHICH MAP resolved the id, and they are
written to fail if the answer ever becomes "the note one" or "neither".

Same shape as `test_integration_backup_rule_version_roundtrip.py`, which guards
`rule_versions` against #3182's `arose_from_id` trap on the same seam.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from scribe.models import async_session
from scribe.models.note import Note
from scribe.models.rule_usage import PULLED, SURFACED, RuleUsageEvent
from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
from scribe.models.user import User
from scribe.services import backup
from tests.helpers import ensure_user

pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]

OWNER_USERNAME = "rule_usage_roundtrip_owner"
RESTORED_USERNAME = "rule_usage_roundtrip_restored"


async def _purge_books(username: str) -> None:
    """user -> rulebook -> topic -> rule is ON DELETE CASCADE the whole way,
    so dropping the books clears the rules this file made.

    `rule_usage_events` is deliberately FK-FREE, so its rows do NOT cascade —
    that is the property under test elsewhere (telemetry outlives what it
    describes). They are cleared explicitly below.
    """
    async with async_session() as s:
        users = (await s.execute(
            select(User).where(User.username == username)
        )).scalars().all()
        for user in users:
            books = (await s.execute(
                select(Rulebook).where(Rulebook.owner_user_id == user.id)
            )).scalars().all()
            for book in books:
                await s.delete(book)
            for note in (await s.execute(
                select(Note).where(Note.user_id == user.id)
            )).scalars().all():
                await s.delete(note)
        await s.commit()


async def _purge_usage(rule_ids: set[int]) -> None:
    if not rule_ids:
        return
    async with async_session() as s:
        for ev in (await s.execute(
            select(RuleUsageEvent).where(RuleUsageEvent.rule_id.in_(rule_ids))
        )).scalars().all():
            await s.delete(ev)
        await s.commit()


async def _purge_restored() -> None:
    await _purge_books(RESTORED_USERNAME)
    async with async_session() as s:
        for user in (await s.execute(
            select(User).where(User.username == RESTORED_USERNAME)
        )).scalars().all():
            await s.delete(user)
        await s.commit()


@pytest_asyncio.fixture(autouse=True)
async def _no_leftovers():
    """SETUP ONLY — see the sibling file for why a database call after a
    `yield` here orphans a pooled connection and breaks unrelated tests."""
    await _purge_restored()
    await _purge_books(OWNER_USERNAME)


@pytest_asyncio.fixture
async def source():
    """One rule with a surfaced/pulled pair — plus a NOTE that will hold the
    rule's id in the restored database.

    That note is the whole trick. Without it, a restore that ran rule ids
    through `note_id_map` would simply drop them and the test would read as a
    pass-by-absence. With it, the wrong map produces a plausible, populated,
    entirely wrong result — which is the failure actually being guarded.
    """
    async with async_session() as s:
        owner = await ensure_user(s, OWNER_USERNAME)
        uid = owner.id
        await s.commit()

    async with async_session() as s:
        book = Rulebook(owner_user_id=uid, title="Environment facts")
        s.add(book)
        await s.flush()
        topic = RulebookTopic(rulebook_id=book.id, title="ci")
        s.add(topic)
        await s.flush()
        rule = Rule(
            topic_id=topic.id,
            title="A wait with no deadline is a bug",
            statement="Every wait on something that can fail to answer carries one.",
        )
        s.add(rule)
        # A note in the same export, so the target database has a note id to
        # collide with. Its own id is irrelevant; what matters is that the
        # note map is populated and would resolve to something.
        note = Note(user_id=uid, title="a note that must not receive rule telemetry",
                    body="decoy")
        s.add(note)
        await s.flush()
        s.add_all([
            RuleUsageEvent(
                user_id=uid, rule_id=rule.id,
                event=SURFACED, source="write_path_rule",
            ),
            RuleUsageEvent(
                user_id=uid, rule_id=rule.id,
                event=PULLED, source="mcp_get_rule",
            ),
            # No actor. The arm can fire for an unauthenticated hook call, and
            # a user who later leaves must not take the evidence with them.
            RuleUsageEvent(
                user_id=None, rule_id=rule.id,
                event=SURFACED, source="write_path_rule",
            ),
        ])
        await s.commit()
        book_id, rule_id, note_id = book.id, rule.id, note.id

    async with async_session() as s:
        user_rows = backup._user_rows(
            [(await s.execute(select(User).where(User.id == uid))).scalars().one()]
        )
        book_rows = backup._rulebook_rows(
            [(await s.execute(select(Rulebook).where(Rulebook.id == book_id)))
             .scalars().one()]
        )
        topic_rows = backup._topic_rows(
            (await s.execute(
                select(RulebookTopic).where(RulebookTopic.rulebook_id == book_id)
            )).scalars().all()
        )
        rule_rows = backup._rule_rows(
            [(await s.execute(select(Rule).where(Rule.id == rule_id))).scalars().one()]
        )
        note_rows = backup._note_rows(
            [(await s.execute(select(Note).where(Note.id == note_id))).scalars().one()]
        )
        usage_rows = backup._rule_usage_event_rows(
            (await s.execute(
                select(RuleUsageEvent).where(RuleUsageEvent.rule_id == rule_id)
                .order_by(RuleUsageEvent.id)
            )).scalars().all()
        )
    user_rows[0]["username"] = RESTORED_USERNAME

    yield {
        "payload": {
            "version": backup.BACKUP_VERSION,
            "users": user_rows,
            "rulebooks": book_rows,
            "rulebook_topics": topic_rows,
            "rules": rule_rows,
            "notes": note_rows,
            "rule_usage_events": usage_rows,
        },
        "source_rule_id": rule_id,
        "source_user_id": uid,
    }

    await _purge_usage({rule_id})
    async with async_session() as s:
        book = await s.get(Rulebook, book_id)
        if book is not None:
            await s.delete(book)
        note = await s.get(Note, note_id)
        if note is not None:
            await s.delete(note)
        await s.commit()


@pytest_asyncio.fixture
async def restored(source):
    await backup.restore_full_backup(source["payload"])
    async with async_session() as s:
        user = (await s.execute(
            select(User).where(User.username == RESTORED_USERNAME)
        )).scalars().first()
        assert user is not None, "the payload's user was not restored"
        book = (await s.execute(
            select(Rulebook).where(Rulebook.owner_user_id == user.id)
        )).scalars().one()
        topic = (await s.execute(
            select(RulebookTopic).where(RulebookTopic.rulebook_id == book.id)
        )).scalars().one()
        rule = (await s.execute(
            select(Rule).where(Rule.topic_id == topic.id)
        )).scalars().one()
        note = (await s.execute(
            select(Note).where(Note.user_id == user.id)
        )).scalars().one()
        events = (await s.execute(
            select(RuleUsageEvent).where(RuleUsageEvent.rule_id == rule.id)
            .order_by(RuleUsageEvent.id)
        )).scalars().all()
    yield {
        "user": user, "rule": rule, "note": note,
        "events": events, "source": source,
    }

    await _purge_usage({rule.id})
    await _purge_restored()


async def test_every_event_comes_back(restored):
    """The count first: every shape assertion below reads the same on an empty
    list, so without this a restore that dropped all three would pass them."""
    assert len(restored["events"]) == 3


async def test_the_events_attach_to_the_RESTORED_rule(restored):
    """The remap, on the column that matters."""
    new_rule_id = restored["rule"].id
    source_rule_id = restored["source"]["source_rule_id"]
    assert new_rule_id != source_rule_id, (
        "the restore reused the source id, so this test cannot tell a remap "
        "from a copy — the fixture is not proving what it claims"
    )
    assert {e.rule_id for e in restored["events"]} == {new_rule_id}


async def test_no_event_landed_on_the_note_id(restored):
    """THE ONE THIS TABLE EXISTS FOR.

    If `rule_id` were ever resolved through `note_id_map` — the shape it would
    have had as a column on `note_usage_events` — these rows would come back
    pointing at the restored NOTE's id. Populated, plausible, and describing a
    record that was never surfaced.
    """
    note_id = restored["note"].id
    landed_on_note = [e for e in restored["events"] if e.rule_id == note_id]
    assert not landed_on_note, (
        f"{len(landed_on_note)} usage event(s) resolved to the note's id "
        f"({note_id}) instead of the rule's. The rule id went through the "
        "note map — telemetry that is wrong rather than missing, and that "
        "nothing downstream can detect."
    )


async def test_the_actor_is_remapped_and_a_missing_one_survives(restored):
    """`user_id` is an id in the source database too — the same trap one
    column over. And the actorless row must not be dropped: the arm can fire
    for an unauthenticated hook call, so requiring an actor would discard the
    surfacings of exactly the surface being measured."""
    attributed = [e for e in restored["events"] if e.user_id is not None]
    orphaned = [e for e in restored["events"] if e.user_id is None]
    assert len(attributed) == 2
    assert len(orphaned) == 1, (
        "the event with no actor did not come back. Telemetry outlives the "
        "account it was recorded for; dropping it silently lowers the "
        "surfaced count that the pull-through ratio divides by."
    )
    assert {e.user_id for e in attributed} == {restored["user"].id}
    assert restored["user"].id != restored["source"]["source_user_id"]


async def test_the_event_and_source_survive(restored):
    """The two fields the ratio is computed from. A restore that kept the rows
    and lost these would preserve a count of nothing in particular."""
    pairs = {(e.event, e.source) for e in restored["events"]}
    assert pairs == {
        (SURFACED, "write_path_rule"),
        (PULLED, "mcp_get_rule"),
    }
    assert sum(1 for e in restored["events"] if e.event == SURFACED) == 2
    assert sum(1 for e in restored["events"] if e.event == PULLED) == 1