brain-of-reese/tests/unit/test_svg_image_ingest.py

"""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: ``<title>`` + ``<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</title>"
    b"<desc>Two switches uplink to the router.</desc>"
    b'<g><path d="M0 0h10"/><text>rack-1</text></g>'
    b"</svg>"
)

#: The text :func:`app.rag.svg_text.extract_svg_text` yields for
#: :data:`SVG_TOPO` (pinned literally here so a change to the extraction
#: contract breaks BOTH modules, not silently).
SVG_TOPO_TEXT = "Server rack topology\nTwo switches uplink to the router.\nrack-1"

#: A pure-shape drawing: no words at all (the LOCKED A2 class).
SVG_SHAPES_ONLY = (
    b'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64">'
    b'<path d="M0 0h10"/><path d="M0 20h10"/><circle cx="1" cy="1" r="2"/>'
    b"</svg>"
)

#: 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"<svg>" + b"\n" * (MAX_SVG_BYTES + 1) + b"</svg>"

#: Well-formed bytes, unparseable markup (the second refusal class).
SVG_MALFORMED = b'<svg xmlns="http://www.w3.org/2000/svg"><title>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 ``![alt](topo.svg)``
    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![The rack plan](topo.svg)\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