159 lines
5.1 KiB
Python
159 lines
5.1 KiB
Python
import os
|
|
from pathlib import Path
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.services import engine
|
|
from app.services.store import document_store
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not engine.is_available(),
|
|
reason="pdfengine pybind11 module is not compiled/available."
|
|
)
|
|
|
|
CORPUS_DIR = Path(__file__).parent.parent.parent / "corpus"
|
|
HELLO_WORLD_PDF = CORPUS_DIR / "basic" / "hello_world.pdf"
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_store():
|
|
with document_store._lock:
|
|
document_store._documents.clear()
|
|
yield
|
|
|
|
def test_upload_document_success(client: TestClient):
|
|
assert HELLO_WORLD_PDF.exists(), f"Test corpus file not found at {HELLO_WORLD_PDF}"
|
|
|
|
with open(HELLO_WORLD_PDF, "rb") as f:
|
|
response = client.post(
|
|
"/documents",
|
|
files={"file": (HELLO_WORLD_PDF.name, f, "application/pdf")}
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
payload = response.json()
|
|
assert "id" in payload
|
|
assert payload["filename"] == HELLO_WORLD_PDF.name
|
|
assert payload["sizeBytes"] == HELLO_WORLD_PDF.stat().st_size
|
|
assert payload["totalPages"] == 1
|
|
assert payload["status"] == "ready"
|
|
|
|
def test_upload_document_invalid(client: TestClient):
|
|
response = client.post(
|
|
"/documents",
|
|
files={"file": ("test.pdf", b"not-a-pdf-file-content", "application/pdf")}
|
|
)
|
|
assert response.status_code == 400
|
|
assert "invalid pdf" in response.json()["detail"].lower()
|
|
|
|
def test_list_and_get_document(client: TestClient):
|
|
# Upload one
|
|
with open(HELLO_WORLD_PDF, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents",
|
|
files={"file": (HELLO_WORLD_PDF.name, f, "application/pdf")}
|
|
)
|
|
doc_id = upload_resp.json()["id"]
|
|
|
|
# List
|
|
list_resp = client.get("/documents")
|
|
assert list_resp.status_code == 200
|
|
docs = list_resp.json()
|
|
assert len(docs) == 1
|
|
assert docs[0]["id"] == doc_id
|
|
|
|
# Get details
|
|
get_resp = client.get(f"/documents/{doc_id}")
|
|
assert get_resp.status_code == 200
|
|
assert get_resp.json()["filename"] == HELLO_WORLD_PDF.name
|
|
|
|
# Get non-existent
|
|
fake_resp = client.get("/documents/non-existent-uuid")
|
|
assert fake_resp.status_code == 404
|
|
|
|
def test_delete_document(client: TestClient):
|
|
with open(HELLO_WORLD_PDF, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents",
|
|
files={"file": (HELLO_WORLD_PDF.name, f, "application/pdf")}
|
|
)
|
|
doc_id = upload_resp.json()["id"]
|
|
|
|
# Delete
|
|
del_resp = client.delete(f"/documents/{doc_id}")
|
|
assert del_resp.status_code == 200
|
|
assert del_resp.json() == {"success": True}
|
|
|
|
# Verify deleted
|
|
get_resp = client.get(f"/documents/{doc_id}")
|
|
assert get_resp.status_code == 404
|
|
|
|
def test_render_page_standard_and_compat(client: TestClient):
|
|
with open(HELLO_WORLD_PDF, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents",
|
|
files={"file": (HELLO_WORLD_PDF.name, f, "application/pdf")}
|
|
)
|
|
doc_id = upload_resp.json()["id"]
|
|
|
|
# Test standard test suite route
|
|
render_resp = client.get(f"/documents/{doc_id}/pages/0/render?dpi=150")
|
|
assert render_resp.status_code == 200
|
|
assert render_resp.headers["content-type"] == "image/png"
|
|
assert len(render_resp.content) > 0
|
|
|
|
# Test compat route (used by frontend)
|
|
compat_resp = client.get(f"/render/{doc_id}?page=0&zoom=1.5")
|
|
assert compat_resp.status_code == 200
|
|
assert compat_resp.headers["content-type"] == "image/png"
|
|
assert len(compat_resp.content) > 0
|
|
|
|
# Test page out of bounds
|
|
fail_resp = client.get(f"/documents/{doc_id}/pages/5/render")
|
|
assert fail_resp.status_code == 404
|
|
|
|
def test_extract_page_text(client: TestClient):
|
|
with open(HELLO_WORLD_PDF, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents",
|
|
files={"file": (HELLO_WORLD_PDF.name, f, "application/pdf")}
|
|
)
|
|
doc_id = upload_resp.json()["id"]
|
|
|
|
text_resp = client.get(f"/documents/{doc_id}/pages/0/text")
|
|
assert text_resp.status_code == 200
|
|
payload = text_resp.json()
|
|
assert "text" in payload
|
|
assert "hello" in payload["text"].lower()
|
|
|
|
def test_apply_edits_and_incremental_save(client: TestClient):
|
|
with open(HELLO_WORLD_PDF, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents",
|
|
files={"file": (HELLO_WORLD_PDF.name, f, "application/pdf")}
|
|
)
|
|
doc_id = upload_resp.json()["id"]
|
|
|
|
edits_payload = {
|
|
"operations": [
|
|
{
|
|
"type": "add_text",
|
|
"pageIndex": 0,
|
|
"data": {
|
|
"text": "Edited Text Annotation",
|
|
"x": 100.0,
|
|
"y": 150.0
|
|
}
|
|
}
|
|
]
|
|
}
|
|
|
|
edits_resp = client.post(f"/documents/{doc_id}/edits", json=edits_payload)
|
|
assert edits_resp.status_code == 200
|
|
payload = edits_resp.json()
|
|
assert payload["success"] is True
|
|
new_doc_id = payload["newDocumentId"]
|
|
assert new_doc_id != doc_id
|
|
|
|
render_resp = client.get(f"/documents/{new_doc_id}/pages/0/render")
|
|
assert render_resp.status_code == 200
|
|
assert render_resp.headers["content-type"] == "image/png" |