feat(gallery): tag→gallery nav from modal chips (#5) + OR/exclude tag scope (#6a)
Cluster A, milestone #97. #5: clicking an image-modal tag chip's body now closes the modal and opens the gallery filtered for that one tag (fresh filter); ✕/kebab stay as the explicit remove/rename controls. #6a (backend of OR/exclude filtering): gallery_service._apply_scope gains a structured tag model — tag_or_groups (AND-of-OR: one EXISTS(tag_id IN group) per group) + tag_exclude (NOT EXISTS(tag_id IN exclude)) — layered additively on the existing tag_ids AND path so cursors/facets/deep-links are untouched. Threaded through scroll/timeline/jump_cursor/facets/similar + facets common dict; _require_single_filter rejects post_id combined with OR/exclude. API parses tag_or (repeatable → one OR-group each) + tag_not (csv exclude). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XCUHUGQLrBrkgyk1t49kpX
This commit is contained in:
@@ -2,7 +2,8 @@ from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.app.models import ImageRecord
|
||||
from backend.app.models import ImageRecord, Tag, TagKind
|
||||
from backend.app.models.tag import image_tag
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
@@ -114,3 +115,62 @@ async def test_scroll_date_from_param(client, db):
|
||||
resp = await client.get("/api/gallery/scroll?date_from=2099-01-01&limit=10")
|
||||
assert resp.status_code == 200
|
||||
assert (await resp.get_json())["images"] == []
|
||||
|
||||
|
||||
async def _seed_tagged(db):
|
||||
"""3 images: [0]→a, [1]→b, [2]→x. Returns (records, {a,b,x} tag ids)."""
|
||||
records = []
|
||||
base = datetime.now(UTC)
|
||||
for i in range(3):
|
||||
r = ImageRecord(
|
||||
path=f"/images/t/{i}.jpg", sha256=f"t{i:063d}",
|
||||
size_bytes=1, mime="image/jpeg", width=1, height=1,
|
||||
origin="imported_filesystem", integrity_status="unknown",
|
||||
)
|
||||
r.created_at = base - timedelta(minutes=i)
|
||||
r.effective_date = r.created_at
|
||||
db.add(r)
|
||||
records.append(r)
|
||||
a = Tag(name="apia", kind=TagKind.general)
|
||||
b = Tag(name="apib", kind=TagKind.general)
|
||||
x = Tag(name="apix", kind=TagKind.general)
|
||||
db.add_all([a, b, x])
|
||||
await db.flush()
|
||||
await db.execute(image_tag.insert().values([
|
||||
{"image_record_id": records[0].id, "tag_id": a.id, "source": "manual"},
|
||||
{"image_record_id": records[1].id, "tag_id": b.id, "source": "manual"},
|
||||
{"image_record_id": records[2].id, "tag_id": x.id, "source": "manual"},
|
||||
]))
|
||||
await db.commit()
|
||||
return records, {"a": a.id, "b": b.id, "x": x.id}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_tag_or_param(client, db):
|
||||
# ?tag_or=a,b is one OR-group → images carrying a OR b (not x-only).
|
||||
records, ids = await _seed_tagged(db)
|
||||
resp = await client.get(f"/api/gallery/scroll?tag_or={ids['a']},{ids['b']}&limit=10")
|
||||
assert resp.status_code == 200
|
||||
got = {i["id"] for i in (await resp.get_json())["images"]}
|
||||
assert got == {records[0].id, records[1].id}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_tag_not_param(client, db):
|
||||
# ?tag_not=x excludes the x-tagged image.
|
||||
records, ids = await _seed_tagged(db)
|
||||
resp = await client.get(f"/api/gallery/scroll?tag_not={ids['x']}&limit=10")
|
||||
assert resp.status_code == 200
|
||||
got = {i["id"] for i in (await resp.get_json())["images"]}
|
||||
assert got == {records[0].id, records[1].id}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_repeated_tag_or_groups_anded(client, db):
|
||||
# Two ?tag_or= groups AND together: (a) AND (b) → no single image has both.
|
||||
records, ids = await _seed_tagged(db)
|
||||
resp = await client.get(
|
||||
f"/api/gallery/scroll?tag_or={ids['a']}&tag_or={ids['b']}&limit=10"
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert (await resp.get_json())["images"] == []
|
||||
|
||||
@@ -305,6 +305,93 @@ async def test_scroll_multi_tag_and(db):
|
||||
assert {i.id for i in just_a.images} == {images[0].id, images[1].id}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_tag_or_group(db):
|
||||
# #6: one OR-group — match images carrying AT LEAST ONE of the group's tags.
|
||||
images = await _seed_images(db, 3)
|
||||
a = Tag(name="ora", kind=TagKind.general)
|
||||
b = Tag(name="orb", kind=TagKind.general)
|
||||
db.add_all([a, b])
|
||||
await db.flush()
|
||||
await db.execute(image_tag.insert().values([
|
||||
{"image_record_id": images[0].id, "tag_id": a.id, "source": "manual"},
|
||||
{"image_record_id": images[1].id, "tag_id": b.id, "source": "manual"},
|
||||
# images[2] carries neither.
|
||||
]))
|
||||
svc = GalleryService(db)
|
||||
page = await svc.scroll(cursor=None, limit=10, tag_or_groups=[[a.id, b.id]])
|
||||
assert {i.id for i in page.images} == {images[0].id, images[1].id}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_tag_or_groups_anded(db):
|
||||
# Two OR-groups AND together: image must satisfy (a OR b) AND (c).
|
||||
images = await _seed_images(db, 3)
|
||||
a = Tag(name="aaa", kind=TagKind.general)
|
||||
b = Tag(name="bbb", kind=TagKind.general)
|
||||
c = Tag(name="ccc", kind=TagKind.general)
|
||||
db.add_all([a, b, c])
|
||||
await db.flush()
|
||||
await db.execute(image_tag.insert().values([
|
||||
{"image_record_id": images[0].id, "tag_id": a.id, "source": "manual"},
|
||||
{"image_record_id": images[0].id, "tag_id": c.id, "source": "manual"},
|
||||
{"image_record_id": images[1].id, "tag_id": b.id, "source": "manual"},
|
||||
{"image_record_id": images[2].id, "tag_id": a.id, "source": "manual"},
|
||||
]))
|
||||
svc = GalleryService(db)
|
||||
page = await svc.scroll(
|
||||
cursor=None, limit=10, tag_or_groups=[[a.id, b.id], [c.id]],
|
||||
)
|
||||
# Only images[0] has both an (a-or-b) tag AND c.
|
||||
assert [i.id for i in page.images] == [images[0].id]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_tag_exclude(db):
|
||||
# #6: exclude — drop images carrying ANY excluded tag.
|
||||
images = await _seed_images(db, 3)
|
||||
x = Tag(name="xxx", kind=TagKind.general)
|
||||
db.add(x)
|
||||
await db.flush()
|
||||
await db.execute(image_tag.insert().values(
|
||||
image_record_id=images[0].id, tag_id=x.id, source="manual",
|
||||
))
|
||||
svc = GalleryService(db)
|
||||
page = await svc.scroll(cursor=None, limit=10, tag_exclude=[x.id])
|
||||
assert {i.id for i in page.images} == {images[1].id, images[2].id}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_or_and_exclude_compose(db):
|
||||
# The structured model composes: (a OR b) AND NOT x.
|
||||
images = await _seed_images(db, 3)
|
||||
a = Tag(name="ca", kind=TagKind.general)
|
||||
b = Tag(name="cb", kind=TagKind.general)
|
||||
x = Tag(name="cx", kind=TagKind.general)
|
||||
db.add_all([a, b, x])
|
||||
await db.flush()
|
||||
await db.execute(image_tag.insert().values([
|
||||
{"image_record_id": images[0].id, "tag_id": a.id, "source": "manual"},
|
||||
{"image_record_id": images[0].id, "tag_id": x.id, "source": "manual"},
|
||||
{"image_record_id": images[1].id, "tag_id": b.id, "source": "manual"},
|
||||
]))
|
||||
svc = GalleryService(db)
|
||||
page = await svc.scroll(
|
||||
cursor=None, limit=10, tag_or_groups=[[a.id, b.id]], tag_exclude=[x.id],
|
||||
)
|
||||
# images[0] matches the OR-group but is excluded by x; images[1] survives.
|
||||
assert [i.id for i in page.images] == [images[1].id]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_require_single_filter_rejects_post_id_with_or_or_exclude(db):
|
||||
svc = GalleryService(db)
|
||||
with pytest.raises(ValueError, match="post_id cannot be combined"):
|
||||
await svc.scroll(cursor=None, limit=10, post_id=5, tag_or_groups=[[1]])
|
||||
with pytest.raises(ValueError, match="post_id cannot be combined"):
|
||||
await svc.scroll(cursor=None, limit=10, post_id=5, tag_exclude=[1])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_media_filter(db):
|
||||
imgs = await _seed_images(db, 2) # image/jpeg
|
||||
|
||||
Reference in New Issue
Block a user