ducoterra
e0f11e7aa8
All gates verified green — no code changes were needed on this pass. Final report: ## Phase 126 — PDF support: final verification pass **Verified (all tasks 01–05 already complete in working tree):** - `pymupdf` dep in `pyproject.toml`/`uv.lock`; `.env.example` documents the `pdf` token - `app/rag/pdf.py` (text + image extraction, MIME mapping), page-aware `_chunk_pdf`, importer `_index_pdf_file` + embedded-image pass, archives covered via extension walk **Test / lint / coverage results:** - `uv run pytest` → 2997 passed - `uv run pytest tests/e2e/test_pdf_support.py -v --no-cov` (isolation) → 6 passed - `uv run pytest --cov=app --cov-report=term-missing` → TOTAL **99%** (>90% gate), 2997 passed - `uv run ruff check . && uv run pyright` → clean (0 errors) **Completion criteria:** 1. PDF → doc with text/chunks/embedding/summary — ✅ (integration + E2E Sources-page tests) 2. `BOR_IMAGES=true`: embedded images → standalone image docs — ✅ (integration + E2E chat-chip test) 3. `BOR_IMAGES=false`: images skipped, text indexed — ✅ (both suites) 4. PDF renders as extracted text in viewer — ✅ (E2E) 5. Test/coverage/lint gates — ✅ (above) 6. Commit + phase move — harness handles (per protocol, no `git` on this pass) **Decisions:** none new; inline `[Embedded image: …]` placeholders were a documented task-04 deviation (simplified spec wins) — not a defect. No defects found; repo left functional with changes in the working tree. **Next pending phase:** none — `todo/` is empty once 126 is moved.
1072 lines
41 KiB
Python
1072 lines
41 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
|
|
from collections.abc import 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 (
|
|
PAGE_SEPARATOR,
|
|
PdfImage,
|
|
_mime_from_image_info,
|
|
_page_image_entries,
|
|
extract_pdf_images,
|
|
extract_pdf_text,
|
|
)
|
|
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]) -> 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)
|
|
|
|
|
|
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_still_becomes_a_document(
|
|
db, tmp_path: Path, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
"""An image-only PDF (no extractable text) still becomes a document
|
|
— with empty content (the empty byte string's digest), no content
|
|
chunks, and a warning (the signal). The summary path still runs
|
|
(best-effort on the empty content)."""
|
|
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)
|
|
doc = _doc(db, "pdfempty", "pics.pdf")
|
|
assert doc.content == ""
|
|
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] == []
|
|
assert doc.summary is not None # best-effort on the empty content
|
|
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, "pdfempty")
|
|
|
|
|
|
def test_corrupt_pdf_file_becomes_an_empty_document(db, tmp_path: Path) -> None:
|
|
"""A .pdf file that is not a PDF at all: the parser's fail-soft
|
|
contract (no text, no images) — the document lands with empty
|
|
content and no chunks, and the sync stays green."""
|
|
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:
|
|
summary = asyncio.run(import_sources([root], llm, session=db))
|
|
assert (summary.files, summary.added, summary.errors) == (1, 1, 0)
|
|
doc = _doc(db, "pdfbad", "broken.pdf")
|
|
assert doc.content == "" and doc.title == "broken"
|
|
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] == []
|
|
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}]
|