"""Unit tests: PDF parser module (phase 126 task 02, locked A1/A2). Covers ``app/rag/pdf.py``: text extraction with ``---PAGE-N---`` separators (A2), embedded image extraction (original encoded bytes), MIME mapping (extension first, then image type code, else the ``application/octet-stream`` fallback), and the fail-soft contract — unreadable files and broken pages/images return empty results, never raise. Temp PDFs are built with PyMuPDF itself (no binary fixtures). """ from __future__ import annotations import asyncio import hashlib import logging import tarfile from collections.abc import Callable, Sequence from datetime import timedelta from pathlib import Path from types import SimpleNamespace from typing import Any import pymupdf as fitz import pytest from sqlalchemy import select from app.config import Settings from app.models import Chunk, Document from app.rag import importer from app.rag import pdf as pdf_mod from app.rag.importer import import_sources from app.rag.llm import EmbeddingError, LLMError from app.rag.pdf import ( _XMLISH_LINE_RE, NONPRINT_CHAR_FRACTION, PAGE_SEPARATOR, XMLISH_LINE_FRACTION, PdfImage, _mime_from_image_info, _page_image_entries, extract_pdf_images, extract_pdf_text, is_usable_pdf_text, render_pdf_pages, ) from tests.fakes import FakeEmbedder PNG_MAGIC = b"\x89PNG" JPEG_MAGIC = b"\xff\xd8\xff" # --------------------------------------------------------------------------- # Phase 126, task 04 — PDF indexing in the importer # --------------------------------------------------------------------------- DESCRIPTION = ( "A network diagram of the homelab VLANs: the core switch, the " "router, and three labeled subnets." ) class _VisionLLM(FakeEmbedder): """The phase-122 mock VISION client (task 03's pattern): the CHAT model answers the multimodal describe call with the fixed :data:`DESCRIPTION`; text (``lite``) calls keep the ``FakeEmbedder`` behaviour. Every chat call's model is recorded (``chat_models``).""" def __init__(self) -> None: super().__init__() self.chat_models: list[str | None] = [] async def chat(self, messages, model=None): self.chat_calls.append(list(messages)) self.chat_models.append(model) user = next((m["content"] for m in messages if m.get("role") == "user"), "") if isinstance(user, list): return DESCRIPTION first = user.split() return "Summary of " + (first[0] if first else "") class _VisionFailsLLM(FakeEmbedder): """A chat model that REJECTS the multimodal describe call (the non-vision model of the LOCKED A3 note). Text calls keep the ``FakeEmbedder`` behaviour.""" def __init__(self) -> None: super().__init__() self.chat_models: list[str | None] = [] async def chat(self, messages, model=None): self.chat_calls.append(list(messages)) self.chat_models.append(model) user = next((m["content"] for m in messages if m.get("role") == "user"), "") if isinstance(user, list): raise LLMError("simulated non-vision chat model (test sentinel)") first = user.split() return "Summary of " + (first[0] if first else "") class _FirstImageEmbedFailsLLM(FakeEmbedder): """Describes two images (the first with the :data:`EMBED_FAIL` sentinel) and FAILS the embed of any text carrying the sentinel — drives the per-image embed-failure branch (the pass moves on; the second image is still indexed).""" EMBED_FAIL = "EMBED-FAIL-SENTINEL" def __init__(self) -> None: super().__init__() self.chat_models: list[str | None] = [] self._describe_calls = 0 async def chat(self, messages, model=None): self.chat_calls.append(list(messages)) self.chat_models.append(model) user = next((m["content"] for m in messages if m.get("role") == "user"), "") if isinstance(user, list): self._describe_calls += 1 if self._describe_calls == 1: return f"First image. {self.EMBED_FAIL} details." return "Second image, a clean description." first = user.split() return "Summary of " + (first[0] if first else "") async def embed( self, texts: list[str], *, on_progress: Callable[[int, int], None] | None = None, ) -> list[list[float]]: if any(self.EMBED_FAIL in t for t in texts): raise EmbeddingError("simulated embed failure (test sentinel)") return await super().embed(texts, on_progress=on_progress) def _pdf_llm[ LLM: FakeEmbedder ]( tmp_path: Path, llm_cls: type[LLM] = _VisionLLM, images: bool = True, **kwargs: Any, ) -> LLM: """A fake LLM whose settings admit the ``pdf`` import token and carry the phase-122 image knobs (the image dir under *tmp_path*); *llm_cls* may be a mock vision client (the ``FakeEmbedder`` subclass carrying extra state such as ``chat_models``).""" kwargs.setdefault("_env_file", None) kwargs.setdefault("images", images) kwargs.setdefault("image_dir", str(tmp_path / "images")) kwargs.setdefault("import_extensions", "md,txt,pdf") llm = llm_cls() # pyright: ignore[reportCallIssue] llm.settings = Settings(**kwargs) # pyright: ignore[reportCallIssue] return llm def _build_pdf( path: Path, pages: Sequence[tuple[str, Sequence[bytes]]], title: str | None = None, ) -> Path: """One temp PDF: *pages* is (page text, embedded-image bytes) per page; *title* sets the metadata title (``None`` = leave it empty). Built with PyMuPDF itself (the phase-126 convention — no binary fixtures).""" doc = fitz.open() for text, imgs in pages: page = doc.new_page() if text: page.insert_text((72, 72), text) y = 150 for data in imgs: pix = fitz.Pixmap(data) page.insert_image( fitz.Rect(72, y, 72 + pix.width, y + pix.height), stream=data ) y += pix.height + 10 if title is not None: doc.set_metadata({"title": title}) doc.save(path) doc.close() return path def _doc(db, source: str, rel: str) -> Document: doc = db.scalar( select(Document).where(Document.source == source, Document.path == rel) ) assert doc is not None, f"no documents row for ({source!r}, {rel!r})" return doc def _cleanup_source(db, source: str) -> None: for doc in db.scalars(select(Document).where(Document.source == source)).all(): db.delete(doc) db.commit() # --- _extract_pdf_title ---------------------------------------------------- def test_extract_pdf_title_prefers_metadata_title(tmp_path: Path) -> None: path = _build_pdf(tmp_path / "t.pdf", [("text", [])], title=" My Report ") assert importer._extract_pdf_title(path, fallback="t") == "My Report" def test_extract_pdf_title_without_metadata_uses_fallback(tmp_path: Path) -> None: path = _build_pdf(tmp_path / "report.pdf", [("text", [])]) assert importer._extract_pdf_title(path, fallback="report") == "report" def test_extract_pdf_title_empty_or_blank_metadata_uses_fallback(tmp_path: Path) -> None: empty = _build_pdf(tmp_path / "a.pdf", [("text", [])], title="") assert importer._extract_pdf_title(empty, fallback="a") == "a" blank = _build_pdf(tmp_path / "b.pdf", [("text", [])], title=" ") assert importer._extract_pdf_title(blank, fallback="b") == "b" def test_extract_pdf_title_is_capped_to_the_column_width(tmp_path: Path) -> None: long_title = "x" * 600 path = _build_pdf(tmp_path / "long.pdf", [("text", [])], title=long_title) title = importer._extract_pdf_title(path, fallback="long") assert title == "x" * 500 and len(title) == 500 def test_extract_pdf_title_unreadable_file_uses_fallback(tmp_path: Path) -> None: bad = tmp_path / "bad.pdf" bad.write_bytes(b"not a pdf at all") assert importer._extract_pdf_title(bad, fallback="bad") == "bad" # --- the PDF index path (import_sources, real pipeline, fake LLM) ---------- def test_pdf_becomes_a_document_with_text_chunks_and_summary( db, tmp_path: Path ) -> None: """The primary path: a PDF in a source → sync → the document's ``content`` is the extracted text (the ``---PAGE-N---`` markers, A2), the page-aware chunks carry the ``Page N:`` anchor, every chunk (content + summary) is embedded, the title is the metadata title, and the phase-30/118 summary is stored.""" root = tmp_path / "pdfsrc" root.mkdir() _build_pdf( root / "report.pdf", [("First page text about networking.", []), ("Second page text about storage.", [])], title="Quarterly Report", ) llm = _pdf_llm(tmp_path) try: summary = asyncio.run(import_sources([root], llm, session=db)) assert (summary.files, summary.added) == (1, 1) assert summary.formats == {"pdf": 1} doc = _doc(db, "pdfsrc", "report.pdf") assert doc.title == "Quarterly Report" assert doc.is_image is False and doc.image_path is None assert doc.content == ( f"{PAGE_SEPARATOR.format(page=1)}\nFirst page text about networking.\n\n" f"{PAGE_SEPARATOR.format(page=2)}\nSecond page text about storage." ) chunks = db.scalars( select(Chunk).where(Chunk.document_id == doc.id).order_by(Chunk.position) ).all() content_chunks = [c for c in chunks if not c.is_summary] assert [c.content for c in content_chunks] == [ "Page 1:\nFirst page text about networking.", "Page 2:\nSecond page text about storage.", ] for c in chunks: assert c.embedding is not None and len(c.embedding) == 768 summary_chunks = [c for c in chunks if c.is_summary] assert len(summary_chunks) == 1 and summary_chunks[0].position == -1 assert doc.summary is not None assert doc.summary.endswith("Source: pdfsrc/report.pdf") finally: _cleanup_source(db, "pdfsrc") def test_pdf_embedded_images_become_image_documents(db, tmp_path: Path) -> None: """A3 with the toggle ON: each embedded image joins the phase-122 pipeline — the single seam describes it (the CHAT model), the copy lands in ``image_dir`` named by the doc id, and the doc row carries the synthetic ``#page=#img=`` path, ``is_image`` + ``image_path``, and the description as content AND summary (the phase-122 convention — the summary chunk mirrors it).""" root = tmp_path / "pdfimg" root.mkdir() png = _png_bytes(40, 30) _build_pdf(root / "report.pdf", [("Text page one.", []), ("Text page two.", [png])]) llm = _pdf_llm(tmp_path) try: summary = asyncio.run(import_sources([root], llm, session=db)) # The PDF + its one embedded image: both count added. assert (summary.files, summary.added, summary.images_failed) == (1, 2, 0) image_rel = "report.pdf#page=2#img=0.png" img = _doc(db, "pdfimg", image_rel) assert img.is_image is True and img.image_path is not None assert img.title == "Embedded image from report page 2" assert img.content == DESCRIPTION and img.summary == DESCRIPTION copy = Path(img.image_path) assert copy.parent == Path(llm.settings.image_dir).expanduser() assert copy.name == f"{img.id}.png" # row and copy agree on the id assert copy.read_bytes() == png chunks = db.scalars( select(Chunk).where(Chunk.document_id == img.id).order_by(Chunk.position) ).all() assert all(c.embedding is not None and len(c.embedding) == 768 for c in chunks) summary_chunks = [c for c in chunks if c.is_summary] assert len(summary_chunks) == 1 and summary_chunks[0].content == DESCRIPTION # The vision describe call went to the CHAT model (the PDF's own # lite summary is the first chat call — the phase-30 path). assert llm.chat_models == [ llm.settings.llm_summary_model, llm.settings.llm_chat_model, ] finally: _cleanup_source(db, "pdfimg") def test_pdf_embedded_images_skipped_when_images_off( db, tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: """The toggle OFF: the PDF text indexes normally, the embedded images are skipped (one debug line, NO warning, NO ``images_failed`` — a skip is not a failure), no image doc, no copy — and not even the image dir.""" root = tmp_path / "pdfoff" root.mkdir() _build_pdf( root / "report.pdf", [("Text page one.", [_png_bytes(40, 30)]), ("Text page two.", [])], ) llm = _pdf_llm(tmp_path, images=False) try: with caplog.at_level(logging.DEBUG, logger="app.importer"): summary = asyncio.run(import_sources([root], llm, session=db)) assert (summary.files, summary.added, summary.images_failed) == (1, 1, 0) doc = _doc(db, "pdfoff", "report.pdf") assert "Text page one." in doc.content and doc.summary is not None assert ( db.scalar( select(Document).where( Document.source == "pdfoff", Document.is_image.is_(True) ) ) is None ), "toggle off: no image documents" assert not Path(llm.settings.image_dir).expanduser().exists(), ( "no copy — and not even the image dir — may be left behind" ) # The skip is debug-logged (not a warning, not a failure). debugs = [ r for r in caplog.records if "images toggle off" in r.getMessage() ] assert len(debugs) == 1 and debugs[0].levelno == logging.DEBUG assert not [ r for r in caplog.records if "description failed" in r.getMessage() ] # No vision call at all — only the PDF's lite summary. assert llm.chat_models == [llm.settings.llm_summary_model] finally: _cleanup_source(db, "pdfoff") def test_pdf_unchanged_rerun_skips_everything(db, tmp_path: Path) -> None: """The unchanged path (same text digest) is the text path's rule: no re-extract, no re-embed, NO image pass — the second sync costs no vision and no embed calls at all.""" root = tmp_path / "pdfonce" root.mkdir() _build_pdf( root / "report.pdf", [("Text page one.", [_png_bytes(40, 30)]), ("Text page two.", [])], ) llm = _pdf_llm(tmp_path) try: s1 = asyncio.run(import_sources([root], llm, session=db)) assert (s1.added, s1.unchanged) == (2, 0) calls1, chat1 = len(llm.calls), len(llm.chat_models) s2 = asyncio.run(import_sources([root], llm, session=db)) assert (s2.files, s2.added, s2.updated, s2.unchanged) == (1, 0, 0, 1) assert len(llm.calls) == calls1, "unchanged: no re-embed" assert len(llm.chat_models) == chat1, "unchanged: no vision/summary calls" finally: _cleanup_source(db, "pdfonce") def test_pdf_unchanged_path_backfills_and_honors_manual_date( db, tmp_path: Path ) -> None: """The unchanged path's phase-118/D1/D4 dance, PDF flavour: a NULL-summary unchanged doc gets the summary backfill (counted ``summary_backfilled``, never ``summaries``); a manual-date row (D1) is not touched at all (the backfill runs BEFORE the early-return — the flag protects the DATE only); and with the flag off the stored date REFRESHES from the file's mtime (D4 — counted ``dates_updated``, still ``unchanged``).""" root = tmp_path / "pdfdate" root.mkdir() _build_pdf(root / "report.pdf", [("Text page one.", []), ("Text page two.", [])]) llm = _pdf_llm(tmp_path, images=False) try: s1 = asyncio.run(import_sources([root], llm, session=db)) assert s1.added == 1 doc = _doc(db, "pdfdate", "report.pdf") # Run 2 — simulate a pre-phase-30 row: NULL summary + a manual # date correction. doc.summary = None doc.created_at_manual = True doc.created_at = doc.created_at - timedelta(days=30) db.commit() manual_date = doc.created_at s2 = asyncio.run(import_sources([root], llm, session=db)) assert (s2.unchanged, s2.summary_backfilled, s2.summaries) == (1, 1, 0) assert s2.dates_updated == 0, "the manual date must survive the sync" doc2 = _doc(db, "pdfdate", "report.pdf") assert doc2.created_at_manual is True and doc2.created_at == manual_date assert doc2.summary is not None, "the backfill restored the summary" # Run 3 — the manual flag off: the date REFRESHES from the # file's mtime (D4) — and a date-only refresh is still # unchanged (no KB-change count). doc2.created_at_manual = False doc2.created_at = doc2.created_at - timedelta(days=60) db.commit() s3 = asyncio.run(import_sources([root], llm, session=db)) assert (s3.unchanged, s3.dates_updated, s3.added, s3.updated) == (1, 1, 0, 0) doc3 = _doc(db, "pdfdate", "report.pdf") assert doc3.created_at != manual_date finally: _cleanup_source(db, "pdfdate") def test_pdf_text_change_updates_doc_and_keeps_unchanged_images( db, tmp_path: Path ) -> None: """A text change re-indexes the PDF (updated) and re-runs the image pass — but an embedded image whose BYTES are unchanged (same digest) is skipped: no re-describe (no new vision call for it), the doc id and the copy path are stable.""" root = tmp_path / "pdfchg" root.mkdir() png = _png_bytes(40, 30) _build_pdf( root / "report.pdf", [("Original first page.", []), ("Second page.", [png])], ) llm = _pdf_llm(tmp_path) try: s1 = asyncio.run(import_sources([root], llm, session=db)) assert s1.added == 2 image_rel = "report.pdf#page=2#img=0.png" img1 = _doc(db, "pdfchg", image_rel) # Same image bytes, changed page-1 text. (root / "report.pdf").unlink() _build_pdf( root / "report.pdf", [("Revised first page.", []), ("Second page.", [png])], ) s2 = asyncio.run(import_sources([root], llm, session=db)) assert (s2.added, s2.updated, s2.unchanged) == (0, 1, 0) img2 = _doc(db, "pdfchg", image_rel) assert img2.id == img1.id and img2.image_path == img1.image_path assert img2.image_path is not None assert Path(img2.image_path).read_bytes() == png # The lite summary of the changed PDF fired again — but NO new # describe call (the image bytes are unchanged). assert llm.chat_models == [ llm.settings.llm_summary_model, # run 1: PDF summary llm.settings.llm_chat_model, # run 1: the image description llm.settings.llm_summary_model, # run 2: the revised PDF summary ] finally: _cleanup_source(db, "pdfchg") def test_pdf_without_text_takes_the_screenshot_fallback( db, tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: """Phase 133 (task 03): an image-only PDF (an EMPTY text layer — the gate's other target) takes the page-screenshot fallback: the single page is rendered to a PNG and vision-described (the mock's canned :data:`DESCRIPTION`), and the doc lands with the ``---PAGE-N---``- anchored description as content (chunked + summarized as any other PDF). The phase-126 empty-content behavior is retired for layers that CAN be screenshotted — it survives only for an unopenable file (``test_corrupt_pdf_file_becomes_an_empty_document``), where the render fails too. The loud fallback warning fires; the phase- 126 "no extractable text" warning does NOT (the doc is not empty). """ root = tmp_path / "pdfempty" root.mkdir() _build_pdf(root / "pics.pdf", [("", [_png_bytes(40, 30)])]) llm = _pdf_llm(tmp_path, images=False) try: with caplog.at_level(logging.INFO, logger="app.importer"): summary = asyncio.run(import_sources([root], llm, session=db)) assert (summary.files, summary.added) == (1, 1) assert summary.pdfs_screenshotted == 1 doc = _doc(db, "pdfempty", "pics.pdf") assert doc.content == f"{PAGE_SEPARATOR.format(page=1)}\n{DESCRIPTION}" assert doc.content_hash == hashlib.sha256(doc.content.encode("utf-8")).hexdigest() chunks = db.scalars( select(Chunk).where(Chunk.document_id == doc.id).order_by(Chunk.position) ).all() content_chunks = [c for c in chunks if not c.is_summary] assert [c.content for c in content_chunks] == [f"Page 1:\n{DESCRIPTION}"] assert doc.summary is not None # The loud fallback warning (the FINDING's signature) fired — # and the phase-126 "no extractable text" warning did NOT (the # doc is not empty — the screenshots ARE the content). warnings = [ r for r in caplog.records if "text layer unusable" in r.getMessage() ] assert len(warnings) == 1 and warnings[0].levelno == logging.WARNING assert "falling back to 1 page screenshots" in warnings[0].getMessage() assert not [ r for r in caplog.records if "no extractable text" in r.getMessage() ] finally: _cleanup_source(db, "pdfempty") def test_corrupt_pdf_file_becomes_an_empty_document( db, tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: """A .pdf file that is not a PDF at all: the parser's fail-soft contract (no text, no images) AND the phase-133 fallback's render-less branch (``render_pdf_pages`` → ``[]`` → the helper's ``""``) — the document lands with empty content (the empty byte string's digest) and no chunks, the fallback is still counted (the gate fired, the render produced nothing), and the phase-126 "no extractable text" warning fires — the doc really is empty.""" root = tmp_path / "pdfbad" root.mkdir() (root / "broken.pdf").write_bytes(b"this is not a pdf at all") llm = _pdf_llm(tmp_path) try: with caplog.at_level(logging.INFO, logger="app.importer"): summary = asyncio.run(import_sources([root], llm, session=db)) assert (summary.files, summary.added, summary.errors) == (1, 1, 0) assert summary.pdfs_screenshotted == 1 # the gate fired doc = _doc(db, "pdfbad", "broken.pdf") assert doc.content == "" and doc.title == "broken" assert doc.content_hash == hashlib.sha256(b"").hexdigest() chunks = db.scalars( select(Chunk).where(Chunk.document_id == doc.id).order_by(Chunk.position) ).all() assert [c for c in chunks if not c.is_summary] == [] warnings = [ r for r in caplog.records if "no extractable text" in r.getMessage() ] assert len(warnings) == 1 and warnings[0].levelno == logging.WARNING finally: _cleanup_source(db, "pdfbad") def test_pdf_embedded_image_unsupported_mime_is_skipped( db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: """An embedded image whose mime maps to no ``BOR_IMAGE_EXTENSIONS`` member (a TIFF under the default set) is skipped with a warning — NOT a failure (``images_failed`` stays 0), no row, no copy.""" root = tmp_path / "pdfmime" root.mkdir() _build_pdf(root / "report.pdf", [("Text page.", [])]) monkeypatch.setattr( importer, "extract_pdf_images", lambda _path: [ PdfImage(page=1, index=0, width=5, height=5, data=b"fakedata", mime="image/tiff") ], ) llm = _pdf_llm(tmp_path) try: with caplog.at_level(logging.INFO, logger="app.importer"): summary = asyncio.run(import_sources([root], llm, session=db)) assert (summary.files, summary.added, summary.images_failed) == (1, 1, 0) assert ( db.scalar( select(Document).where( Document.source == "pdfmime", Document.is_image.is_(True) ) ) is None ) assert not Path(llm.settings.image_dir).expanduser().exists() warnings = [ r for r in caplog.records if "unsupported extension" in r.getMessage() ] assert len(warnings) == 1 and "image/tiff" in warnings[0].getMessage() finally: _cleanup_source(db, "pdfmime") def test_pdf_embedded_image_failed_description_skips_image_keeps_pdf( db, tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: """LOCKED A3 fail-soft: a failed description skips the IMAGE entirely (no row, no copy — and not even the image dir), counts ``images_failed`` + a warning, and the PDF's text index is unaffected (already committed).""" root = tmp_path / "pdffail" root.mkdir() _build_pdf( root / "report.pdf", [("Text page one.", []), ("Text page two.", [_png_bytes(40, 30)])], ) llm = _pdf_llm(tmp_path, _VisionFailsLLM) try: with caplog.at_level(logging.INFO, logger="app.importer"): summary = asyncio.run(import_sources([root], llm, session=db)) assert (summary.files, summary.added, summary.images_failed) == (1, 1, 1) doc = _doc(db, "pdffail", "report.pdf") assert "Text page one." in doc.content and doc.summary is not None assert ( db.scalar( select(Document).where( Document.source == "pdffail", Document.is_image.is_(True) ) ) is None ), "a failed description must leave NO image row" assert not Path(llm.settings.image_dir).expanduser().exists(), ( "no copy — and not even the image dir — may be left behind" ) warnings = [ r for r in caplog.records if "embedded image description failed" in r.getMessage() ] assert len(warnings) == 1 and warnings[0].levelno == logging.WARNING finally: _cleanup_source(db, "pdffail") def test_pdf_embedded_image_embed_failure_keeps_the_others( db, tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: """A per-image embed failure rolls back THAT image only (no row, no orphan copy), counts ``errors`` + the error line, and the pass moves on — the second image is still indexed, and the PDF's text index (committed earlier) is untouched.""" root = tmp_path / "pdfemb" root.mkdir() _build_pdf( root / "report.pdf", [ ("Text page one.", []), ("Text page two.", [_png_bytes(40, 30)]), ("Text page three.", [_png_bytes(30, 20)]), ], ) llm = _pdf_llm(tmp_path, _FirstImageEmbedFailsLLM) try: with caplog.at_level(logging.INFO, logger="app.importer"): summary = asyncio.run(import_sources([root], llm, session=db)) # PDF + the surviving second image; the first image's embed failed. assert (summary.added, summary.errors, summary.images_failed) == (2, 1, 0) errors = [ r for r in caplog.records if "embed failed" in r.getMessage() ] assert len(errors) == 1 and errors[0].levelno == logging.ERROR # Only the second image (page 3) landed. img = _doc(db, "pdfemb", "report.pdf#page=3#img=0.png") assert img.is_image is True assert ( db.scalar( select(Document).where( Document.source == "pdfemb", Document.path == "report.pdf#page=2#img=0.png", ) ) is None ) # No orphan copy of the failed image: exactly one file in the dir. copies = list(Path(llm.settings.image_dir).expanduser().iterdir()) assert len(copies) == 1 and copies[0].name == f"{img.id}.png" finally: _cleanup_source(db, "pdfemb") def test_pdf_changed_embedded_image_replaces_the_copy(db, tmp_path: Path) -> None: """A PDF whose text AND an embedded image's bytes changed: the PDF is ``updated`` and the image doc is ``updated`` in place — same id, the stale copy is deleted before the new one is written (one file, the new bytes), and the image is re-described (a second vision call).""" root = tmp_path / "pdfrep" root.mkdir() png_a = _png_bytes(40, 30) png_b = _png_bytes(25, 18) # different size → different bytes _build_pdf( root / "report.pdf", [("Original first page.", []), ("Second page.", [png_a])], ) llm = _pdf_llm(tmp_path) try: s1 = asyncio.run(import_sources([root], llm, session=db)) assert s1.added == 2 image_rel = "report.pdf#page=2#img=0.png" img1 = _doc(db, "pdfrep", image_rel) assert img1.image_path is not None copy1 = Path(img1.image_path) # Changed page-1 text AND changed image bytes. (root / "report.pdf").unlink() _build_pdf( root / "report.pdf", [("Revised first page.", []), ("Second page.", [png_b])], ) s2 = asyncio.run(import_sources([root], llm, session=db)) assert (s2.added, s2.updated, s2.unchanged) == (0, 2, 0) img2 = _doc(db, "pdfrep", image_rel) assert img2.id == img1.id, "the changed image keeps its doc id" assert img2.image_path == img1.image_path, "the copy path is stable" assert img2.image_path is not None assert Path(img2.image_path).read_bytes() == png_b # Exactly one copy in the dir — the stale one was deleted. copies = list(Path(llm.settings.image_dir).expanduser().iterdir()) assert len(copies) == 1 and copies[0] == copy1 # The re-describe fired (a second vision call after run 2's lite # summary of the revised PDF). assert llm.chat_models == [ llm.settings.llm_summary_model, # run 1: PDF summary llm.settings.llm_chat_model, # run 1: the image description llm.settings.llm_summary_model, # run 2: the revised PDF summary llm.settings.llm_chat_model, # run 2: the re-description ] finally: _cleanup_source(db, "pdfrep") def _png_bytes(width: int, height: int, alpha: bool = False) -> bytes: pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, width, height), 1 if alpha else 0) pix.clear_with(200) return pix.tobytes("png") def _jpeg_bytes(width: int, height: int) -> bytes: pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, width, height)) pix.clear_with(200) return pix.tobytes("jpeg") @pytest.fixture() def single_page_pdf(tmp_path: Path) -> Path: path = tmp_path / "single.pdf" doc = fitz.open() doc.new_page().insert_text((72, 72), "Hello single page.") doc.save(path) doc.close() return path @pytest.fixture() def multi_page_pdf(tmp_path: Path) -> Path: """3 pages: p1 text, p2 text + PNG, p3 text + JPEG twice + gray PNG.""" path = tmp_path / "multi.pdf" doc = fitz.open() doc.new_page().insert_text((72, 72), "Alpha text one.") p2 = doc.new_page() p2.insert_text((72, 72), "Beta text two.") p2.insert_image(fitz.Rect(72, 150, 172, 180), stream=_png_bytes(40, 30)) p3 = doc.new_page() p3.insert_text((72, 72), "Gamma text three.") p3.insert_image(fitz.Rect(72, 150, 172, 180), stream=_jpeg_bytes(40, 30)) p3.insert_image(fitz.Rect(200, 150, 260, 172), stream=_jpeg_bytes(40, 30)) p3.insert_image(fitz.Rect(72, 200, 102, 230), stream=_png_bytes(15, 15)) doc.save(path) doc.close() return path @pytest.fixture() def image_only_pdf(tmp_path: Path) -> Path: path = tmp_path / "image_only.pdf" doc = fitz.open() doc.new_page().insert_image(fitz.Rect(72, 72, 200, 150), stream=_png_bytes(80, 60)) doc.save(path) doc.close() return path @pytest.fixture() def blank_middle_page_pdf(tmp_path: Path) -> Path: """p1 text, p2 blank (no text), p3 text.""" path = tmp_path / "blank_middle.pdf" doc = fitz.open() doc.new_page().insert_text((72, 72), "First text.") doc.new_page() doc.new_page().insert_text((72, 72), "Third text.") doc.save(path) doc.close() return path # --------------------------------------------------------------------------- # extract_pdf_text # --------------------------------------------------------------------------- def test_single_page_text_has_exactly_one_separator(single_page_pdf: Path) -> None: text = extract_pdf_text(single_page_pdf) assert text == f"{PAGE_SEPARATOR.format(page=1)}\nHello single page." assert text.count("---PAGE-") == 1 def test_multi_page_text_separators_and_order(multi_page_pdf: Path) -> None: text = extract_pdf_text(multi_page_pdf) assert text == ( f"{PAGE_SEPARATOR.format(page=1)}\nAlpha text one.\n\n" f"{PAGE_SEPARATOR.format(page=2)}\nBeta text two.\n\n" f"{PAGE_SEPARATOR.format(page=3)}\nGamma text three." ) assert [f"---PAGE-{n}---" for n in (1, 2, 3)] == [ line for line in text.splitlines() if line.startswith("---PAGE-") ] def test_blank_middle_page_gets_no_separator(blank_middle_page_pdf: Path) -> None: text = extract_pdf_text(blank_middle_page_pdf) expected = ( f"{PAGE_SEPARATOR.format(page=1)}\nFirst text.\n\n" f"{PAGE_SEPARATOR.format(page=3)}\nThird text." ) assert text == expected assert "---PAGE-2---" not in text def test_image_only_pdf_returns_empty_string(image_only_pdf: Path) -> None: assert extract_pdf_text(image_only_pdf) == "" def test_missing_file_text_returns_empty_no_raise(tmp_path: Path) -> None: assert extract_pdf_text(tmp_path / "nope.pdf") == "" def test_garbage_file_text_returns_empty_no_raise(tmp_path: Path) -> None: bad = tmp_path / "garbage.pdf" bad.write_bytes(b"this is not a pdf at all") assert extract_pdf_text(bad) == "" def test_broken_page_is_skipped(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: """A page whose get_text raises is logged/skipped; other pages stay.""" path = tmp_path / "fake.pdf" path.write_bytes(b"%PDF-1.4 fake") class _Page: def __init__(self, text: str | None) -> None: self._text = text def get_text(self, flavor: str) -> str: if self._text is None: raise RuntimeError("boom: page unreadable") return self._text monkeypatch.setattr( pdf_mod.fitz, "open", lambda _p: _FakeDoc([_Page(None), _Page("Fine text.\n")]), ) assert extract_pdf_text(path) == f"{PAGE_SEPARATOR.format(page=2)}\nFine text." # --------------------------------------------------------------------------- # extract_pdf_images # --------------------------------------------------------------------------- def test_pdf_without_images_returns_empty_list(single_page_pdf: Path) -> None: assert extract_pdf_images(single_page_pdf) == [] def test_embedded_images_extracted_in_order(multi_page_pdf: Path) -> None: images = extract_pdf_images(multi_page_pdf) assert [(img.page, img.index) for img in images] == [ (2, 0), (3, 0), (3, 1), (3, 2), ] # page 2: 40x30 PNG img = images[0] assert (img.width, img.height) == (40, 30) assert img.mime == "image/png" assert img.data.startswith(PNG_MAGIC) # page 3: 40x30 JPEG, then the same JPEG again (distinct index) for img in images[1:3]: assert (img.width, img.height) == (40, 30) assert img.mime == "image/jpeg" assert img.data.startswith(JPEG_MAGIC) assert images[1].data == images[2].data # same embedded xref # page 3: 15x15 PNG img = images[3] assert (img.width, img.height) == (15, 15) assert img.mime == "image/png" assert img.data.startswith(PNG_MAGIC) # every entry is a well-formed PdfImage of the right shape for img in images: assert isinstance(img, PdfImage) assert isinstance(img.page, int) and img.page >= 1 assert isinstance(img.index, int) and img.index >= 0 assert isinstance(img.data, bytes) and img.data assert isinstance(img.mime, str) and img.mime.startswith("image/") def test_image_data_round_trips_through_fitz(multi_page_pdf: Path) -> None: """Extracted bytes are genuine decodable image bytes (not garbage).""" images = extract_pdf_images(multi_page_pdf) assert images for img in images: pix = fitz.Pixmap(img.data) assert (pix.width, pix.height) == (img.width, img.height) def test_missing_file_images_returns_empty_no_raise(tmp_path: Path) -> None: assert extract_pdf_images(tmp_path / "nope.pdf") == [] def test_garbage_file_images_returns_empty_no_raise(tmp_path: Path) -> None: bad = tmp_path / "garbage.pdf" bad.write_bytes(b"definitely not a pdf") assert extract_pdf_images(bad) == [] def _fake_image_page(page_entries: list[dict[str, Any]]) -> SimpleNamespace: return SimpleNamespace(get_image_info=lambda xrefs: [dict(e) for e in page_entries]) class _FakeDoc: """A stand-in fitz Document: given pages, stubbed extract_image.""" def __init__(self, pages: list[Any], extract: Any = None) -> None: self._pages = pages self._extract = extract def __len__(self) -> int: return len(self._pages) def __getitem__(self, index: int) -> Any: return self._pages[index] def extract_image(self, xref: int) -> Any: if self._extract is None: pytest.fail("extract_image() called unexpectedly") return self._extract(xref) def __enter__(self) -> _FakeDoc: return self def __exit__(self, *args: object) -> bool: return False def test_image_without_xref_is_skipped( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: path = tmp_path / "fake.pdf" path.write_bytes(b"%PDF-1.4 fake") doc = _FakeDoc( [_fake_image_page([{"xref": 0, "width": 5, "height": 5}])], lambda x: pytest.fail("no call"), ) monkeypatch.setattr(pdf_mod.fitz, "open", lambda _p: doc) assert extract_pdf_images(path) == [] def test_image_extraction_failure_is_skipped( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: path = tmp_path / "fake.pdf" path.write_bytes(b"%PDF-1.4 fake") def _boom(xref: int) -> None: raise RuntimeError("corrupt xref") doc = _FakeDoc([_fake_image_page([{"xref": 7, "width": 5, "height": 5}])], _boom) monkeypatch.setattr(pdf_mod.fitz, "open", lambda _p: doc) assert extract_pdf_images(path) == [] def test_image_with_empty_data_is_skipped( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: path = tmp_path / "fake.pdf" path.write_bytes(b"%PDF-1.4 fake") doc = _FakeDoc( [_fake_image_page([{"xref": 7, "width": 5, "height": 5}])], lambda xref: {"image": b"", "ext": "png", "width": 5, "height": 5}, ) monkeypatch.setattr(pdf_mod.fitz, "open", lambda _p: doc) assert extract_pdf_images(path) == [] def test_image_listing_failure_is_skipped( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """A page whose image listing raises (beyond the version fallback) is skipped.""" path = tmp_path / "fake.pdf" path.write_bytes(b"%PDF-1.4 fake") def _boom(xrefs: bool) -> None: raise RuntimeError("page image listing failed") page = SimpleNamespace( get_image_info=_boom, get_images=lambda full: pytest.fail("should not be called"), ) doc = _FakeDoc([page]) monkeypatch.setattr(pdf_mod.fitz, "open", lambda _p: doc) assert extract_pdf_images(path) == [] def test_extracted_ext_wins_over_info_ext( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """The image info's `ext` is unreliable (None) in PyMuPDF; the ext of the actually returned bytes must drive the MIME type.""" path = tmp_path / "fake.pdf" path.write_bytes(b"%PDF-1.4 fake") doc = _FakeDoc( [_fake_image_page([{"xref": 7, "width": 5, "height": 5, "ext": None, "type": 3}])], lambda xref: {"image": b"x" * 10, "ext": "jpeg", "width": 5, "height": 5}, ) monkeypatch.setattr(pdf_mod.fitz, "open", lambda _p: doc) images = extract_pdf_images(path) assert len(images) == 1 assert images[0].mime == "image/jpeg" # ext beats the PNG type code # --------------------------------------------------------------------------- # _mime_from_image_info # --------------------------------------------------------------------------- @pytest.mark.parametrize( ("info", "expected"), [ ({"ext": "png"}, "image/png"), ({"ext": "PNG"}, "image/png"), ({"ext": "jpg"}, "image/jpeg"), ({"ext": "jpeg"}, "image/jpeg"), ({"ext": "gif"}, "image/gif"), ({"ext": "webp"}, "image/webp"), ({"ext": "tif"}, "image/tiff"), ({"ext": "tiff"}, "image/tiff"), # extension wins over the type code ({"ext": "png", "type": 2}, "image/png"), # known type codes (phase-126 design) ({"type": 2}, "image/jpeg"), ({"type": 3}, "image/png"), ({"type": 4}, "image/gif"), ({"type": 8}, "image/tiff"), ({"type": 9}, "image/webp"), # unknown / missing → octet-stream fallback ({"type": 7}, "application/octet-stream"), ({"type": 42}, "application/octet-stream"), ({"type": None, "ext": None}, "application/octet-stream"), ({"ext": "unknown-ext"}, "application/octet-stream"), ({"type": "not-an-int"}, "application/octet-stream"), ({}, "application/octet-stream"), ], ) def test_mime_from_image_info(info: dict[str, Any], expected: str) -> None: assert _mime_from_image_info(info) == expected # --------------------------------------------------------------------------- # _page_image_entries (PyMuPDF version fallback) # --------------------------------------------------------------------------- def test_page_image_entries_prefers_get_image_info() -> None: page = SimpleNamespace( get_image_info=lambda xrefs: [ {"xref": 1, "width": 4, "height": 5, "ext": None}, ], get_images=lambda full: pytest.fail("should not be called"), ) entries = _page_image_entries(page) assert entries == [{"xref": 1, "width": 4, "height": 5, "ext": None}] # returned dicts are copies — mutating them cannot touch the page entries[0]["xref"] = 999 assert page.get_image_info(xrefs=True)[0]["xref"] == 1 def test_page_image_entries_falls_back_on_old_pymupdf() -> None: def _no_xrefs_kwarg(xrefs: bool) -> None: raise TypeError("get_image_info() got an unexpected keyword argument 'xrefs'") page = SimpleNamespace( get_image_info=_no_xrefs_kwarg, # (xref, smask, width, height, bpc, colorspace, alt_cs, name, filter) get_images=lambda full: [(7, 0, 40, 30, 8, "DeviceRGB", "", "Im0", "DCTDecode", 0)], ) entries = _page_image_entries(page) assert entries == [{"xref": 7, "width": 40, "height": 30}] # --------------------------------------------------------------------------- # is_usable_pdf_text (phase 133 task 02 — the deterministic garbage gate) # --------------------------------------------------------------------------- #: The owner's untracked fixture (the stable reference — the same #: provenance as the phase-133 task-01 integration regression pin: #: the tarball, never the extracted PDF, is what the test starts #: from, and it never touches the repo). TARBALL = Path(__file__).resolve().parents[2] / "test_sources" / "hnb_2026_guide.tar.gz" #: The inner file's name (it keeps its name through extraction — the #: task-01 convention). PDF_NAME = "HBI-2026-Guide-Year-Round_260704_131221.pdf" @pytest.fixture(scope="module") def pre126_garbage_text(tmp_path_factory: pytest.TempPathFactory) -> str: """THE REAL GARBAGE (the task's airtight pin, module-scoped): the owner's exact PDF read with the PRE-PHASE-126 recipe — ``read_text(encoding="utf-8", errors="replace")`` + ``.replace("\\x00", "")`` — the 46%-XML-ish, U+FFFD-heavy soup that was chunked and embedded before the parser shipped (the FINDING: 138,170 non-empty lines, ~46% XML-ish, the owner's ```` preview). The tarball's single member is extracted into a temp dir (the task-01 integration pin's pattern, shared — the fixture stays out of the repo).""" if not TARBALL.is_file(): pytest.skip( "test_sources/hnb_2026_guide.tar.gz is missing — the " "untracked owner fixture (2.6 MB, one 48-page Illustrator PDF)" ) root = tmp_path_factory.mktemp("pre126") with tarfile.open(TARBALL, "r:gz") as tar: tar.extractall(root, filter="data") member = root / PDF_NAME return member.read_text(encoding="utf-8", errors="replace").replace("\x00", "") def test_gate_clean_text_with_page_separators_is_usable() -> None: """Real prose + the phase-126 ``---PAGE-N---`` separators: the happy path the gate must NEVER block (a clean PDF never burns a vision call — the gate is the only gate, and the happy path stays byte-identical to phase 126).""" clean = ( f"{PAGE_SEPARATOR.format(page=1)}\n2026 U.S. Benefits Guide\n" "\n" "Covered employees receive comprehensive medical, dental, and\n" "vision benefits under this plan document.\n" "\n" f"{PAGE_SEPARATOR.format(page=2)}\nPremiums are deducted from\n" "each paycheck; the plan year runs January 1 through\n" "December 31." ) assert is_usable_pdf_text(clean) is True def test_gate_empty_text_layer_is_unusable() -> None: """The empty text layer — a scanned/image-only PDF has nothing to read — is the fallback's other target.""" assert is_usable_pdf_text("") is False assert is_usable_pdf_text(" \n\t\n \r\n") is False def test_gate_xml_heavy_layer_is_unusable() -> None: """The raw-XMP / binary-soup signature: ≥ 30% of the non-empty lines look like tags (````, ````, ``value" for i in range(30)] prose = [f"Line {i} of ordinary prose." for i in range(70)] assert is_usable_pdf_text("\n".join(tags + prose)) is False # 30/100 def test_gate_borderline_document_about_xml_stays_usable() -> None: """ASSUMPTION A8's edge from below: a real document ABOUT XML whose tag-like lines stay UNDER the 0.30 bar (29%) is usable — the documented false-positive cost is paid only at/above the bar.""" tags = [f"text" for i in range(29)] prose = [ f"Paragraph {i} explains the markup in plain words." for i in range(71) ] assert is_usable_pdf_text("\n".join(tags + prose)) is True # 29/100 def test_gate_at_and_above_the_xml_bar_is_unusable() -> None: """The threshold's edges from above: EXACTLY 30% tag-like lines is already unusable (the bar is ≥, not >), and 31% is too — a document ABOUT XML that reaches the bar pays the documented ASSUMPTION A8 cost (the fallback still extracts it correctly via vision).""" for n in (30, 31): tags = [f"text" for i in range(n)] prose = [ f"Paragraph {i} of ordinary explanation." for i in range(100 - n) ] assert is_usable_pdf_text("\n".join(tags + prose)) is False def test_gate_replacement_char_soup_is_unusable() -> None: """The SECOND signal, alone: a ≥ 5% U+FFFD load (the ``errors="replace"`` binary-soup signal) is unusable even with few tag-like lines — the XML-ish gate does not mask it, and this soup has none (the two failure signals fire independently).""" soup = "Benefits summary paragraph. " * 24 + "\ufffd" * 60 # ~7.9% assert is_usable_pdf_text(soup) is False def test_gate_structural_newlines_and_tabs_are_printable() -> None: """The documented exclusion: the ``\\n``/``\\t``/``\\r`` the extractor legitimately emits are NOT non-printable — a newline/tab/indent-heavy document is printable (counting them would flag every prose document).""" text = "\n\t".join(f"Indented line {i} with a tab." for i in range(20)) assert is_usable_pdf_text(text) is True def test_gate_real_pre126_garbage_is_unusable( pre126_garbage_text: str, ) -> None: """THE REAL GARBAGE (the airtight pin): the owner's exact file, read with the pre-126 recipe (the module fixture) — the reproduced 46%-XML-ish, U+FFFD-heavy soup — must be flagged unusable. Both signals fire on the REAL string (the FINDING's evidence, pinned against the named thresholds: if the owner ever re-cuts the fixture, the drift shows up here — the pin has teeth, the same contract as the task-01 suite).""" assert is_usable_pdf_text(pre126_garbage_text) is False # The owner's exact preview line is in the reproduced soup. assert "" in pre126_garbage_text lines = [line for line in pre126_garbage_text.splitlines() if line.strip()] xml_frac = sum(1 for line in lines if _XMLISH_LINE_RE.search(line)) / len(lines) nonprint_frac = sum( 1 for ch in pre126_garbage_text if ch == "\ufffd" or (ord(ch) < 32 and ch not in "\t\n\r") ) / len(pre126_garbage_text) # Both signals independently clear their named bars on the real # string — neither signal can silently die. assert xml_frac >= XMLISH_LINE_FRACTION assert nonprint_frac >= NONPRINT_CHAR_FRACTION # --------------------------------------------------------------------------- # render_pdf_pages (phase 133 task 03 — the fallback's visual input) # --------------------------------------------------------------------------- def _two_page_text_pdf(path: Path) -> Path: """A 2-page temp PDF with EXPLICIT letter pages (612x792pt — deterministic render sizes: 1275x1650 px at 150 DPI, 612x792 at 72 DPI) and page-specific text (the render must be page-specific, so the pages' PNG bytes must differ).""" doc = fitz.open() doc.new_page(width=612, height=792).insert_text((72, 72), "First page marker text.") doc.new_page(width=612, height=792).insert_text((72, 72), "Second page marker text.") doc.save(path) doc.close() return path def test_render_pdf_pages_returns_one_png_per_page_in_order( tmp_path: Path, ) -> None: """A 2-page PDF → 2 PNGs in PAGE ORDER (list index i is page i+1), each a genuine decodable PNG at the default 150-DPI render (612x792pt letter → 1275x1650 px), and page-specific (page 1's bytes != page 2's — different text on the page).""" path = _two_page_text_pdf(tmp_path / "two.pdf") pages = render_pdf_pages(path) assert len(pages) == 2 for png in pages: assert png.startswith(PNG_MAGIC) pix = fitz.Pixmap(png) assert (pix.width, pix.height) == (1275, 1650) assert pages[0] != pages[1] # the render is page-specific def test_render_pdf_pages_honors_the_dpi_argument(tmp_path: Path) -> None: """ASSUMPTION A8's fixed 150 DPI is the DEFAULT; an explicit *dpi* changes the render (72 DPI → the page's natural 612x792 px).""" path = _two_page_text_pdf(tmp_path / "two.pdf") pix = fitz.Pixmap(render_pdf_pages(path, dpi=72)[0]) assert (pix.width, pix.height) == (612, 792) def test_render_pdf_pages_unopenable_file_returns_empty_list( tmp_path: Path, caplog: pytest.LogCaptureFixture, ) -> None: """A file that cannot be OPENED yields ``[]`` (the caller's existing empty-content path — the doc lands with the no-extractable-text behavior, unchanged) + a warning (the fail-soft style).""" bad = tmp_path / "bad.pdf" bad.write_bytes(b"this is not a pdf at all") with caplog.at_level(logging.WARNING, logger="app.rag.pdf"): assert render_pdf_pages(bad) == [] assert render_pdf_pages(tmp_path / "missing.pdf") == [] warnings = [ r for r in caplog.records if "could not open" in r.getMessage() ] assert len(warnings) == 2 and all(w.levelno == logging.WARNING for w in warnings) def test_render_pdf_pages_zero_page_doc_returns_empty_list( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: """A zero-page PDF (openable, no pages — this PyMuPDF refuses to SAVE such a file, so it is stubbed) yields ``[]`` — the loop runs zero times, nothing renders.""" path = tmp_path / "zero.pdf" path.write_bytes(b"%PDF-1.4 fake") monkeypatch.setattr(pdf_mod.fitz, "open", lambda _p: _FakeDoc([])) assert render_pdf_pages(path) == [] def test_render_pdf_pages_broken_page_is_zero_bytes_at_its_index( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: """A page whose get_pixmap raises → ``b""`` at ITS index (the caller's placeholder slot — the page index NEVER shifts, so ``---PAGE-N---`` stays truthful); the good pages keep their PNGs at their own indices, and the failure is logged (the :func:`extract_pdf_text` fail-soft style).""" path = tmp_path / "fake.pdf" path.write_bytes(b"%PDF-1.4 fake") class _RenderPage: def __init__(self, data: bytes | None) -> None: self._data = data def get_pixmap(self, dpi: int) -> Any: if self._data is None: raise RuntimeError("boom: page render failed") return SimpleNamespace(tobytes=lambda _fmt: self._data) doc = _FakeDoc( [ _RenderPage(b"\x89PNG-fake-page-1"), _RenderPage(None), _RenderPage(b"\x89PNG-fake-page-3"), ] ) monkeypatch.setattr(pdf_mod.fitz, "open", lambda _p: doc) assert render_pdf_pages(path) == [ b"\x89PNG-fake-page-1", b"", b"\x89PNG-fake-page-3", ]