ducoterra
25de518dc9
All verification is complete — everything is green. No repo-side changes were needed in this pass (the phase's work in the working tree is correct and complete).
## Phase 133 final verification report
- **Verified (already implemented in working tree):** `is_usable_pdf_text` gate (0.30 XML-ish / 0.05 non-printable / empty), `render_pdf_pages`, `_fallback_page_screenshots` + `pdfs_screenshotted` counter in `app/rag/{pdf,importer}.py`; regression pin, unit/integration/E2E suites.
- **Completion criteria:**
1. Real guide clean through real pipeline — ✅ `test_pdf_hnb_guide_regression` + E2E preview/viewer checks (0 XML-ish lines, first-page text present).
2. Garbage PDF → vision fallback end to end — ✅ unit + E2E (stored content = canned per-page descriptions, garbage not stored, sync `detail` shape unchanged).
3. Clean PDF never burns a fallback vision call — ✅ unit + E2E (byte-faithful content).
4. Gates — ✅ `uv run pytest --cov=app --cov-report=term-missing`: **3129 passed, app/ 99%** (>90%); `uv run pytest tests/e2e/test_garbage_pdf.py -v --no-cov`: **3 passed** in isolation.
5. No behavior change to completed phases — ✅ entire suite green (all prior-phase tests included); `ruff check .` + `pyright` clean.
6. Production stale row — post-deploy re-sync verification (live infra, no repo work; no live changes made — deploy of `main` is the owner's step).
- **Documentation:** README.md reviewed in full — every env default cross-checked against `app/config.py`, page routes against `app/main.py`, scripts/Containerfile/compose flags, `.env.example`; the phase-133 PDF bullet added in the working tree is accurate. AGENTS.md verified consistent (all referenced files/commands exist). **No changes needed.**
- **Next pending phase:** `134_sync_cancel`.
1358 lines
53 KiB
Python
1358 lines
53 KiB
Python
"""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 "<empty>")
|
|
|
|
|
|
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 "<empty>")
|
|
|
|
|
|
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 "<empty>")
|
|
|
|
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=<N>#img=<index>`` 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
|
|
``<illustrator:StartupProfile>`` 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 (``<tag>``, ``</tag>``, ``<tag attr=`` — the
|
|
regex covers them all, including the PDF-structure ``<</Key…``
|
|
noise)."""
|
|
tags = [f"<xmp:Prop{i}>value</xmp:Prop{i}>" 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"<example:{i}>text</example:{i}>" 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"<example:{i}>text</example:{i}>" 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 "<illustrator:StartupProfile>" 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",
|
|
]
|