114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gewu.adapters.pdf_extractor import PdfTextExtractor
|
|
|
|
|
|
def _pdf(pages: tuple[str, ...]) -> bytes:
|
|
objects: list[bytes] = []
|
|
page_ids = [4 + 2 * index for index in range(len(pages))]
|
|
objects.extend(
|
|
[
|
|
b"<< /Type /Catalog /Pages 2 0 R >>",
|
|
(
|
|
f"<< /Type /Pages /Kids [{' '.join(f'{page_id} 0 R' for page_id in page_ids)}] "
|
|
f"/Count {len(pages)} >>"
|
|
).encode(),
|
|
b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
|
|
]
|
|
)
|
|
for page_id, text in zip(page_ids, pages, strict=True):
|
|
content_id = page_id + 1
|
|
escaped = text.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
|
|
stream = f"BT /F1 12 Tf 72 720 Td ({escaped}) Tj ET".encode()
|
|
page = (
|
|
f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] "
|
|
f"/Resources << /Font << /F1 3 0 R >> >> /Contents {content_id} 0 R >>"
|
|
)
|
|
objects.append(page.encode())
|
|
objects.append(
|
|
b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + stream + b"\nendstream"
|
|
)
|
|
output = bytearray(b"%PDF-1.4\n")
|
|
offsets = [0]
|
|
for number, obj in enumerate(objects, 1):
|
|
offsets.append(len(output))
|
|
output.extend(f"{number} 0 obj\n".encode() + obj + b"\nendobj\n")
|
|
xref = len(output)
|
|
output.extend(f"xref\n0 {len(offsets)}\n0000000000 65535 f \n".encode())
|
|
for offset in offsets[1:]:
|
|
output.extend(f"{offset:010d} 00000 n \n".encode())
|
|
output.extend(
|
|
f"trailer\n<< /Size {len(offsets)} /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n".encode()
|
|
)
|
|
return bytes(output)
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not all(shutil.which(item) for item in ("bwrap", "pdftotext", "prlimit")),
|
|
reason="Linux PDF sandbox tools unavailable",
|
|
)
|
|
def test_pdf_extraction_preserves_page_locations_and_parser_identity() -> None:
|
|
record = PdfTextExtractor().extract(
|
|
"project:source:r1",
|
|
_pdf(("First page evidence.", "Second page evidence.")),
|
|
"application/pdf",
|
|
)
|
|
assert record.status == "complete"
|
|
assert record.parser_name == "gewu-poppler-pdftotext"
|
|
assert [block.page for block in record.blocks] == [1, 2]
|
|
assert [block.original_text.strip() for block in record.blocks] == [
|
|
"First page evidence.",
|
|
"Second page evidence.",
|
|
]
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not all(shutil.which(item) for item in ("bwrap", "pdftotext", "prlimit")),
|
|
reason="Linux PDF sandbox tools unavailable",
|
|
)
|
|
def test_empty_pdf_page_is_reported_as_partial() -> None:
|
|
record = PdfTextExtractor().extract(
|
|
"project:source:r1", _pdf(("Text page.", "")), "application/pdf"
|
|
)
|
|
assert record.status == "partial"
|
|
assert record.blocks[0].page == 1
|
|
assert record.warnings
|
|
|
|
|
|
def test_pdf_parser_rejects_bad_signature_without_launching_parser() -> None:
|
|
with pytest.raises(ValueError, match="signature"):
|
|
PdfTextExtractor().extract("project:source:r1", b"not a pdf", "application/pdf")
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not shutil.which("bwrap") or not shutil.which("prlimit"), reason="bubblewrap unavailable"
|
|
)
|
|
def test_sandbox_cannot_read_host_temporary_file(tmp_path: Path) -> None:
|
|
sentinel = tmp_path / "gewu-host-secret-fixture"
|
|
sentinel.write_text("must-not-be-visible", encoding="utf-8")
|
|
extractor = PdfTextExtractor()
|
|
if not extractor._pdftotext:
|
|
pytest.skip("pdftotext unavailable")
|
|
command = extractor._command()
|
|
command[-6:] = ["/usr/bin/cat", str(sentinel)]
|
|
with tempfile.TemporaryFile() as output, tempfile.TemporaryFile() as errors:
|
|
result = subprocess.run(
|
|
command,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=output,
|
|
stderr=errors,
|
|
env={},
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
assert result.returncode != 0
|
|
output.seek(0)
|
|
assert b"must-not-be-visible" not in output.read()
|