GEWU/tests/adapters/test_pdf_extractor.py

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()