FabledScribe/tests/test_integration_rule_versions.py

"""Real-Postgres tests for the rule-history write path (milestone 323 step 2).

Every assertion here is about ORDERING or ABSENCE, which is why none of them
can be made with mocks.

What a version is for: it holds what the rule USED TO SAY. The edit destroys
that, and nothing can recompute it — the rescoping of rule 79 had to be
hand-copied into a task log to survive (#3237), which is not a process, it is
a person remembering.

Two of these pin a DELIBERATE ABSENCE. `note_versions` carries a 300-second
interval gate and a 50-version cap, both defences against note autosave. Rules
have no autosave, so here those guards would only ever discard a real edit.
Nothing in the code says so on its own, and "make it consistent with the
sibling" is a plausible-sounding change that would silently start dropping
history — so the absence is a test, not a comment.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from scribe.models import async_session
from scribe.models.rulebook import Rule, Rulebook
from scribe.models.rule_version import RuleVersion
from scribe.services import rule_versions as rv_svc
from scribe.services import rulebooks as rulebooks_svc
from tests.helpers import ensure_user

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

OWNER_USERNAME = "rule_history_owner"


@pytest_asyncio.fixture
async def constraint():
    """One rule carrying a check, with no history yet.

    CLEANED UP AT SETUP, NOT TEARDOWN, and that is forced. `update_rule` fires
    a detached `asyncio.create_task(upsert_rule_embedding(...))` that opens
    its own connection and UPDATEs the rule row. A teardown that deleted the
    rulebook would race it: the delete cascade-locks the rule the embedding
    task is writing, and Postgres kills one of them with a deadlock. Purging
    at setup instead runs on a fresh loop, after the previous test's loop
    closed and cancelled whatever it left in flight.
    """
    async with async_session() as s:
        owner = await ensure_user(s, OWNER_USERNAME)
        uid = owner.id
        await s.commit()
        # A previous failed run would leave a book whose topic title collides.
        for book in (await s.execute(
            select(Rulebook).where(Rulebook.owner_user_id == uid)
        )).scalars().all():
            await s.delete(book)
        await s.commit()

    book = await rulebooks_svc.create_rulebook(uid, "History fixtures")
    topic = await rulebooks_svc.create_topic(book.id, uid, "ci")
    rule = await rulebooks_svc.create_rule(
        when_to_apply="when the moment this fixture stands in for arises",
        topic.id, uid, "The runner has no bash",
        "Write every `run:` step in POSIX sh.",
        why="the image ships no bash",
        verify_with="read the workflow's shell setting",
    )

    return {"uid": uid, "rule_id": rule.id}


async def _versions(rule_id: int) -> list[RuleVersion]:
    async with async_session() as s:
        return list((await s.execute(
            select(RuleVersion).where(RuleVersion.rule_id == rule_id)
            .order_by(RuleVersion.id)
        )).scalars().all())


async def test_an_edit_leaves_the_text_it_replaced(constraint):
    """The headline. The version holds the OLD statement — the rule row
    already holds the new one, so a snapshot of the new text would record
    nothing that was not already there."""
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"],
        statement="Write every `run:` step in POSIX sh, and set shell: sh.",
    )
    [version] = await _versions(constraint["rule_id"])
    assert version.statement == "Write every `run:` step in POSIX sh."
    assert version.why == "the image ships no bash"


async def test_the_actor_is_recorded(constraint):
    """`user_id` is who made the edit — the question an audit trail over a
    binding instruction is actually asked."""
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"], title="The runner ships no bash",
    )
    [version] = await _versions(constraint["rule_id"])
    assert version.user_id == constraint["uid"]


async def test_an_edit_that_changes_no_text_writes_nothing(constraint):
    """Git semantics. `update_rule` is reached by callers that resend every
    field, so without the skip a form saved twice would file an identical
    snapshot and the history would stop reading as a list of edits."""
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"],
        statement="Write every `run:` step in POSIX sh.",
        title="The runner has no bash",
    )
    assert await _versions(constraint["rule_id"]) == []


async def test_reordering_is_not_an_edit(constraint):
    """`order_index` goes through the same door but is not text. A run of
    "moved rule 4 above rule 3" snapshots would bury the edits somebody is
    looking for."""
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"], order_index=7,
    )
    assert await _versions(constraint["rule_id"]) == []


async def test_the_snapshot_holds_the_CHECK_that_was_in_force(constraint):
    """The ordering trap. `update_rule` drops `verified_at` when `verify_with`
    changes, and rewriting the check is exactly the edit whose history matters
    most — so the snapshot has to be taken before the field loop, or it
    records the new check against the old wording."""
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"],
        verify_with="run `sh -c 'echo $0'` in a job step",
    )
    [version] = await _versions(constraint["rule_id"])
    assert version.verify_with == "read the workflow's shell setting"

    async with async_session() as s:
        rule = await s.get(Rule, constraint["rule_id"])
        assert rule.verify_with == "run `sh -c 'echo $0'` in a job step"


async def test_clearing_a_field_is_an_edit_worth_recording(constraint):
    """A rule that stops being a constraint loses its check entirely, and the
    version is then the only place the retired check exists. `clear` is a
    separate argument from the field loop — a snapshot wired to only one of
    the two would lose exactly this case."""
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"], clear=("verify_with",),
    )
    [version] = await _versions(constraint["rule_id"])
    assert version.verify_with == "read the workflow's shell setting"

    async with async_session() as s:
        rule = await s.get(Rule, constraint["rule_id"])
        assert rule.verify_with is None


async def test_rapid_successive_edits_are_all_kept(constraint):
    """DELIBERATE ABSENCE — no MIN_VERSION_INTERVAL_SECONDS.

    `note_versions` skips a snapshot taken within 300 seconds of the last one,
    because note autosave fires every 60 and would burn all 50 slots. Rules
    have no autosave: every one of these is a deliberate update_rule, and
    three edits in the same second are three edits. Porting the sibling's gate
    here would silently keep only the first.
    """
    for n in (1, 2, 3):
        await rulebooks_svc.update_rule(
            constraint["rule_id"], constraint["uid"], statement=f"revision {n}",
        )
    versions = await _versions(constraint["rule_id"])
    assert [v.statement for v in versions] == [
        "Write every `run:` step in POSIX sh.", "revision 1", "revision 2",
    ]


async def test_the_service_carries_no_pruning_machinery(constraint):
    """DELIBERATE ABSENCE — no MAX_VERSIONS and no pruning DELETE.

    Asserted structurally because the alternative is 51 real edits to prove a
    negative. The cap exists in `note_versions` to bound autosave volume; a
    rule is edited a handful of times in its life, so a cap here could only
    ever discard the one edit somebody went looking for.
    """
    assert not hasattr(rv_svc, "MAX_VERSIONS")
    assert not hasattr(rv_svc, "MIN_VERSION_INTERVAL_SECONDS")


async def test_history_reads_newest_first(constraint):
    """The reader's question is "what did this just say?", so the most recent
    supersession has to be the first row."""
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"], statement="first change",
    )
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"], statement="second change",
    )
    listed = await rv_svc.list_versions(constraint["rule_id"])
    assert [v.statement for v in listed] == [
        "first change", "Write every `run:` step in POSIX sh.",
    ]


# ── The read path is ACL-scoped (milestone 323 step 3) ─────────────────
#
# Rule 47: every read of user data is scoped by owner. A version carries a
# `user_id`, which makes it tempting to scope the history by it — that would
# be wrong in both directions, and these pin which way round it goes.


@pytest_asyncio.fixture
async def stranger():
    """A second user who owns nothing in the fixture above."""
    async with async_session() as s:
        other = await ensure_user(s, "rule_history_stranger")
        uid = other.id
        await s.commit()
    return uid


async def test_the_owner_reads_the_history(constraint):
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"], statement="reworded",
    )
    versions = await rulebooks_svc.list_rule_versions(
        constraint["rule_id"], constraint["uid"],
    )
    assert [v.statement for v in versions] == [
        "Write every `run:` step in POSIX sh.",
    ]


async def test_a_stranger_reads_nothing(constraint, stranger):
    """None, not an empty list. The two mean different things — "not your
    rule" versus "this rule has never been reworded" — and collapsing them
    would tell a caller the rule exists and is unedited."""
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"], statement="reworded",
    )
    assert await rulebooks_svc.list_rule_versions(
        constraint["rule_id"], stranger,
    ) is None


async def test_a_version_cannot_be_read_through_a_DIFFERENT_rule(constraint):
    """The check and the fetch have to agree about which rule is in play.

    The second rule is owned by the SAME user on purpose — an id that simply
    does not exist would pass on the ownership check alone and prove nothing
    about the `rule_id` clause. Here the caller genuinely owns the rule they
    name and genuinely owns the version's rule, and the read must still
    refuse, because scoping by version id alone is how a snapshot leaks
    through whichever rule the caller happens to be able to see.
    """
    await rulebooks_svc.update_rule(
        constraint["rule_id"], constraint["uid"], statement="reworded",
    )
    [version] = await _versions(constraint["rule_id"])

    async with async_session() as s:
        rule = await s.get(Rule, constraint["rule_id"])
        topic_id = rule.topic_id
    sibling = await rulebooks_svc.create_rule(
        when_to_apply="when the moment this fixture stands in for arises",
        topic_id, constraint["uid"], "A different rule", "Unrelated.",
    )

    assert await rulebooks_svc.get_rule_version(
        sibling.id, version.id, constraint["uid"],
    ) is None
    # And through its own rule it reads fine — or the assertion above would
    # pass for the wrong reason.
    assert await rulebooks_svc.get_rule_version(
        constraint["rule_id"], version.id, constraint["uid"],
    ) is not None


async def test_the_actor_does_not_grant_the_read(constraint, stranger):
    """The inverse mistake. `user_id` on a version is the ACTOR, so scoping
    the history by it would hand someone the snapshots they personally wrote
    on a rule that has since moved out of their reach — and would hide, from
    the rule's owner, every edit somebody else made."""
    async with async_session() as s:
        s.add(RuleVersion(
            rule_id=constraint["rule_id"], user_id=stranger,
            title="written by someone else", statement="a stranger's edit",
        ))
        await s.commit()

    assert await rulebooks_svc.list_rule_versions(
        constraint["rule_id"], stranger,
    ) is None

    owner_view = await rulebooks_svc.list_rule_versions(
        constraint["rule_id"], constraint["uid"],
    )
    assert "a stranger's edit" in [v.statement for v in owner_view]