FabledCurator/tests/test_importer.py

"""Importer integration tests.

These exercise the real Importer against a real Postgres + a real filesystem
(via tmp_path). The DB fixture rolls back after each test so the import_root
fixture is fresh per-test.
"""

from pathlib import Path

import pytest
from PIL import Image
from sqlalchemy import select

from backend.app.models import Artist, ImageRecord, ImportSettings
from backend.app.services.importer import Importer, SkipReason
from backend.app.services.thumbnailer import Thumbnailer

pytestmark = pytest.mark.integration


@pytest.fixture
def import_layout(tmp_path):
    import_root = tmp_path / "import"
    images_root = tmp_path / "images"
    import_root.mkdir()
    images_root.mkdir()
    return import_root, images_root


def _make_jpeg(path: Path, size: tuple[int, int] = (800, 600)):
    path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", size, color=(120, 200, 80)).save(path, "JPEG")


def _make_png_rgba(path: Path, size: tuple[int, int], alpha: int):
    path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGBA", size, color=(120, 200, 80, alpha)).save(path, "PNG")


@pytest.fixture
def importer(db_sync, import_layout):
    import_root, images_root = import_layout
    settings = db_sync.execute(select(ImportSettings).where(ImportSettings.id == 1)).scalar_one()
    thumbnailer = Thumbnailer(images_root=images_root)
    return Importer(
        session=db_sync,
        images_root=images_root,
        import_root=import_root,
        thumbnailer=thumbnailer,
        settings=settings,
    )


def test_import_one_happy_path(importer, import_layout):
    import_root, images_root = import_layout
    src = import_root / "Alice" / "first.jpg"
    _make_jpeg(src)
    result = importer.import_one(src)
    assert result.status == "imported"
    record = importer.session.get(ImageRecord, result.image_id)
    assert record.path.startswith(str(images_root / "Alice"))
    assert record.path.endswith(".jpg")


def test_folder_creates_artist_and_links_artist_id(importer, import_layout):
    # FC-2d-vii-c: folder import creates the Artist and sets the canonical
    # image_record.artist_id — no artist-kind Tag (that path was retired;
    # the no-tag invariant is covered by test_importer_artist_id.py).
    import_root, _ = import_layout
    src = import_root / "Alice" / "first.jpg"
    _make_jpeg(src)
    result = importer.import_one(src)

    artist = importer.session.execute(
        select(Artist).where(Artist.slug == "alice")
    ).scalar_one()
    assert artist.name == "Alice"

    record = importer.session.get(ImageRecord, result.image_id)
    assert record.artist_id == artist.id


def test_dedup_by_hash(importer, import_layout):
    import_root, _ = import_layout
    src1 = import_root / "Alice" / "first.jpg"
    src2 = import_root / "Bob" / "duplicate.jpg"
    _make_jpeg(src1)
    src2.parent.mkdir(parents=True, exist_ok=True)
    src2.write_bytes(src1.read_bytes())

    r1 = importer.import_one(src1)
    r2 = importer.import_one(src2)
    assert r1.status == "imported"
    assert r2.status == "skipped"
    assert r2.skip_reason == SkipReason.duplicate_hash


def test_min_width_filter(importer, import_layout):
    import_root, _ = import_layout
    importer.settings.min_width = 1000  # mutate the in-memory copy for this test
    src = import_root / "small.jpg"
    _make_jpeg(src, size=(500, 500))
    result = importer.import_one(src)
    assert result.status == "skipped"
    assert result.skip_reason == SkipReason.too_small


def test_transparent_filter(importer, import_layout):
    import_root, _ = import_layout
    importer.settings.skip_transparent = True
    importer.settings.transparency_threshold = 0.5
    src = import_root / "ghost.png"
    _make_png_rgba(src, size=(100, 100), alpha=0)  # fully transparent
    result = importer.import_one(src)
    assert result.status == "skipped"
    assert result.skip_reason == SkipReason.too_transparent


def test_unsupported_extension(importer, import_layout):
    # FC-2d-iii: non-media is no longer skipped — it's captured as a
    # PostAttachment so nothing a post contained is lost.
    import_root, _ = import_layout
    src = import_root / "Alice" / "notes.txt"
    src.parent.mkdir(parents=True, exist_ok=True)
    src.write_text("hello")
    result = importer.import_one(src)
    assert result.status == "attached"


def test_root_level_file_has_no_artist(importer, import_layout):
    import_root, _ = import_layout
    src = import_root / "loose.jpg"
    _make_jpeg(src)
    importer.import_one(src)
    artists = importer.session.execute(select(Artist)).scalars().all()
    assert artists == []


def test_pil_load_oserror_in_transparency_check_skips_not_raises(
    importer, import_layout, monkeypatch,
):
    """PIL.verify() only validates header structure — broken pixel data
    only surfaces when load() actually decodes. The importer must catch
    the OSError and return a skipped: invalid_image result so the Celery
    autoretry loop doesn't bounce the same broken file forever.
    Operator hit this 2026-05-25 with a corrupt JPEG in the IR set."""
    import_root, _ = import_layout
    src = import_root / "Bob" / "corrupt.png"
    # Make a real RGBA PNG so the has_alpha path engages.
    _make_png_rgba(src, (100, 100), alpha=128)

    importer.settings.skip_transparent = True
    importer.settings.transparency_threshold = 0.5

    # Force the next _transparency_pct call to raise as if PIL's load()
    # blew up on truncated pixel data.
    def _boom(_self, _src):
        raise OSError("broken data stream when reading image file")
    monkeypatch.setattr(
        type(importer), "_transparency_pct", _boom,
    )

    result = importer.import_one(src)
    assert result.status == "skipped"
    assert result.skip_reason == SkipReason.invalid_image
    assert "transparency check" in (result.error or "")


def test_pil_load_oserror_in_phash_compute_skips_not_raises(
    importer, import_layout, monkeypatch,
):
    """Same shape as the transparency-check guard, but for the phash
    compute block — the OTHER place PIL.load() runs implicitly during
    the dedup pipeline."""
    import_root, _ = import_layout
    src = import_root / "Carol" / "corrupt.jpg"
    _make_jpeg(src)

    # Disable transparency check so we reach the phash compute block.
    importer.settings.skip_transparent = False

    from backend.app.services import importer as importer_module

    def _boom(_im):
        raise OSError("broken data stream when reading image file")
    monkeypatch.setattr(importer_module, "compute_phash", _boom)

    result = importer.import_one(src)
    assert result.status == "skipped"
    assert result.skip_reason == SkipReason.invalid_image
    assert "phash compute" in (result.error or "")