ducoterra
5e87622b30
All verification complete — everything is 100% green. Final report: **Phase 137 final verification pass — all tasks already complete, all criteria verified** - **Full suite:** `uv run pytest --cov=app --cov-report=term-missing` → 3272 passed, exit 0; `app/` coverage **99%** (>90% ✓) - **Phase E2E in isolation:** `uv run pytest tests/e2e/test_pdf_image_context.py -v --no-cov` → **4 passed** (forward context, display line, standalone parity, both-blocks markdown case) - **Pre-existing suites in isolation:** `test_pdf_support.py` + `test_image_anchor_context.py` + `test_image_documents.py` → **14 passed**; ripple-edited `test_garbage_pdf.py` → **3 passed** - **Lint/types:** `uv run ruff check .` → all passed; `uv run pyright` → 0 errors **Completion criteria:** (1) forward-path context in vision msg + stored `image_context`/summary — ✓ (unit builder/seam + integration fake-LLM + E2E); (2) both blocks document-first + standalone describe byte-identical — ✓ (combiner + standalone pins); (3) backfill with context + zero-vision idempotent re-sync + parent-gone origin-only — ✓ (6 integration tests); (4) suggestion block leads with `From <source>/<pdf> (page N)` — ✓ (renderer units + E2E); (5) all gates green — ✓ above; (6) no behavior change in phases 122/125/126 + sync `detail` shape untouched — ✓ (verified in code: new counter is log-only); (7) ruff+pyright — ✓; (8) commit — left to the harness per pipeline rules (working tree untouched by me). **Documentation:** README.md audited in full — the new phase-137 bullet verified against code (6 000-char cap, log-only `image_context_backfilled=N`, `<document>`-body locator placement), all env tables/defaults cross-checked vs `app/config.py` and `.env.example` — consistent, no changes needed. AGENTS.md: every rule/command verified — consistent, no changes needed. **Notable deviations (pre-existing, documented in task-05 report, forced by design):** `test_pdf_support.py`/`test_garbage_pdf.py` got minimal ripple edits (the "UNEDITED" pin was unsatisfiable once describe calls carry context by design); the unit `test_image_anchor_context.py` CONTEXT_PREFIX pin generalized per ASSUMPTION A5; one extra mock line (`_seed_context_line` accepts `From …`). **Next pending phase:** none — `todo/` holds only this phase.
1900 lines
78 KiB
Python
1900 lines
78 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), the per-page text map (phase 137 task 01 —
|
|
``extract_pdf_page_texts``, the document-context raw material),
|
|
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 re
|
|
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 (
|
|
PDF_IMAGE_CONTEXT_MAX_CHARS,
|
|
_combine_image_contexts,
|
|
_pdf_image_context_block,
|
|
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_page_texts,
|
|
extract_pdf_text,
|
|
is_usable_pdf_text,
|
|
render_pdf_pages,
|
|
)
|
|
from app.rag.retriever import TRUNCATION_MARKER
|
|
from app.rag.summarizer import DESCRIBE_PROMPT
|
|
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 enriched content (phase 137, task 02: the
|
|
document-context block FIRST, then the description — the summary
|
|
chunk mirrors it, the phase-122 convention)."""
|
|
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)
|
|
|
|
# Phase 137 (task 02): the document context — the origin line +
|
|
# page 2's text — is the stored ``image_context`` and the
|
|
# ``content``/``summary`` enrichment (context first, then the
|
|
# description — the text the embedder indexes carries it).
|
|
expected_context = (
|
|
f"Extracted from {root.name}/report.pdf (page 2):\nText page two."
|
|
)
|
|
expected_content = f"{expected_context}\n\n{DESCRIPTION}"
|
|
|
|
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.image_context == expected_context
|
|
assert img.content == expected_content and img.summary == expected_content
|
|
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 == expected_content
|
|
|
|
# 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")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Phase 137 (task 02) — the embedded image's document context
|
|
# ---------------------------------------------------------------------------
|
|
|
|
#: The exact generalized context prefix phase 137 (task 02, ASSUMPTION
|
|
#: A5) appends to the locked prompt (pinned here AND in the
|
|
#: summarizer's code) — it covers BOTH block kinds (the provenance's
|
|
#: ``Extracted from …`` document context and the phase-125
|
|
#: ``Referenced in …`` markdown references).
|
|
NEUTRAL_CONTEXT_PREFIX = (
|
|
"\n\nThis image is part of a document. The following context "
|
|
"describes where it comes from and what it is used for:\n"
|
|
)
|
|
|
|
|
|
class _CapturingVisionLLM(FakeEmbedder):
|
|
"""The task-02 capturing vision client: the CHAT model answers
|
|
every multimodal describe call with the fixed :data:`DESCRIPTION`
|
|
and records the message's TEXT PART (``text_parts`` — the context's
|
|
assertion surface: the locked prompt + the prefix + the context);
|
|
text (``lite``) calls keep the ``FakeEmbedder`` behaviour."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.text_parts: list[str] = []
|
|
|
|
async def chat(self, messages, model=None):
|
|
self.chat_calls.append(list(messages))
|
|
user = next((m["content"] for m in messages if m.get("role") == "user"), "")
|
|
if isinstance(user, list):
|
|
self.text_parts.extend(p["text"] for p in user if p.get("type") == "text")
|
|
return DESCRIPTION
|
|
first = user.split()
|
|
return "Summary of " + (first[0] if first else "<empty>")
|
|
|
|
|
|
# --- the block builder — ``_pdf_image_context_block`` -------------------
|
|
|
|
|
|
def test_pdf_image_context_block_header_and_page_text() -> None:
|
|
"""A page WITH text → the exact two-line shape: the origin line
|
|
(a stable single line — NO page-total), then the page's
|
|
(already-stripped) text."""
|
|
assert _pdf_image_context_block("src", "report.pdf", 2, {2: "The page text."}) == (
|
|
"Extracted from src/report.pdf (page 2):\nThe page text."
|
|
)
|
|
# The value is stripped by the builder too (defensive: a caller's
|
|
# unstripped map entry cannot leak trailing newlines into the
|
|
# block).
|
|
assert _pdf_image_context_block("s", "r.pdf", 1, {1: " Padded. "}) == (
|
|
"Extracted from s/r.pdf (page 1):\nPadded."
|
|
)
|
|
|
|
|
|
def test_pdf_image_context_block_caps_page_text_at_the_cap() -> None:
|
|
"""ASSUMPTION A3's cap: a ~7000-char page is cut EXACTLY at
|
|
:data:`PDF_IMAGE_CONTEXT_MAX_CHARS` (6000), with the shared
|
|
:data:`TRUNCATION_MARKER` on its own line (the summarizer's
|
|
existing truncation convention); a page that fits exactly at the
|
|
cap is NOT truncated."""
|
|
page_text = "w" * 7000
|
|
assert _pdf_image_context_block("src", "report.pdf", 1, {1: page_text}) == (
|
|
"Extracted from src/report.pdf (page 1):\n"
|
|
+ "w" * PDF_IMAGE_CONTEXT_MAX_CHARS
|
|
+ "\n"
|
|
+ TRUNCATION_MARKER
|
|
)
|
|
exact = "x" * PDF_IMAGE_CONTEXT_MAX_CHARS
|
|
assert _pdf_image_context_block("s", "r.pdf", 1, {1: exact}) == (
|
|
f"Extracted from s/r.pdf (page 1):\n{exact}"
|
|
)
|
|
|
|
|
|
def test_pdf_image_context_block_textless_page_is_origin_line_only() -> None:
|
|
"""A text-less page (no key in the map — and a whitespace-only
|
|
value) → the origin line ALONE: the provenance never depends on
|
|
the page having text (the owner's core complaint)."""
|
|
assert _pdf_image_context_block("s", "r.pdf", 3, {}) == "Extracted from s/r.pdf (page 3):"
|
|
assert _pdf_image_context_block("s", "r.pdf", 3, {3: " \n\t "}) == (
|
|
"Extracted from s/r.pdf (page 3):"
|
|
)
|
|
|
|
|
|
def test_pdf_image_context_block_unusable_layer_is_origin_line_only() -> None:
|
|
"""ASSUMPTION A7: the phase-133 gate failed (``text_usable=False``)
|
|
→ the "page text" is the pre-126 soup and is dropped — the origin
|
|
line only, even with text present."""
|
|
assert _pdf_image_context_block(
|
|
"s", "r.pdf", 1, {1: "<xmp:garbage>value</xmp:garbage>"}, text_usable=False
|
|
) == "Extracted from s/r.pdf (page 1):"
|
|
|
|
|
|
# --- the combiner — ``_combine_image_contexts`` -------------------------
|
|
|
|
|
|
def test_combine_image_contexts_doc_only() -> None:
|
|
"""No markdown reference (``None`` or empty) → the doc block
|
|
alone (the provenance is never empty)."""
|
|
block = "Extracted from s/r.pdf (page 2):\nText."
|
|
assert _combine_image_contexts(block, None) == block
|
|
assert _combine_image_contexts(block, "") == block
|
|
|
|
|
|
def test_combine_image_contexts_document_first_markdown_second() -> None:
|
|
"""The exact combined shape (A2): the doc block FIRST, a blank
|
|
line, then the markdown block(s) — the markdown block is
|
|
BYTE-IDENTICAL inside the combined value (the phase-125 shape is
|
|
never lost) and precedes nothing (provenance before purpose)."""
|
|
block = "Extracted from s/r.pdf (page 2):\nText."
|
|
md = 'Referenced in s/guide.md:\n- Alt text: "pic"\n Context: why'
|
|
assert _combine_image_contexts(block, md) == f"{block}\n\n{md}"
|
|
assert md in _combine_image_contexts(block, md), "the markdown block survives verbatim"
|
|
|
|
|
|
# --- the forward path — the capturing fake LLM through ``import_sources`` ---
|
|
|
|
|
|
def test_pdf_embedded_image_vision_call_carries_the_document_context(
|
|
db, tmp_path: Path
|
|
) -> None:
|
|
"""THE forward path (the task's core proof): a PDF with page text +
|
|
one embedded image → the vision message's text part is
|
|
``DESCRIBE_PROMPT`` + the (now-neutral) prefix + the exact
|
|
document-context block (the prefix bytes asserted explicitly), the
|
|
stored ``image_context`` is the SAME block, and
|
|
``content``/``summary`` are ``<block>\\n\\n<description>`` — the
|
|
text the embedder indexes carries the context. A second (unchanged)
|
|
sync re-describes NOTHING (the phase-126 digest skip is
|
|
untouched)."""
|
|
root = tmp_path / "pdfctx"
|
|
root.mkdir()
|
|
png = _png_bytes(40, 30)
|
|
_build_pdf(root / "report.pdf", [("Text page one.", []), ("Text page two.", [png])])
|
|
llm = _pdf_llm(tmp_path, _CapturingVisionLLM)
|
|
try:
|
|
summary = asyncio.run(import_sources([root], llm, session=db))
|
|
assert (summary.files, summary.added, summary.images_failed) == (1, 2, 0)
|
|
|
|
expected_block = f"Extracted from {root.name}/report.pdf (page 2):\nText page two."
|
|
# (a) the vision model received the context — the locked prompt
|
|
# + the exact generalized prefix + the block.
|
|
assert llm.text_parts == [DESCRIBE_PROMPT + NEUTRAL_CONTEXT_PREFIX + expected_block]
|
|
assert llm.text_parts[0].startswith(DESCRIBE_PROMPT), "the locked prompt is untouched"
|
|
# (b) the stored image_context IS the same combined value.
|
|
img = _doc(db, root.name, "report.pdf#page=2#img=0.png")
|
|
assert img.image_context == expected_block
|
|
# (c) content/summary carry it too — context first, then the
|
|
# description (the embedder indexes the same text).
|
|
expected_content = f"{expected_block}\n\n{DESCRIPTION}"
|
|
assert img.content == expected_content
|
|
assert img.summary == expected_content
|
|
chunks = db.scalars(select(Chunk).where(Chunk.document_id == img.id)).all()
|
|
packed = "\n".join(expected_content.split("\n\n"))
|
|
assert [c.content for c in chunks if not c.is_summary] == [packed]
|
|
assert [c.content for c in chunks if c.is_summary] == [expected_content]
|
|
|
|
# The unchanged-image digest skip is untouched: a re-sync of the
|
|
# same PDF re-describes nothing (no new chat call at all — the
|
|
# unchanged path skips the image pass entirely).
|
|
calls1 = len(llm.chat_calls)
|
|
s2 = asyncio.run(import_sources([root], llm, session=db))
|
|
assert (s2.added, s2.updated, s2.unchanged) == (0, 0, 1)
|
|
assert len(llm.chat_calls) == calls1, "unchanged: no re-describe"
|
|
finally:
|
|
_cleanup_source(db, root.name)
|
|
|
|
|
|
def test_pdf_embedded_image_with_markdown_reference_carries_both_blocks(
|
|
db, tmp_path: Path
|
|
) -> None:
|
|
"""A2's combining at the real seam: a same-source markdown
|
|
references the image by its synthetic path → the combined context
|
|
is the DOCUMENT block FIRST, a blank line, then the phase-125
|
|
``Referenced in`` block — the vision message carries exactly that
|
|
(generalized prefix + the combined value), and the stored
|
|
``image_context``/``content``/``summary`` do too (exact-string
|
|
pin)."""
|
|
root = tmp_path / "pdfboth"
|
|
root.mkdir()
|
|
(root / "guide.md").write_text(
|
|
"The VLAN map explains the subnet layout.\n"
|
|
"\n"
|
|
"\n"
|
|
"\n"
|
|
"Only expand the lab subnet after review.\n",
|
|
encoding="utf-8",
|
|
)
|
|
png = _png_bytes(40, 30)
|
|
_build_pdf(root / "report.pdf", [("Text page one.", []), ("Text page two.", [png])])
|
|
llm = _pdf_llm(tmp_path, _CapturingVisionLLM)
|
|
try:
|
|
summary = asyncio.run(import_sources([root], llm, session=db))
|
|
assert (summary.files, summary.added, summary.images_failed) == (2, 3, 0)
|
|
|
|
doc_block = f"Extracted from {root.name}/report.pdf (page 2):\nText page two."
|
|
md_block = (
|
|
f"Referenced in {root.name}/guide.md:\n"
|
|
'- Alt text: "VLAN map"\n'
|
|
" Context: The VLAN map explains the subnet layout.\n"
|
|
"\n"
|
|
"Only expand the lab subnet after review."
|
|
)
|
|
combined = f"{doc_block}\n\n{md_block}"
|
|
assert llm.text_parts == [DESCRIBE_PROMPT + NEUTRAL_CONTEXT_PREFIX + combined]
|
|
img = _doc(db, root.name, "report.pdf#page=2#img=0.png")
|
|
assert img.image_context == combined
|
|
expected_content = f"{combined}\n\n{DESCRIPTION}"
|
|
assert img.content == expected_content
|
|
assert img.summary == expected_content
|
|
finally:
|
|
_cleanup_source(db, root.name)
|
|
|
|
|
|
def test_pdf_embedded_image_textless_page_still_carries_the_origin_line(
|
|
db, tmp_path: Path
|
|
) -> None:
|
|
"""The owner's core complaint, pinned at the real seam: an image
|
|
on a TEXT-LESS page (page 1 carries text, so the layer is usable —
|
|
this is the text-less-PAGE case, not the garbage-LAYER case) → the
|
|
block is the origin line ONLY, and it is STILL stored + enriched
|
|
(the provenance never depends on the page having text)."""
|
|
root = tmp_path / "pdftextless"
|
|
root.mkdir()
|
|
png = _png_bytes(40, 30)
|
|
_build_pdf(root / "report.pdf", [("Text page one.", []), ("", [png])])
|
|
llm = _pdf_llm(tmp_path, _CapturingVisionLLM)
|
|
try:
|
|
summary = asyncio.run(import_sources([root], llm, session=db))
|
|
assert (summary.files, summary.added, summary.images_failed) == (1, 2, 0)
|
|
expected_block = f"Extracted from {root.name}/report.pdf (page 2):"
|
|
assert llm.text_parts == [DESCRIBE_PROMPT + NEUTRAL_CONTEXT_PREFIX + expected_block]
|
|
img = _doc(db, root.name, "report.pdf#page=2#img=0.png")
|
|
assert img.image_context == expected_block
|
|
assert img.content == f"{expected_block}\n\n{DESCRIPTION}"
|
|
assert img.summary == img.content
|
|
finally:
|
|
_cleanup_source(db, root.name)
|
|
|
|
|
|
def test_pdf_embedded_image_garbage_layer_gets_origin_line_only(
|
|
db, tmp_path: Path
|
|
) -> None:
|
|
"""ASSUMPTION A7 at the real seam: a PDF whose text layer fails the
|
|
phase-133 gate (the XML-soup class) → the pass's gate run yields
|
|
``text_usable=False`` → the image's block is the origin line ONLY —
|
|
the garbage "page text" never reaches the vision model (asserted on
|
|
the captured text part). The document itself takes the
|
|
page-screenshot fallback (context-free calls — the bare locked
|
|
prompt)."""
|
|
root = tmp_path / "pdfgarbage"
|
|
root.mkdir()
|
|
doc = fitz.open()
|
|
page = doc.new_page()
|
|
for i in range(10):
|
|
page.insert_text((72, 72 + i * 14), f"<foo:bar>{i}</foo:bar>")
|
|
pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, 40, 30), 0)
|
|
pix.clear_with(200)
|
|
page.insert_image(fitz.Rect(72, 220, 112, 250), stream=pix.tobytes("png"))
|
|
doc.save(root / "report.pdf")
|
|
doc.close()
|
|
llm = _pdf_llm(tmp_path, _CapturingVisionLLM)
|
|
try:
|
|
summary = asyncio.run(import_sources([root], llm, session=db))
|
|
assert summary.pdfs_screenshotted == 1 # the gate fired on the doc too
|
|
assert (summary.added, summary.images_failed) == (2, 0)
|
|
expected_block = f"Extracted from {root.name}/report.pdf (page 1):"
|
|
# The fallback's page-screenshot call is context-free (the bare
|
|
# locked prompt); the IMAGE's call carries the origin-line-only
|
|
# block (A7 — the soup never reaches the model as "text").
|
|
assert llm.text_parts == [
|
|
DESCRIBE_PROMPT,
|
|
DESCRIBE_PROMPT + NEUTRAL_CONTEXT_PREFIX + expected_block,
|
|
]
|
|
img = _doc(db, root.name, "report.pdf#page=1#img=0.png")
|
|
assert img.image_context == expected_block
|
|
assert img.content == f"{expected_block}\n\n{DESCRIPTION}"
|
|
finally:
|
|
_cleanup_source(db, root.name)
|
|
|
|
|
|
# --- the regression pins — the standalone path stays byte-identical ----
|
|
|
|
|
|
def test_standalone_image_vision_call_stays_byte_identical(db, tmp_path: Path) -> None:
|
|
"""The phase-122 regression pin: a standalone image (no markdown
|
|
reference, no PDF) → the describe message's text part is the BARE
|
|
``DESCRIBE_PROMPT`` — byte-identical to pre-phase-137 (the
|
|
generalized prefix touches ONLY the context-carrying message),
|
|
``image_context`` stays NULL, and the content is the bare
|
|
description."""
|
|
root = tmp_path / "pdfstandalone"
|
|
root.mkdir()
|
|
(root / "notes.md").write_text("# Notes\n\nNo references here.\n", encoding="utf-8")
|
|
(root / "pic.png").write_bytes(_png_bytes(40, 30))
|
|
llm = _pdf_llm(tmp_path, _CapturingVisionLLM)
|
|
try:
|
|
summary = asyncio.run(import_sources([root], llm, session=db))
|
|
assert (summary.files, summary.added, summary.images_failed) == (2, 2, 0)
|
|
assert llm.text_parts == [DESCRIBE_PROMPT], "standalone: the bare prompt"
|
|
img = _doc(db, root.name, "pic.png")
|
|
assert img.image_context is None, "standalone: the context stays NULL"
|
|
assert img.content == DESCRIPTION and img.summary == DESCRIPTION
|
|
finally:
|
|
_cleanup_source(db, root.name)
|
|
|
|
|
|
def test_standalone_markdown_referenced_image_uses_the_neutral_prefix(
|
|
db, tmp_path: Path
|
|
) -> None:
|
|
"""The phase-125 path under the generalized prefix: a standalone
|
|
image with a markdown reference → the message carries the markdown
|
|
block with the (now-neutral) prefix, and the STORED values are
|
|
byte-identical to the phase-125 shape (only the prompt's prefix
|
|
wording moved — the stored data did not)."""
|
|
root = tmp_path / "pdfstandmd"
|
|
root.mkdir()
|
|
(root / "guide.md").write_text(
|
|
"Why this diagram exists.\n\n\n\nWhat follows.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(root / "pic.png").write_bytes(_png_bytes(40, 30))
|
|
llm = _pdf_llm(tmp_path, _CapturingVisionLLM)
|
|
try:
|
|
asyncio.run(import_sources([root], llm, session=db))
|
|
md_block = (
|
|
f"Referenced in {root.name}/guide.md:\n"
|
|
'- Alt text: "The red square"\n'
|
|
" Context: Why this diagram exists.\n"
|
|
"\n"
|
|
"What follows."
|
|
)
|
|
assert llm.text_parts == [DESCRIBE_PROMPT + NEUTRAL_CONTEXT_PREFIX + md_block]
|
|
img = _doc(db, root.name, "pic.png")
|
|
assert img.image_context == md_block, "the stored value is the unchanged data"
|
|
assert img.content == f"{md_block}\n\n{DESCRIPTION}"
|
|
finally:
|
|
_cleanup_source(db, root.name)
|
|
|
|
|
|
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",
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# extract_pdf_page_texts (phase 137 task 01 — the per-page text map)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
#: The owner's real multi-page fixture (untracked — ``test_sources/``
|
|
#: is gitignored; unpacked from ``test_sources/pdf_tests.tar``): the
|
|
#: D&D player's handbook, 293 pages.
|
|
HANDBOOK = Path(__file__).resolve().parents[2] / "test_sources" / "players_handbook.pdf"
|
|
|
|
#: The owner's single-page test PDF (same provenance).
|
|
HORSE_PDF = Path(__file__).resolve().parents[2] / "test_sources" / "pdf_horse_test.pdf"
|
|
|
|
@pytest.fixture(scope="module")
|
|
def handbook_texts() -> tuple[str, dict[int, str]]:
|
|
"""The handbook's ``extract_pdf_text`` output AND page map, each
|
|
captured ONCE (module-scoped — a full 293-page extraction is ~0.5 s
|
|
and two tests need the same pair)."""
|
|
return extract_pdf_text(HANDBOOK), extract_pdf_page_texts(HANDBOOK)
|
|
|
|
|
|
def _parse_page_blocks(joined: str) -> dict[int, str]:
|
|
"""Parse :func:`extract_pdf_text`'s output back into
|
|
``{1-indexed page: block body}`` (the page-for-page comparison's
|
|
reference: the blocks are ``---PAGE-N---\n<body>`` joined with
|
|
``"\\n\\n"``, the body is stripped — so the body ends 2 chars
|
|
before the next separator)."""
|
|
matches = [m for m in re.finditer(r"---PAGE-(\d+)---\n", joined)]
|
|
blocks: dict[int, str] = {}
|
|
for i, m in enumerate(matches):
|
|
end = matches[i + 1].start() - 2 if i + 1 < len(matches) else len(joined)
|
|
blocks[int(m.group(1))] = joined[m.end():end]
|
|
return blocks
|
|
|
|
|
|
@pytest.mark.skipif(not HANDBOOK.is_file(), reason="handbook fixture missing")
|
|
def test_handbook_page_texts_agree_with_extract_pdf_text_page_for_page(
|
|
handbook_texts: tuple[str, dict[int, str]],
|
|
) -> None:
|
|
"""THE page-for-page agreement (the task's core proof): parsing
|
|
``extract_pdf_text``'s ``---PAGE-N---`` blocks back out and
|
|
comparing with ``extract_pdf_page_texts`` — the two functions must
|
|
agree EXACTLY (same keys, same stripped values)."""
|
|
text, page_texts = handbook_texts
|
|
assert page_texts == _parse_page_blocks(text)
|
|
# 293 or fewer entries, 1-indexed keys: the handbook has 293 pages
|
|
# and two of them (the cover, page 1, and page 280) are image-only
|
|
# → 291 entries. The keys are the REAL 1-indexed page numbers, so
|
|
# the first key is 2 (the cover carries no text — no key).
|
|
assert 0 < len(page_texts) <= 293
|
|
assert len(page_texts) == 291
|
|
assert min(page_texts) == 2
|
|
assert max(page_texts) == 293
|
|
assert all(isinstance(k, int) and 1 <= k <= 293 for k in page_texts)
|
|
# every value is non-empty and stripped
|
|
for value in page_texts.values():
|
|
assert isinstance(value, str) and value == value.strip() and value
|
|
|
|
|
|
@pytest.mark.skipif(not HANDBOOK.is_file(), reason="handbook fixture missing")
|
|
def test_handbook_page_texts_spot_check_known_page_body(
|
|
handbook_texts: tuple[str, dict[int, str]],
|
|
) -> None:
|
|
"""Spot-check against the CURRENT (pre-refactor) output: page 2's
|
|
body (the first text page — the Contents list) and page 293's body
|
|
(the index tail) carry the known substrings captured from the
|
|
phase-126 output."""
|
|
text, page_texts = handbook_texts
|
|
assert text.startswith(f"{PAGE_SEPARATOR.format(page=2)}\nC o n t e n t s")
|
|
assert "P r e f a c e" in page_texts[2] # the TOC renders letter-spaced
|
|
assert "School of Abjuration (wizard)" in page_texts[293]
|
|
# the page-2 map value is byte-identical to the parsed block body
|
|
assert page_texts[2] == _parse_page_blocks(text)[2]
|
|
|
|
|
|
@pytest.mark.skipif(not HORSE_PDF.is_file(), reason="horse fixture missing")
|
|
def test_horse_page_texts_match_extract_pdf_text_whatever_it_is() -> None:
|
|
"""The one-page fixture: asserted AGAINST ``extract_pdf_text``'s
|
|
output for the same file, whatever it is — a text page means
|
|
exactly one entry (key 1), a text-less page means ``{}``."""
|
|
text = extract_pdf_text(HORSE_PDF)
|
|
page_texts = extract_pdf_page_texts(HORSE_PDF)
|
|
parsed = _parse_page_blocks(text)
|
|
assert page_texts == parsed
|
|
if not parsed:
|
|
assert page_texts == {}
|
|
else:
|
|
assert list(page_texts) == [1] # exactly one entry, 1-indexed
|
|
assert "Below is a picture of my horse" in page_texts[1]
|
|
assert page_texts[1] == page_texts[1].strip()
|
|
|
|
|
|
def test_missing_file_page_texts_returns_empty_dict_no_raise(
|
|
tmp_path: Path, caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Fail-soft: an unreadable file → ``{}`` (no raise, one warning —
|
|
the ``extract_pdf_text`` open contract, the ``pdf:`` house prefix)."""
|
|
with caplog.at_level(logging.WARNING, logger="app.rag.pdf"):
|
|
assert extract_pdf_page_texts(tmp_path / "nope.pdf") == {}
|
|
warnings = [
|
|
r for r in caplog.records if "could not open" in r.getMessage()
|
|
]
|
|
assert len(warnings) == 1 and warnings[0].levelno == logging.WARNING
|
|
|
|
|
|
def test_garbage_file_page_texts_returns_empty_dict_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_page_texts(bad) == {}
|
|
|
|
|
|
def test_page_texts_broken_page_is_skipped(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path,
|
|
) -> None:
|
|
"""The shared helper's per-page failure branch, through the NEW
|
|
call site: a page whose get_text raises is logged/skipped (no key);
|
|
the other pages keep their (stripped) text."""
|
|
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. ")]),
|
|
)
|
|
assert extract_pdf_page_texts(path) == {2: "Fine text."}
|
|
|
|
|
|
# --- the phase-126 regression pin (the shared-iteration refactor) -------
|
|
# Captured from the PRE-refactor output NOW (phase 137 task 01): the
|
|
# pins assert the separator structure + length + sha256 — byte-identity
|
|
# without hardcoding a 300-page string.
|
|
|
|
_HANDBOOK_SHA256 = "a0832fc585d4ecf41579198acb3a45761a0f51b0d60d5a282ea8f4d7a84cbeb6"
|
|
_HANDBOOK_LEN = 1_268_411
|
|
_HANDBOOK_BLOCKS = 291 # 293 pages minus the two image-only pages (1, 280)
|
|
|
|
_HORSE_SHA256 = "ab998617803b71fbd814000f66d0dde83c2b6eb7f679f82b3da7ca06b4d2ecca"
|
|
_HORSE_EXACT = (
|
|
f"{PAGE_SEPARATOR.format(page=1)}\nBelow is a picture of my horse. "
|
|
"It may look like a cat but it\u2019s my horse. I\u2019ve just always "
|
|
"called it that."
|
|
)
|
|
|
|
|
|
@pytest.mark.skipif(not HANDBOOK.is_file(), reason="handbook fixture missing")
|
|
def test_extract_pdf_text_regression_pin_handbook() -> None:
|
|
"""``extract_pdf_text`` on the 293-page handbook is BYTE-IDENTICAL
|
|
to the pre-refactor (phase-126) output: same separator structure
|
|
(291 blocks, ``---PAGE-2---`` first — the cover is image-only —
|
|
``---PAGE-293---`` last, ascending), same length, same sha256."""
|
|
text = extract_pdf_text(HANDBOOK)
|
|
assert len(text) == _HANDBOOK_LEN
|
|
assert hashlib.sha256(text.encode("utf-8")).hexdigest() == _HANDBOOK_SHA256
|
|
headers = [line for line in text.splitlines() if line.startswith("---PAGE-")]
|
|
assert len(headers) == _HANDBOOK_BLOCKS
|
|
assert headers[0] == PAGE_SEPARATOR.format(page=2)
|
|
assert headers[-1] == PAGE_SEPARATOR.format(page=293)
|
|
assert [int(h[len("---PAGE-"):-3]) for h in headers] == sorted(
|
|
int(h[len("---PAGE-"):-3]) for h in headers
|
|
)
|
|
# the known substring from the first text page (captured pre-refactor)
|
|
assert text.startswith(f"{PAGE_SEPARATOR.format(page=2)}\nC o n t e n t s")
|
|
|
|
|
|
@pytest.mark.skipif(not HORSE_PDF.is_file(), reason="horse fixture missing")
|
|
def test_extract_pdf_text_regression_pin_horse() -> None:
|
|
"""The one-page fixture, exact byte pin (short enough to pin in
|
|
full): the pre-refactor string, character for character."""
|
|
text = extract_pdf_text(HORSE_PDF)
|
|
assert text == _HORSE_EXACT
|
|
assert hashlib.sha256(text.encode("utf-8")).hexdigest() == _HORSE_SHA256
|