"""Phase-2 faceted refine: new composable filter params + the facets() aggregate (platform / curation-flags / date-range counts, scoped live to the current filter with per-group minus-self semantics).""" from datetime import UTC, datetime import pytest from backend.app.models import ( Artist, ImageProvenance, ImageRecord, Post, Source, Tag, TagKind, ) from backend.app.models.tag import image_tag from backend.app.services.gallery_service import ( UNSOURCED_PLATFORM, GalleryService, ) pytestmark = pytest.mark.integration _BASE = datetime(2026, 1, 1, 12, 0, tzinfo=UTC) async def _artist(db, name): a = Artist(name=name, slug=name.lower()) db.add(a) await db.flush() return a async def _sourced_image(db, n, platform, artist, eff): """A downloaded image with a Source(platform) + Post + provenance row.""" s = Source( artist_id=artist.id, platform=platform, url=f"https://{platform}.test/{artist.slug}/{n}", ) db.add(s) await db.flush() p = Post(source_id=s.id, artist_id=artist.id, external_post_id=f"{platform}-{n}") db.add(p) await db.flush() rec = ImageRecord( path=f"/images/s/{platform}-{n}.jpg", sha256=f"a{n:063d}", size_bytes=1, mime="image/jpeg", width=1, height=1, origin="downloaded", integrity_status="ok", artist_id=artist.id, primary_post_id=p.id, ) rec.created_at = eff rec.effective_date = eff db.add(rec) await db.flush() db.add(ImageProvenance(image_record_id=rec.id, post_id=p.id, source_id=s.id)) await db.flush() return rec async def _fs_image(db, n, eff, artist=None): """A filesystem-imported image: no provenance/source (unsourced), artist optional.""" rec = ImageRecord( path=f"/images/fs/{n}.jpg", sha256=f"b{n:063d}", size_bytes=1, mime="image/jpeg", width=1, height=1, origin="imported_filesystem", integrity_status="unknown", artist_id=artist.id if artist else None, ) rec.created_at = eff rec.effective_date = eff db.add(rec) await db.flush() return rec async def _tag(db, image, name): t = Tag(name=name, kind=TagKind.general) db.add(t) await db.flush() await db.execute(image_tag.insert().values( image_record_id=image.id, tag_id=t.id, source="manual")) await db.flush() return t # --- facets() aggregate ---------------------------------------------------- @pytest.mark.asyncio async def test_facets_platform_counts(db): ar = await _artist(db, "Ar") await _sourced_image(db, 1, "patreon", ar, _BASE) await _sourced_image(db, 2, "patreon", ar, _BASE) await _sourced_image(db, 3, "pixiv", ar, _BASE) await _fs_image(db, 4, _BASE) # unsourced → null bucket svc = GalleryService(db) f = await svc.facets() counts = {p["value"]: p["count"] for p in f.platforms} assert counts["patreon"] == 2 assert counts["pixiv"] == 1 assert counts[None] == 1 assert f.total == 4 @pytest.mark.asyncio async def test_facets_counts_image_under_each_platform(db): """A single image cross-posted to two platforms counts under both (DISTINCT image per platform), but is one image in total.""" ar = await _artist(db, "Ar") rec = await _sourced_image(db, 1, "patreon", ar, _BASE) s2 = Source(artist_id=ar.id, platform="pixiv", url="https://pixiv.test/ar/x") db.add(s2) await db.flush() p2 = Post(source_id=s2.id, artist_id=ar.id, external_post_id="pixiv-x") db.add(p2) await db.flush() db.add(ImageProvenance(image_record_id=rec.id, post_id=p2.id, source_id=s2.id)) await db.flush() svc = GalleryService(db) f = await svc.facets() counts = {p["value"]: p["count"] for p in f.platforms} assert counts["patreon"] == 1 assert counts["pixiv"] == 1 assert f.total == 1 @pytest.mark.asyncio async def test_facets_curation_flags(db): ar = await _artist(db, "Ar") tagged = await _fs_image(db, 1, _BASE, artist=ar) await _tag(db, tagged, "x") await _fs_image(db, 2, _BASE, artist=ar) # untagged, has artist await _fs_image(db, 3, _BASE, artist=None) # untagged, no artist svc = GalleryService(db) f = await svc.facets() assert f.untagged == 2 assert f.no_artist == 1 @pytest.mark.asyncio async def test_facets_date_bounds(db): ar = await _artist(db, "Ar") await _fs_image(db, 1, datetime(2020, 1, 1, tzinfo=UTC), artist=ar) await _fs_image(db, 2, datetime(2026, 6, 1, tzinfo=UTC), artist=ar) svc = GalleryService(db) f = await svc.facets() assert f.date_min == datetime(2020, 1, 1, tzinfo=UTC) assert f.date_max == datetime(2026, 6, 1, tzinfo=UTC) @pytest.mark.asyncio async def test_facets_platform_minus_self(db): """The platform group ignores its OWN active selection so siblings stay visible/switchable; total still honors the platform filter.""" ar = await _artist(db, "Ar") await _sourced_image(db, 1, "patreon", ar, _BASE) await _sourced_image(db, 2, "pixiv", ar, _BASE) svc = GalleryService(db) f = await svc.facets(platform="patreon") counts = {p["value"]: p["count"] for p in f.platforms} assert counts.get("pixiv") == 1 assert counts.get("patreon") == 1 assert f.total == 1 @pytest.mark.asyncio async def test_facets_other_filter_scopes_platforms(db): """A non-platform filter (media) narrows the platform counts.""" ar = await _artist(db, "Ar") await _sourced_image(db, 1, "patreon", ar, _BASE) vid = await _sourced_image(db, 2, "patreon", ar, _BASE) vid.mime = "video/mp4" await db.flush() svc = GalleryService(db) f = await svc.facets(media_type="image") counts = {p["value"]: p["count"] for p in f.platforms} assert counts["patreon"] == 1 # --- new scroll filter params ---------------------------------------------- @pytest.mark.asyncio async def test_scroll_platform_filter(db): ar = await _artist(db, "Ar") pat = await _sourced_image(db, 1, "patreon", ar, _BASE) await _sourced_image(db, 2, "pixiv", ar, _BASE) svc = GalleryService(db) page = await svc.scroll(cursor=None, limit=10, platform="patreon") assert [i.id for i in page.images] == [pat.id] @pytest.mark.asyncio async def test_scroll_platform_unsourced(db): ar = await _artist(db, "Ar") await _sourced_image(db, 1, "patreon", ar, _BASE) fs = await _fs_image(db, 2, _BASE) svc = GalleryService(db) page = await svc.scroll(cursor=None, limit=10, platform=UNSOURCED_PLATFORM) assert [i.id for i in page.images] == [fs.id] @pytest.mark.asyncio async def test_scroll_untagged_filter(db): ar = await _artist(db, "Ar") tagged = await _fs_image(db, 1, _BASE, artist=ar) await _tag(db, tagged, "x") untag = await _fs_image(db, 2, _BASE, artist=ar) svc = GalleryService(db) page = await svc.scroll(cursor=None, limit=10, untagged=True) assert [i.id for i in page.images] == [untag.id] @pytest.mark.asyncio async def test_scroll_no_artist_filter(db): ar = await _artist(db, "Ar") await _fs_image(db, 1, _BASE, artist=ar) none = await _fs_image(db, 2, _BASE, artist=None) svc = GalleryService(db) page = await svc.scroll(cursor=None, limit=10, no_artist=True) assert [i.id for i in page.images] == [none.id] @pytest.mark.asyncio async def test_scroll_date_range_filter(db): ar = await _artist(db, "Ar") old = await _fs_image(db, 1, datetime(2020, 1, 1, tzinfo=UTC), artist=ar) new = await _fs_image(db, 2, datetime(2026, 6, 1, tzinfo=UTC), artist=ar) svc = GalleryService(db) page = await svc.scroll( cursor=None, limit=10, date_from=datetime(2026, 1, 1, tzinfo=UTC)) assert [i.id for i in page.images] == [new.id] page2 = await svc.scroll( cursor=None, limit=10, date_to=datetime(2021, 1, 1, tzinfo=UTC)) assert [i.id for i in page2.images] == [old.id] @pytest.mark.asyncio async def test_scroll_combines_new_filters_and(db): """platform + untagged + no_artist compose with AND.""" ar = await _artist(db, "Ar") # match: unsourced, untagged, no artist match = await _fs_image(db, 1, _BASE, artist=None) # decoys violating one clause each await _fs_image(db, 2, _BASE, artist=ar) # has artist tagged = await _fs_image(db, 3, _BASE, artist=None) await _tag(db, tagged, "x") # tagged svc = GalleryService(db) page = await svc.scroll( cursor=None, limit=10, platform=UNSOURCED_PLATFORM, untagged=True, no_artist=True) assert [i.id for i in page.images] == [match.id]