app/tests/test_canonical_rollout_api.py

"""HTTP tests for canonical paragraph rollout (scan + apply guards)."""

import pytest

from app.db.models import Document, IngestStatus
from app.models.schemas import SearchResult


class _FakeVS:
    def __init__(self, rows: list[SearchResult]) -> None:
        self._rows = rows

    def search(self, query: str, tenant_id: str, k: int = 10, **kwargs: object) -> list[SearchResult]:
        return self._rows

    def delete_document(self, doc_id: str) -> None:
        return None

    def count_for_doc(self, doc_id: str) -> int:
        return 0


@pytest.mark.asyncio
async def test_canonical_scan_returns_matches(
    async_client: object,
    test_db: object,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    doc_id = "cccccccc-cccc-cccc-cccc-cccccccccccc"
    tenant = "tenant_test"
    doc = Document(
        id=doc_id,
        tenant_id=tenant,
        filename="guidance.docx",
        file_path="/tmp/guidance.docx",
        status=IngestStatus.complete,
    )
    test_db.add(doc)
    await test_db.commit()

    hits = [
        SearchResult(
            chunk_id=f"{doc_id}_00003",
            doc_id=doc_id,
            tenant_id=tenant,
            text=(
                "The damp proof course should be at least 150 mm above external ground level "
                "in accordance with approved guidance for moisture protection."
            ),
            score=0.41,
            section_type="paragraph",
        ),
    ]
    monkeypatch.setattr(
        "app.services.canonical_rollout.get_vectorstore",
        lambda: _FakeVS(hits),
    )

    res = await async_client.post(
        "/content/canonical-scan",
        json={
            "canonical_text": (
                "Damp proof courses must provide adequate clearance above external ground "
                "to limit moisture ingress per current building standards."
            ),
            "limit": 10,
            "min_relevance_percent": 20.0,
            "min_jaccard_vs_canonical": 0.05,
        },
        headers={"X-Tenant-ID": tenant},
    )
    assert res.status_code == 200
    data = res.json()
    assert len(data["matches"]) == 1
    assert data["matches"][0]["document_id"] == doc_id
    assert data["matches"][0]["chunk_id"] == f"{doc_id}_00003"
    assert data["matches"][0]["compatibility"] in ("high", "review", "low")
    assert "damp" in data["matches"][0]["proposed_replacement"].lower()


@pytest.mark.asyncio
async def test_canonical_apply_requires_confirm(
    async_client: object,
) -> None:
    res = await async_client.post(
        "/content/canonical-apply",
        json={
            "document_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
            "chunk_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa_00000",
            "replacement_text": "Updated text.",
            "confirm_destructive_docx": False,
        },
        headers={"X-Tenant-ID": "tenant_test"},
    )
    assert res.status_code == 400


@pytest.mark.asyncio
async def test_canonical_apply_rejects_pdf(
    async_client: object,
    test_db: object,
) -> None:
    doc_id = "dddddddd-dddd-dddd-dddd-dddddddddddd"
    tenant = "tenant_test"
    doc = Document(
        id=doc_id,
        tenant_id=tenant,
        filename="ref.pdf",
        file_path="/tmp/ref.pdf",
        status=IngestStatus.complete,
    )
    test_db.add(doc)
    await test_db.commit()

    res = await async_client.post(
        "/content/canonical-apply",
        json={
            "document_id": doc_id,
            "chunk_id": f"{doc_id}_00000",
            "replacement_text": "x",
            "confirm_destructive_docx": True,
        },
        headers={"X-Tenant-ID": tenant},
    )
    assert res.status_code == 422