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.

    Cleanup is in this fixture's own teardown, never in an autouse one:
    `_dispose_engine` arrives through usefixtures, so it tears down BEFORE an
    autouse fixture would, and a database call after that point orphans a
    pooled connection and breaks the NEXT test to touch Postgres (#3240).
    """
    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(
        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",
    )

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

    async with async_session() as s:
        row = await s.get(Rulebook, book.id)
        if row is not None:
            await s.delete(row)
        await s.commit()


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.",
    ]