docqa/tests/test_api.py

"""HTTP-layer tests: validation, error mapping, and response shape.

Uses FastAPI's TestClient with the real pipeline where inference is needed, so
these exercise the actual serving path rather than a stub.
"""

from __future__ import annotations

import base64

import pytest
from fastapi.testclient import TestClient

from docxextract.api import app


@pytest.fixture(scope="module")
def client():
    with TestClient(app) as c:
        yield c


def test_health(client):
    response = client.get("/health")
    assert response.status_code == 200
    body = response.json()
    assert body["status"] in ("ok", "degraded")
    assert "model_ready" in body


def test_config(client):
    body = client.get("/v1/config").json()
    assert body["max_keys"] > 0
    assert 0.0 <= body["confidence_threshold"] <= 1.0


def test_extract_rejects_unsupported_media(client):
    response = client.post(
        "/v1/extract",
        files={"file": ("notes.txt", b"just text", "text/plain")},
        data={"keys": "Total"},
    )
    assert response.status_code in (415, 422)


def test_extract_rejects_empty_keys(client, sample_pdf):
    response = client.post(
        "/v1/extract",
        files={"file": ("invoice.pdf", sample_pdf, "application/pdf")},
        data={"keys": " , , "},
    )
    assert response.status_code == 422


def test_extract_rejects_too_many_keys(client, sample_pdf):
    response = client.post(
        "/v1/extract",
        files={"file": ("invoice.pdf", sample_pdf, "application/pdf")},
        data={"keys": ",".join(f"key{i}" for i in range(500))},
    )
    assert response.status_code == 422


def test_extract_rejects_corrupt_pdf(client):
    response = client.post(
        "/v1/extract",
        files={"file": ("broken.pdf", b"%PDF-1.7\ngarbage garbage",
                        "application/pdf")},
        data={"keys": "Total"},
    )
    assert response.status_code in (422, 500)


def test_extract_success_shape(client, sample_pdf):
    keys = ["INVOICE NO", "Total Amount"]
    response = client.post(
        "/v1/extract",
        files={"file": ("invoice.pdf", sample_pdf, "application/pdf")},
        data={"keys": ",".join(keys)},
    )
    assert response.status_code == 200
    body = response.json()
    assert [f["key"] for f in body["fields"]] == keys
    for field in body["fields"]:
        assert field["status"] in ("extracted", "low_confidence", "not_found",
                                   "error")
        assert 0.0 <= field["confidence"] <= 1.0
    assert body["document"]["word_count"] > 0
    assert body["latency_ms"] >= 0
    assert response.headers.get("X-Request-ID")


def test_duplicate_keys_collapse(client, sample_pdf):
    response = client.post(
        "/v1/extract",
        files={"file": ("invoice.pdf", sample_pdf, "application/pdf")},
        data={"keys": "Total Amount,Total Amount"},
    )
    body = response.json()
    assert len(body["fields"]) == 1


def test_json_endpoint(client, sample_pdf):
    response = client.post("/v1/extract/json", json={
        "keys": ["INVOICE NO"],
        "document_base64": base64.b64encode(sample_pdf).decode(),
    })
    assert response.status_code == 200
    assert response.json()["fields"][0]["key"] == "INVOICE NO"


def test_json_endpoint_rejects_bad_base64(client):
    response = client.post("/v1/extract/json", json={
        "keys": ["Total"], "document_base64": "!!!not-base64!!!"})
    assert response.status_code == 422


def test_json_endpoint_requires_keys(client, sample_pdf):
    response = client.post("/v1/extract/json", json={
        "keys": [], "document_base64": base64.b64encode(sample_pdf).decode()})
    assert response.status_code == 422


def test_confidence_threshold_is_respected(client, sample_pdf):
    """A threshold of 1.0 must suppress every extraction."""
    response = client.post(
        "/v1/extract",
        files={"file": ("invoice.pdf", sample_pdf, "application/pdf")},
        data={"keys": "INVOICE NO,Total Amount", "confidence_threshold": "1.0"},
    )
    body = response.json()
    assert body["confidence_threshold"] == 1.0
    assert all(f["status"] != "extracted" for f in body["fields"])