"""Unit: phase 181 task 02 — an ``.svg`` indexes as a VECTOR image document.
The pure extractor (:mod:`app.rag.svg_text`) is pinned in
``tests/unit/test_svg_text_extraction.py``; this module pins what the
IMPORTER does with it (LOCKED A1/A2/A7):
* the vector branch of :func:`app.rag.importer._index_image_file` — an
``.svg`` lands ``is_image=True`` with ``image_path`` under
``image_dir``, its ``content`` is the file's OWN extracted text, its
``summary`` mirrors it (the phase-122 convention), and the vision seam
:func:`app.rag.importer._describe_or_skip` is NEVER called (the seam is
patched with a recording stand-in that would return a sentinel — the
empty recording list + the content assertion are the proof, and a
call-counting seam cannot silently succeed);
* the log-only ``svg_indexed`` counter and the ``import: svg indexed``
auditability line (AGENTS rule 10), including ``structural=True`` for
a drawing whose only words are the LOCKED A2 fallback line;
* the fail-soft (LOCKED A2's boundary): a file the extractor REFUSES
(over the byte cap, malformed XML) skips exactly like an undescribed
raster — ``images_failed``, a warning, no row, no orphan copy — while
a merely textless SVG does NOT skip;
* the phase-125 ``image_context`` enrichment prepends for an SVG exactly
as for a raster;
* the phase-181 fast-path kind guard (:func:`_row_kind_mismatch`): a row
imported as TEXT before ``svg`` joined ``BOR_IMAGE_EXTENSIONS`` is
converted on the NEXT sync (same bytes, same ``file_hash`` — and for an
ASCII file the text digest equals the byte digest too, which is why the
image branch's own content-digest skip now requires ``is_image``), and
a row already of the right kind — raster or vector — keeps the
phase-147 fast path byte-for-byte (no second seam call, no second
``svg_indexed``).
"""
from __future__ import annotations
import asyncio
import hashlib
import logging
from pathlib import Path
from typing import Any
import pytest
from sqlalchemy import select
import app.rag.importer as importer
from app.config import Settings
from app.models import Chunk, Document
from app.rag.importer import import_sources
from app.rag.svg_text import MAX_SVG_BYTES
from tests.fakes import FakeEmbedder
#: A diagram that carries words: ```` + ``<desc>`` + a ``<text>``
#: inside a group (document order is the extractor's contract).
SVG_TOPO = (
b'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 120 80">'
b"<title>Server rack topology "
b"Two switches uplink to the router. "
b'rack-1 '
b' "
)
#: The LOCKED A2 structural line for :data:`SVG_SHAPES_ONLY` — built only
#: from facts the file states (stem, shape census, viewBox).
SVG_SHAPES_ONLY_TEXT = (
'Vector diagram "gizmo" — 3 shapes (path×2, circle), viewBox 64×64'
)
#: Over :data:`MAX_SVG_BYTES`: the size guard refuses it BEFORE any parse.
SVG_OVER_CAP = b"" + b"\n" * (MAX_SVG_BYTES + 1) + b" "
#: Well-formed bytes, unparseable markup (the second refusal class).
SVG_MALFORMED = b'oops'
#: A real 1×1 PNG for the raster fast-path regression pins.
PNG_1X1 = b"\x89PNG\r\n\x1a\n" + b"\x00" * 24
#: What a describe seam would hand back if the vector branch ever reached
#: it — never a plausible diagram text, so it cannot pass unnoticed.
VISION_SENTINEL = "SHOULD NEVER BE DESCRIBED: a vision call ran for an SVG."
def _settings(**kwargs: Any) -> Settings:
"""Build Settings without reading a .env file (deterministic tests)."""
kwargs.setdefault("_env_file", None)
return Settings(**kwargs) # pyright: ignore[reportCallIssue] (kwarg exists at runtime)
def _image_llm(tmp_path: Path, **kwargs: Any) -> FakeEmbedder:
"""A fake LLM whose settings index images into *tmp_path*."""
kwargs.setdefault("_env_file", None)
kwargs.setdefault("images", True)
kwargs.setdefault("image_dir", str(tmp_path / "images"))
llm = FakeEmbedder()
llm.settings = Settings(**kwargs) # pyright: ignore[reportCallIssue]
return llm
def _record_description(monkeypatch: pytest.MonkeyPatch) -> list[str]:
"""Patch the vision seam with a recorder (never a real call): the
paths it is asked to describe, always answered with
:data:`VISION_SENTINEL`. An empty list + a non-sentinel ``content``
is the proof that the vector branch bypassed vision entirely."""
handed: list[str] = []
async def _fake(
llm: importer.Embedder,
*,
data: bytes,
source: str,
rel: str,
full_path: Path,
context: str | None = None,
) -> str | None:
handed.append(rel)
return VISION_SENTINEL
monkeypatch.setattr(importer, "_describe_or_skip", _fake)
return handed
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()
# --- the vector branch ------------------------------------------------------
def test_svg_indexes_as_an_image_document_with_no_vision_call(
db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
"""LOCKED A1: the ``.svg`` takes the image branch (``is_image``,
``image_path``, the ``<doc-id>.svg`` copy, the ``summary = content``
convention, one embedded content chunk + the ``is_summary`` chunk)
and its searchable text is the file's own ``<title>``/``<desc>``/
``<text>`` — with the vision seam never invoked. ``svg_indexed`` and
the per-file ``import: svg indexed`` line make the run auditable."""
root = tmp_path / "svgsrc"
root.mkdir()
(root / "topo.svg").write_bytes(SVG_TOPO)
llm = _image_llm(tmp_path)
handed = _record_description(monkeypatch)
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 summary.errors == 0
assert summary.formats == {"svg": 1} # counted under its image token
assert summary.svg_indexed == 1
assert handed == [], "an SVG must never reach the vision seam"
doc = _doc(db, root.name, "topo.svg")
assert doc.is_image is True
assert doc.title == "topo" # the non-markdown stem rule (shared with raster)
assert doc.content == SVG_TOPO_TEXT
assert doc.summary == doc.content # the phase-122 image convention
assert doc.image_context is None # standalone: the pre-phase row shape
assert doc.image_path is not None
copy = Path(doc.image_path)
assert copy.parent == Path(llm.settings.image_dir).expanduser()
assert copy.name == f"{doc.id}.svg" # row and copy agree on the id
assert copy.read_bytes() == SVG_TOPO # the exact bytes, unserved here
chunks = db.scalars(select(Chunk).where(Chunk.document_id == doc.id)).all()
content_chunks = [c for c in chunks if not c.is_summary]
assert [c.content for c in content_chunks] == [doc.content]
summary_chunks = [c for c in chunks if c.is_summary]
assert len(summary_chunks) == 1 and summary_chunks[0].position == -1
assert summary_chunks[0].content == doc.content
for c in chunks:
assert c.embedding is not None and len(c.embedding) == 768
indexed = [
r.getMessage() for r in caplog.records if "import: svg indexed" in r.getMessage()
]
assert indexed == [
f"import: svg indexed source={root.name} path=topo.svg "
f"structural=False chars={len(SVG_TOPO_TEXT)}"
]
line = next(
r.getMessage()
for r in caplog.records
if r.getMessage().startswith("import: summary files=")
)
assert "svg_indexed=1" in line and "images_failed=0" in line
finally:
_cleanup_source(db, root.name)
def test_textless_svg_indexes_with_the_structural_line(
db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
"""LOCKED A2: a diagram of pure ``<path>``s is NOT skipped — it is
indexed with the structural line built from the file (stem, shape
census, viewBox), counted like any other vector doc and logged with
``structural=True`` so the owner can tell it from a worded diagram."""
root = tmp_path / "shapeonly"
root.mkdir()
(root / "gizmo.svg").write_bytes(SVG_SHAPES_ONLY)
llm = _image_llm(tmp_path)
handed = _record_description(monkeypatch)
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 summary.svg_indexed == 1
assert handed == []
doc = _doc(db, root.name, "gizmo.svg")
assert doc.is_image is True
assert doc.content == SVG_SHAPES_ONLY_TEXT
assert doc.summary == doc.content
line = next(
r.getMessage() for r in caplog.records if "import: svg indexed" in r.getMessage()
)
assert "structural=True" in line
finally:
_cleanup_source(db, root.name)
@pytest.mark.parametrize(
("name", "payload"),
[
pytest.param("huge.svg", SVG_OVER_CAP, id="over-the-byte-cap"),
pytest.param("broken.svg", SVG_MALFORMED, id="malformed-xml"),
],
)
def test_unusable_svg_skips_the_doc_counts_and_logs(
db,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
name: str,
payload: bytes,
) -> None:
"""The fail-soft stays LOCKED A3-shaped for the vector case: a file
the extractor refuses skips the doc ENTIRELY — no row, no copy (not
even the image dir), ``images_failed`` + a warning naming the vector
extraction, the sync continues. This is the ONLY skip class the phase
keeps for an SVG (:func:`app.rag.svg_text.extract_svg_text`)."""
root = tmp_path / "badsvg"
root.mkdir()
(root / name).write_bytes(payload)
llm = _image_llm(tmp_path)
handed = _record_description(monkeypatch)
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, 0, 1)
assert summary.svg_indexed == 0
assert handed == []
assert db.scalar(select(Document).where(Document.source == root.name)) is None
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 "svg text extraction failed" in r.getMessage()
]
assert len(warnings) == 1 and warnings[0].levelno == logging.WARNING
assert "image description failed" not in caplog.text
line = next(
r.getMessage()
for r in caplog.records
if r.getMessage().startswith("import: summary files=")
)
assert "images_failed=1" in line and "svg_indexed=0" in line
finally:
_cleanup_source(db, root.name)
def test_markdown_reference_context_prepends_for_an_svg(
db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Phase 125's enrichment is kind-agnostic: a ````
reference in the same source still builds ``image_context`` (from the
file on disk, no vision call) and the doc's content becomes
``<context>\\n\\n<extracted text>`` — context first for retrieval, the
vector file's own words second."""
root = tmp_path / "svganchor"
root.mkdir()
(root / "guide.md").write_text(
"Why this diagram exists.\n\n\n\nWhat follows it.\n",
encoding="utf-8",
)
(root / "topo.svg").write_bytes(SVG_TOPO)
llm = _image_llm(tmp_path)
handed = _record_description(monkeypatch)
try:
summary = asyncio.run(import_sources([root], llm, session=db))
assert (summary.files, summary.added, summary.images_failed) == (2, 2, 0)
assert summary.svg_indexed == 1
assert handed == []
doc = _doc(db, root.name, "topo.svg")
assert doc.is_image is True
assert doc.image_context is not None
assert doc.image_context.startswith(f"Referenced in {root.name}/guide.md:")
assert 'Alt text: "The rack plan"' in doc.image_context
assert doc.content == f"{doc.image_context}\n\n{SVG_TOPO_TEXT}"
assert doc.summary == doc.content
finally:
_cleanup_source(db, root.name)
# --- the fast-path kind guard (the text → image migration) ------------------
def test_text_indexed_svg_row_is_converted_on_the_next_sync(
db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Design D4: the row an earlier run indexed as TEXT (``svg`` not yet
in the image set) is converted by the NEXT sync — same bytes, same
``file_hash`` — and the sync AFTER that is fast again.
The conversion needs BOTH halves of the guard: the fast path in
:func:`_index_file` must fall through, and the image branch's own
content-digest skip must require ``is_image`` — an ASCII file's text
digest equals its byte digest, so without it the row would report
``unchanged`` forever and never gain a copy (asserted below, because
it is the whole reason the guard exists)."""
root = tmp_path / "migrate"
root.mkdir()
(root / "topo.svg").write_bytes(SVG_TOPO)
handed = _record_description(monkeypatch)
# Stage A — the pre-phase state: images OFF, ``svg`` imported as text.
text_llm = _image_llm(
tmp_path,
images=False,
import_extensions=f"{_settings().import_extensions},svg",
)
try:
summary_a = asyncio.run(import_sources([root], text_llm, session=db))
assert (summary_a.files, summary_a.added, summary_a.svg_indexed) == (1, 1, 0)
before = _doc(db, root.name, "topo.svg")
assert before.is_image is False
assert before.image_path is None
assert before.content == SVG_TOPO.decode("utf-8") # the raw XML the owner saw
byte_digest = hashlib.sha256(SVG_TOPO).hexdigest()
assert before.file_hash == byte_digest
# The trap D4 closes: the text digest of a byte-clean file IS the
# byte digest, so no digest comparison can spot the mismatch.
assert before.content_hash == byte_digest
doc_id = before.id
# Stage B — the toggle on: the guard fires, the row converts.
image_llm = _image_llm(tmp_path)
summary_b = asyncio.run(import_sources([root], image_llm, session=db))
assert (summary_b.added, summary_b.updated, summary_b.unchanged) == (0, 1, 0)
assert (summary_b.images_failed, summary_b.svg_indexed) == (0, 1)
assert handed == [], "the conversion must not call vision either"
after = _doc(db, root.name, "topo.svg")
assert after.id == doc_id # converted in place — the id (and copy name) is stable
assert after.is_image is True
assert after.content == SVG_TOPO_TEXT
assert after.summary == after.content
assert after.file_hash == byte_digest # the bytes never changed
# The image branch stamps the BYTES digest as ``content_hash`` —
# the SAME value the text path had already stored, which is why
# that skip needed its own ``is_image`` guard.
assert after.content_hash == byte_digest
copy = Path(str(after.image_path))
assert copy == Path(image_llm.settings.image_dir).expanduser() / f"{doc_id}.svg"
assert copy.read_bytes() == SVG_TOPO
# The text chunks are replaced, not appended: one content chunk
# (the extracted text) + the is_summary mirror.
chunks = db.scalars(select(Chunk).where(Chunk.document_id == doc_id)).all()
assert [c.content for c in chunks if not c.is_summary] == [SVG_TOPO_TEXT]
assert [c.content for c in chunks if c.is_summary] == [SVG_TOPO_TEXT]
# Stage C — the guard fired once: the fast path resumes.
summary_c = asyncio.run(import_sources([root], image_llm, session=db))
assert (summary_c.added, summary_c.updated, summary_c.unchanged) == (0, 0, 1)
assert (summary_c.svg_indexed, summary_c.images_failed) == (0, 0)
assert handed == []
assert _doc(db, root.name, "topo.svg").is_image is True
finally:
_cleanup_source(db, root.name)
def test_correctly_classified_raster_row_still_takes_the_fast_path(
db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The phase-147 fast path survives the guard for every row whose
KIND already agrees: a raster image with a matching ``file_hash`` is
UNCHANGED on the next sync — the seam is not called a second time and
no vector counter moves."""
root = tmp_path / "rasterfast"
root.mkdir()
(root / "pic.png").write_bytes(PNG_1X1)
llm = _image_llm(tmp_path)
handed = _record_description(monkeypatch)
try:
first = asyncio.run(import_sources([root], llm, session=db))
assert (first.added, first.unchanged, first.svg_indexed) == (1, 0, 0)
assert handed == ["pic.png"] # the raster branch DID use the seam
doc = _doc(db, root.name, "pic.png")
assert doc.is_image is True
second = asyncio.run(import_sources([root], llm, session=db))
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
assert second.svg_indexed == 0
assert handed == ["pic.png"], "the fast path must not re-describe the image"
assert _doc(db, root.name, "pic.png").content == VISION_SENTINEL
finally:
_cleanup_source(db, root.name)
def test_correctly_classified_vector_row_still_takes_the_fast_path(
db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Same pin for the vector row: once converted (or freshly indexed)
it is UNCHANGED — no second extraction, no second ``svg_indexed``,
no second copy write."""
root = tmp_path / "vectorfast"
root.mkdir()
(root / "topo.svg").write_bytes(SVG_TOPO)
llm = _image_llm(tmp_path)
handed = _record_description(monkeypatch)
try:
first = asyncio.run(import_sources([root], llm, session=db))
assert (first.added, first.svg_indexed) == (1, 1)
copy = Path(str(_doc(db, root.name, "topo.svg").image_path))
copy.unlink() # the fast path must not re-create it either
second = asyncio.run(import_sources([root], llm, session=db))
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
assert (second.svg_indexed, second.images_failed) == (0, 0)
assert handed == []
assert not copy.exists()
finally:
_cleanup_source(db, root.name)
def test_row_kind_mismatch_table() -> None:
"""The guard's whole truth table (pure — no DB, no filesystem):
it fires ONLY for a TEXT row whose file the CURRENT settings would
index as an image. The reverse direction (an ``is_image`` row while
the toggle is off, or a plain text file) stays on the fast path —
the phase-122 prune guard owns the off state (LOCKED A7)."""
svg = Path("/src/diagrams/topo.svg")
md = Path("/src/notes.md")
text_row = Document(source="s", path="topo.svg", is_image=False)
image_row = Document(source="s", path="topo.svg", is_image=True)
def _settings_(*, images: bool) -> Settings:
return _settings(images=images)
# The migration: the toggle is on, the file is an image, the row is not.
assert importer._row_kind_mismatch(svg, text_row, _settings_(images=True)) is True
# Already an image doc → nothing to migrate.
assert importer._row_kind_mismatch(svg, image_row, _settings_(images=True)) is False
# Toggle off → the walk never admits the file, the prune guard protects
# the row: never force the image path (LOCKED A7).
assert importer._row_kind_mismatch(svg, text_row, _settings_(images=False)) is False
assert importer._row_kind_mismatch(svg, image_row, _settings_(images=False)) is False
# A file outside the image set is a text file either way.
assert importer._row_kind_mismatch(md, text_row, _settings_(images=True)) is False
# An owner-narrowed image set (no ``svg``) is not a mismatch either.
narrow = _settings(images=True, image_extensions="png,jpg")
assert importer._row_kind_mismatch(svg, text_row, narrow) is False