FabledGrimoire/tests/test_integration_record_batch.py

"""Real-Postgres proof that a batch create cannot be interfered with.

What mocks cannot show: that the ids a placeholder is rewritten to are the ids
the rows actually got, that two batches running at the same time each cite
their OWN records, and that a failing batch leaves nothing behind. Those are
properties of the sequence and the transaction, so they are measured against
the real ones.
"""
import asyncio

import pytest
import pytest_asyncio
from sqlalchemy import func, select

from grimoire.models import async_session
from grimoire.models.milestone import Milestone
from grimoire.models.note import Note
from grimoire.models.project import Project
from grimoire.services import record_refs
from grimoire.services.planning import start_planning
from grimoire.services.record_batch import BatchItem, create_batch
from tests.helpers import ensure_user

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


@pytest_asyncio.fixture
async def seeded():
    async with async_session() as s:
        owner = await ensure_user(s, "record_batch_owner")
        project = Project(user_id=owner.id, title="Batch target")
        s.add(project)
        await s.flush()
        ids = {"owner": owner.id, "pid": project.id}
        await s.commit()
    return ids


async def _count_in_project(pid: int) -> tuple[int, int]:
    async with async_session() as s:
        notes = (await s.execute(select(func.count(Note.id)).where(Note.project_id == pid))).scalar()
        milestones = (await s.execute(
            select(func.count(Milestone.id)).where(Milestone.project_id == pid)
        )).scalar()
    return notes, milestones


async def test_a_plan_and_its_steps_cite_each_others_real_ids(seeded):
    owner, pid = seeded["owner"], seeded["pid"]
    out = await start_planning(
        owner, pid, "Ship it",
        body="First {{ref:1}}, then {{ref:2}}.",
        steps=[
            BatchItem(title="Step one", body="Part of {{ref:milestone}}; next is {{ref:2}}."),
            BatchItem(title="Step two", body="Follows {{ref:1}}."),
        ],
    )
    ms = out["milestone"]
    one, two = out["steps"]
    assert ms["body"] == f'First #{one["id"]} "Step one", then #{two["id"]} "Step two".'
    assert one["body"] == f'Part of milestone {ms["id"]} "Ship it"; next is #{two["id"]} "Step two".'
    assert two["body"] == f'Follows #{one["id"]} "Step one".'
    assert one["milestone_id"] == two["milestone_id"] == ms["id"]
    assert one["project_id"] == pid

    # What was returned is what was stored — not a rewrite applied to the
    # response alone.
    async with async_session() as s:
        stored = await s.get(Note, two["id"])
        assert stored.body == f'Follows #{one["id"]} "Step one".'


async def test_concurrent_batches_each_cite_their_own_records(seeded):
    """The operator's scenario: several sessions creating at once. Each batch
    must resolve {{ref:1}} to ITS first record, whatever the others take."""
    owner, pid = seeded["owner"], seeded["pid"]

    async def one_batch(tag: str):
        return await create_batch(owner, [
            BatchItem(title=f"{tag} first"),
            BatchItem(title=f"{tag} second", body="points at {{ref:1}}"),
        ], project_id=pid)

    results = await asyncio.gather(*(one_batch(f"session-{i}") for i in range(6)))

    all_ids = [n.id for _ms, notes in results for n in notes]
    assert len(all_ids) == len(set(all_ids)), "the sequence handed out an id twice"
    for _ms, (first, second) in results:
        assert second.body == f'points at #{first.id} "{first.title}"'
        async with async_session() as s:
            assert (await s.get(Note, first.id)).title == first.title


async def test_a_failing_batch_writes_nothing(seeded):
    owner, pid = seeded["owner"], seeded["pid"]
    before = await _count_in_project(pid)
    with pytest.raises(ValueError, match="Invalid status"):
        await start_planning(owner, pid, "Doomed", steps=[
            BatchItem(title="fine"),
            BatchItem(title="broken", status="someday"),
        ])
    assert await _count_in_project(pid) == before, "a partial batch (or its milestone) was left behind"


async def test_a_guess_above_the_real_highest_id_is_refused(seeded):
    owner, pid = seeded["owner"], seeded["pid"]
    _ms, (made,) = await create_batch(owner, [BatchItem(title="anchor")], project_id=pid)
    # `made` is the newest row, so anything just past it can only be a guess —
    # unless a parallel test in the lane created more since, which only moves
    # the ceiling up and keeps made.id itself a real id.
    with pytest.raises(ValueError, match="does not exist yet"):
        await record_refs.refuse_guessed_ids(f"#{made.id + record_refs.GUESS_WINDOW}")
    await record_refs.refuse_guessed_ids(f"#{made.id}")