apps/api/tests/test_review_queue.py

"""The review queue: what reaches the front of it, and who gets to clear a flag.

Both regressions here only appear once the queue has more than a page of work or
more than one reviewer — which is to say, only in the conditions the queue exists
for. Severity was applied after `LIMIT`, so "most severe first" stopped being true
past page one; and a resolve that lost a race still wrote itself into the
append-only log as though it had won.
"""

from __future__ import annotations

import asyncio
import uuid
from collections.abc import AsyncIterator
from datetime import UTC, datetime, timedelta

import httpx
import pytest
from sqlalchemy import func, select

from app.core.db import transaction
from app.main import create_app
from app.models.enums import (
    AnomalySeverity,
    AnomalyStatus,
    AnomalyType,
    AuditEvent,
    DocumentStatus,
)
from app.models.tables import Anomaly, AuditLog, Document

pytestmark = pytest.mark.integration


@pytest.fixture
async def api(clean_db: None) -> AsyncIterator[httpx.AsyncClient]:
    """The HTTP surface, without running the application lifespan.

    Deliberately not wrapped in `LifespanManager`. Shutdown calls
    `dispose_engine()`, which disposes the *process-wide* engine that the
    session-scoped `engine` fixture also owns — survivable for one module, but a
    second module doing it leaves the pool disposed underneath tests that are
    still using it, and the suite's zero-warning policy turns the resulting
    unclosed socket into a failure. Nothing these tests touch needs the lifespan:
    the schema comes from the fixtures, and `app.state.limiter` is wired in
    `create_app`.
    """
    app = create_app()
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://ledgerlens.test",
        timeout=30.0,
    ) as client:
        yield client


async def _document(suffix: str) -> uuid.UUID:
    document_id = uuid.uuid4()
    async with transaction() as session:
        session.add(
            Document(
                id=document_id,
                file_hash=suffix.rjust(64, "0"),
                filename=f"{suffix}.pdf",
                media_type="application/pdf",
                size_bytes=1024,
                status=str(DocumentStatus.DONE),
            )
        )
    return document_id


async def _anomaly(
    document_id: uuid.UUID, severity: AnomalySeverity, *, age_days: int, fingerprint: str
) -> uuid.UUID:
    anomaly_id = uuid.uuid4()
    async with transaction() as session:
        session.add(
            Anomaly(
                id=anomaly_id,
                document_id=document_id,
                anomaly_type=str(AnomalyType.ROUND_NUMBER),
                severity=str(severity),
                reason=f"{severity} finding",
                evidence={},
                fingerprint=fingerprint,
                status=str(AnomalyStatus.OPEN),
                created_at=datetime.now(UTC) - timedelta(days=age_days),
            )
        )
    return anomaly_id


# --- F-5: severity ordering survives pagination ----------------------------


async def test_the_most_severe_flag_leads_the_queue_even_when_it_is_the_oldest(
    api: httpx.AsyncClient,
) -> None:
    """The failure the Python sort hid: order by date, page, *then* sort.

    With the HIGH finding older than every LOW one, a page-sized LIMIT taken in
    date order contains no HIGH row at all — so re-sorting the page could never
    surface it, and the queue's own description became false at page one.
    """
    document_id = await _document("a1")
    await _anomaly(document_id, AnomalySeverity.HIGH, age_days=90, fingerprint="old-high")
    for day in range(4):
        await _anomaly(
            document_id, AnomalySeverity.LOW, age_days=day, fingerprint=f"recent-low-{day}"
        )

    first_page = (await api.get("/v1/anomalies", params={"limit": 1})).json()
    assert [item["severity"] for item in first_page] == [str(AnomalySeverity.HIGH)]


async def test_the_whole_queue_is_ordered_by_severity_then_recency(
    api: httpx.AsyncClient,
) -> None:
    document_id = await _document("a2")
    await _anomaly(document_id, AnomalySeverity.LOW, age_days=1, fingerprint="low")
    await _anomaly(document_id, AnomalySeverity.HIGH, age_days=2, fingerprint="high")
    await _anomaly(document_id, AnomalySeverity.MEDIUM, age_days=3, fingerprint="medium")

    severities = [item["severity"] for item in (await api.get("/v1/anomalies")).json()]
    assert severities == [
        str(AnomalySeverity.HIGH),
        str(AnomalySeverity.MEDIUM),
        str(AnomalySeverity.LOW),
    ]


async def test_paging_never_repeats_or_skips_a_flag(api: httpx.AsyncClient) -> None:
    """A stable total order is what makes LIMIT/OFFSET paging coherent at all."""
    document_id = await _document("a3")
    for index, severity in enumerate(
        [AnomalySeverity.LOW, AnomalySeverity.HIGH, AnomalySeverity.MEDIUM] * 2
    ):
        await _anomaly(document_id, severity, age_days=index, fingerprint=f"f{index}")

    seen: list[str] = []
    for offset in range(0, 6, 2):
        page = (await api.get("/v1/anomalies", params={"limit": 2, "offset": offset})).json()
        seen.extend(item["id"] for item in page)

    assert len(seen) == len(set(seen)) == 6


# --- F-6: a resolve that loses the race writes nothing ---------------------


async def test_only_one_of_two_simultaneous_reviewers_wins(api: httpx.AsyncClient) -> None:
    """Both read OPEN; Postgres lets one UPDATE through and the other matches zero.

    The loser must not append ANOMALY_RESOLVED naming its own decision — the
    trigger would make that permanent — and must not return 200 carrying the
    winner's outcome as though it were the caller's own.
    """
    document_id = await _document("a4")
    anomaly_id = await _anomaly(
        document_id, AnomalySeverity.HIGH, age_days=0, fingerprint="contested"
    )

    approve, reject = await asyncio.gather(
        api.post(f"/v1/anomalies/{anomaly_id}/resolve", json={"action": "approve"}),
        api.post(f"/v1/anomalies/{anomaly_id}/resolve", json={"action": "reject"}),
    )
    codes = sorted([approve.status_code, reject.status_code])
    assert codes == [200, 409], f"expected exactly one winner, got {codes}"

    loser = approve if approve.status_code == 409 else reject
    assert loser.json()["error"]["code"] == "invalid_state_transition"

    async with transaction() as session:
        resolutions = (
            await session.execute(
                select(func.count())
                .select_from(AuditLog)
                .where(
                    AuditLog.document_id == document_id,
                    AuditLog.event == str(AuditEvent.ANOMALY_RESOLVED),
                )
            )
        ).scalar_one()
    assert resolutions == 1, "the losing reviewer must not appear in the audit trail"


async def test_resolving_an_already_reviewed_flag_is_refused(api: httpx.AsyncClient) -> None:
    document_id = await _document("a5")
    anomaly_id = await _anomaly(
        document_id, AnomalySeverity.MEDIUM, age_days=0, fingerprint="settled"
    )

    assert (
        await api.post(f"/v1/anomalies/{anomaly_id}/resolve", json={"action": "approve"})
    ).status_code == 200
    second = await api.post(f"/v1/anomalies/{anomaly_id}/resolve", json={"action": "reject"})
    assert second.status_code == 409
    assert second.json()["error"]["code"] == "invalid_state_transition"