538 lines
18 KiB
Python
538 lines
18 KiB
Python
import contextlib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.services import engine
|
|
from app.services.store import document_store
|
|
|
|
has_pdfium = False
|
|
if engine.is_available():
|
|
with contextlib.suppress(Exception):
|
|
has_pdfium = engine.require().engine_has_pdfium()
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not engine.is_available() or not has_pdfium,
|
|
reason="pdfengine pybind11 module is not compiled/available, or was compiled without PDFium support.",
|
|
)
|
|
|
|
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):
|
|
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_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_resp = client.get(f"/documents/{doc_id}")
|
|
assert get_resp.status_code == 200
|
|
assert get_resp.json()["filename"] == HELLO_WORLD_PDF.name
|
|
|
|
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"]
|
|
|
|
del_resp = client.delete(f"/documents/{doc_id}")
|
|
assert del_resp.status_code == 200
|
|
assert del_resp.json() == {"success": True}
|
|
|
|
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"]
|
|
|
|
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
|
|
|
|
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
|
|
|
|
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()
|
|
assert "glyphs" in payload
|
|
glyphs = payload["glyphs"]
|
|
assert len(glyphs) > 0
|
|
first_glyph = glyphs[0]
|
|
for key in ["text", "x", "y", "w", "h", "fontSize"]:
|
|
assert key in first_glyph
|
|
|
|
for g in glyphs:
|
|
assert g["fontSize"] != 1.0, f"Fake fontSize 1.0 detected for glyph: {g}"
|
|
assert g["text"] not in ["\r", "\n"], f"Control character detected in glyph bounds: {g}"
|
|
|
|
|
|
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 = {
|
|
"version": "1.0",
|
|
"operations": [
|
|
{
|
|
"id": "op_route_test_123",
|
|
"type": "text_overlay",
|
|
"pageIndex": 0,
|
|
"data": {
|
|
"text": "Edited Text Annotation",
|
|
"x": 100.0,
|
|
"y": 150.0,
|
|
"width": 200.0,
|
|
"height": 20.0,
|
|
"fontSize": 14.0,
|
|
"fontFamily": "Helvetica",
|
|
"color": "#000000",
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
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"
|
|
|
|
text_resp = client.get(f"/documents/{new_doc_id}/pages/0/text")
|
|
assert text_resp.status_code == 200
|
|
assert "Edited Text Annotation" in text_resp.json()["text"]
|
|
|
|
|
|
def test_apply_image_overlay_and_incremental_save(client: TestClient):
|
|
import base64
|
|
import io
|
|
|
|
from PIL import Image
|
|
|
|
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"]
|
|
|
|
img = Image.new("RGBA", (2, 2), color="red")
|
|
buf = io.BytesIO()
|
|
img.save(buf, format="PNG")
|
|
png_bytes = buf.getvalue()
|
|
png_base64 = base64.b64encode(png_bytes).decode("utf-8")
|
|
|
|
edits_payload = {
|
|
"version": "1.0",
|
|
"operations": [
|
|
{
|
|
"id": "op_route_img_test_123",
|
|
"type": "image_overlay",
|
|
"pageIndex": 0,
|
|
"data": {
|
|
"x": 100.0,
|
|
"y": 150.0,
|
|
"width": 200.0,
|
|
"height": 150.0,
|
|
"imageData": f"data:image/png;base64,{png_base64}",
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
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"
|
|
|
|
|
|
def test_apply_page_rotation_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 = {
|
|
"version": "1.0",
|
|
"operations": [
|
|
{
|
|
"id": "op_route_rot_test_123",
|
|
"type": "page_rotation",
|
|
"pageIndex": 0,
|
|
"data": {"rotation": 90},
|
|
}
|
|
],
|
|
}
|
|
|
|
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"
|
|
|
|
|
|
def test_apply_redaction_and_full_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 = {
|
|
"version": "1.0",
|
|
"operations": [
|
|
{
|
|
"id": "op_route_redact_test_123",
|
|
"type": "redaction",
|
|
"pageIndex": 0,
|
|
"data": {
|
|
"x": 0.0,
|
|
"y": 0.0,
|
|
"width": 612.0,
|
|
"height": 792.0,
|
|
"fillColor": "#ffffff",
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
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
|
|
|
|
text_resp = client.get(f"/documents/{new_doc_id}/pages/0/text")
|
|
assert text_resp.status_code == 200
|
|
assert "hello" not in text_resp.json()["text"].lower()
|
|
assert "world" not in text_resp.json()["text"].lower()
|
|
|
|
|
|
def test_apply_page_deletion_and_incremental_save(client: TestClient):
|
|
two_pages_pdf = CORPUS_DIR / "basic" / "hello_world_2_pages.pdf"
|
|
assert two_pages_pdf.exists(), f"hello_world_2_pages.pdf not found at {two_pages_pdf}"
|
|
|
|
with open(two_pages_pdf, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents", files={"file": (two_pages_pdf.name, f, "application/pdf")}
|
|
)
|
|
doc_id = upload_resp.json()["id"]
|
|
assert upload_resp.json()["totalPages"] == 2
|
|
|
|
edits_payload = {
|
|
"version": "1.0",
|
|
"operations": [
|
|
{
|
|
"id": "op_route_del_test_123",
|
|
"type": "page_deletion",
|
|
"pageIndex": 1,
|
|
"data": {},
|
|
}
|
|
],
|
|
}
|
|
|
|
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"]
|
|
|
|
get_resp = client.get(f"/documents/{new_doc_id}")
|
|
assert get_resp.status_code == 200
|
|
assert get_resp.json()["totalPages"] == 1
|
|
|
|
|
|
def test_apply_page_reorder_and_incremental_save(client: TestClient):
|
|
two_pages_pdf = CORPUS_DIR / "basic" / "hello_world_2_pages.pdf"
|
|
assert two_pages_pdf.exists(), f"hello_world_2_pages.pdf not found at {two_pages_pdf}"
|
|
|
|
with open(two_pages_pdf, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents", files={"file": (two_pages_pdf.name, f, "application/pdf")}
|
|
)
|
|
doc_id = upload_resp.json()["id"]
|
|
|
|
edits_payload = {
|
|
"version": "1.0",
|
|
"operations": [
|
|
{
|
|
"id": "op_route_reorder_test_123",
|
|
"type": "page_reorder",
|
|
"pageIndex": 1,
|
|
"data": {"destPageIndex": 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"]
|
|
|
|
get_resp = client.get(f"/documents/{new_doc_id}")
|
|
assert get_resp.status_code == 200
|
|
assert get_resp.json()["totalPages"] == 2
|
|
|
|
|
|
def test_get_document_and_page_fonts(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"]
|
|
|
|
fonts_resp = client.get(f"/documents/{doc_id}/fonts")
|
|
assert fonts_resp.status_code == 200
|
|
fonts = fonts_resp.json()
|
|
assert isinstance(fonts, list)
|
|
|
|
page_fonts_resp = client.get(f"/documents/{doc_id}/pages/0/fonts")
|
|
assert page_fonts_resp.status_code == 200
|
|
page_fonts = page_fonts_resp.json()
|
|
assert isinstance(page_fonts, list)
|
|
assert len(fonts) == len(page_fonts)
|
|
|
|
if len(fonts) > 0:
|
|
f = fonts[0]
|
|
fields = [
|
|
"fontName",
|
|
"type",
|
|
"isEmbedded",
|
|
"isSubset",
|
|
"isVertical",
|
|
"encoding",
|
|
"hasToUnicode",
|
|
"cmapName",
|
|
"cidSystemInfo",
|
|
"subsetTag",
|
|
"sourceType",
|
|
"substitutedFrom",
|
|
"substitutedTo",
|
|
"normalizedFamily",
|
|
"internalFontId",
|
|
"flags",
|
|
"ascent",
|
|
"descent",
|
|
"capHeight",
|
|
]
|
|
for field in fields:
|
|
assert field in f
|
|
|
|
|
|
def test_font_size_and_diagnostics_advanced(client: TestClient):
|
|
utf8_pdf = CORPUS_DIR / "fonts" / "utf-8.pdf"
|
|
vertical_pdf = CORPUS_DIR / "fonts" / "vertical_text.pdf"
|
|
|
|
assert utf8_pdf.exists(), f"utf-8.pdf not found at {utf8_pdf}"
|
|
|
|
with open(utf8_pdf, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents", files={"file": (utf8_pdf.name, f, "application/pdf")}
|
|
)
|
|
assert upload_resp.status_code == 201
|
|
doc_id = upload_resp.json()["id"]
|
|
|
|
fonts_resp = client.get(f"/documents/{doc_id}/fonts")
|
|
assert fonts_resp.status_code == 200
|
|
fonts = fonts_resp.json()
|
|
assert isinstance(fonts, list)
|
|
|
|
for font in fonts:
|
|
assert isinstance(font["fontName"], str)
|
|
assert len(font["fontName"]) > 0
|
|
assert isinstance(font["type"], str)
|
|
assert font["type"] in ["TrueType", "Type1", "CIDFontType0", "CIDFontType2"]
|
|
assert isinstance(font["isEmbedded"], bool)
|
|
assert isinstance(font["isSubset"], bool)
|
|
assert isinstance(font["isVertical"], bool)
|
|
assert isinstance(font["encoding"], str)
|
|
assert isinstance(font["hasToUnicode"], bool)
|
|
assert isinstance(font["cmapName"], str)
|
|
assert isinstance(font["cidSystemInfo"], str)
|
|
assert isinstance(font["subsetTag"], str)
|
|
assert isinstance(font["sourceType"], str)
|
|
assert font["sourceType"] in ["Embedded", "SystemFallback", "Substituted"]
|
|
assert isinstance(font["substitutedFrom"], str)
|
|
assert isinstance(font["substitutedTo"], str)
|
|
assert isinstance(font["normalizedFamily"], str)
|
|
assert isinstance(font["internalFontId"], str)
|
|
assert isinstance(font["flags"], int)
|
|
assert isinstance(font["ascent"], int | float)
|
|
assert isinstance(font["descent"], int | float)
|
|
assert isinstance(font["capHeight"], int | float)
|
|
|
|
if font["isSubset"]:
|
|
assert font["isEmbedded"]
|
|
assert font["sourceType"] == "Embedded"
|
|
assert len(font["subsetTag"]) == 6
|
|
assert font["subsetTag"].isupper()
|
|
assert font["internalFontId"] == font["fontName"]
|
|
else:
|
|
assert len(font["subsetTag"]) == 0
|
|
assert font["internalFontId"] == f"{font['fontName']}_{font['type']}_{font['flags']}"
|
|
|
|
assert "+" not in font["normalizedFamily"]
|
|
assert "," not in font["normalizedFamily"]
|
|
assert "bold" not in font["normalizedFamily"].lower()
|
|
assert "italic" not in font["normalizedFamily"].lower()
|
|
|
|
assert font["ascent"] > 0
|
|
assert font["descent"] < 0
|
|
assert font["capHeight"] > 0
|
|
|
|
text_resp = client.get(f"/documents/{doc_id}/pages/0/text")
|
|
assert text_resp.status_code == 200
|
|
page_data = text_resp.json()
|
|
assert "text" in page_data
|
|
assert "glyphs" in page_data
|
|
|
|
glyphs = page_data["glyphs"]
|
|
assert len(glyphs) > 0
|
|
|
|
for glyph in glyphs:
|
|
assert isinstance(glyph["text"], str)
|
|
assert len(glyph["text"]) > 0
|
|
assert isinstance(glyph["x"], int | float)
|
|
assert isinstance(glyph["y"], int | float)
|
|
assert isinstance(glyph["w"], int | float)
|
|
assert isinstance(glyph["h"], int | float)
|
|
assert isinstance(glyph["fontSize"], int | float)
|
|
|
|
assert glyph["fontSize"] > 0
|
|
assert glyph["fontSize"] < 100
|
|
|
|
if glyph["text"].strip():
|
|
assert glyph["w"] > 0
|
|
assert glyph["h"] > 0
|
|
|
|
if vertical_pdf.exists():
|
|
with open(vertical_pdf, "rb") as f:
|
|
upload_resp = client.post(
|
|
"/documents", files={"file": (vertical_pdf.name, f, "application/pdf")}
|
|
)
|
|
assert upload_resp.status_code == 201
|
|
vert_doc_id = upload_resp.json()["id"]
|
|
|
|
vert_fonts_resp = client.get(f"/documents/{vert_doc_id}/fonts")
|
|
assert vert_fonts_resp.status_code == 200
|
|
vert_fonts = vert_fonts_resp.json()
|
|
|
|
has_vertical = False
|
|
for font in vert_fonts:
|
|
if font["isVertical"]:
|
|
has_vertical = True
|
|
assert (
|
|
"-V" in font["encoding"]
|
|
or "-V" in font["cmapName"]
|
|
or "Identity-V" in font["encoding"]
|
|
)
|
|
|
|
assert has_vertical, "Expected to find a vertical font in vertical_text.pdf"
|
|
|
|
|
|
def test_export_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")}
|
|
)
|
|
assert upload_resp.status_code == 201
|
|
doc_id = upload_resp.json()["id"]
|
|
|
|
export_resp = client.get(f"/documents/{doc_id}/export")
|
|
assert export_resp.status_code == 200
|
|
assert export_resp.headers["content-type"] == "application/pdf"
|
|
assert "attachment" in export_resp.headers["content-disposition"]
|
|
assert HELLO_WORLD_PDF.name in export_resp.headers["content-disposition"]
|
|
|
|
content = export_resp.content
|
|
assert len(content) > 0
|
|
assert content.startswith(b"%PDF")
|