"""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, Tag, TagKind 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_tag(importer, import_layout): import_root, _ = import_layout src = import_root / "Alice" / "first.jpg" _make_jpeg(src) importer.import_one(src) artist = importer.session.execute( select(Artist).where(Artist.slug == "alice") ).scalar_one() assert artist.name == "Alice" tag = importer.session.execute( select(Tag).where(Tag.name == "Alice").where(Tag.kind == TagKind.artist) ).scalar_one() assert tag.id is not None 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): 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 == "skipped" assert result.skip_reason == SkipReason.invalid_image 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 == []