feat(fc2a): add GalleryService — cursor scroll, timeline, image detail with neighbors
Cursor format: base64(iso8601_created_at|image_id). Pagination key is (created_at DESC, id DESC) so we don't drift when new imports land between page loads. Timeline groups by date_part(year, month) so the sidebar can render year-month jump buckets. get_image_with_tags returns full image detail plus prev/next ids so the modal viewer can navigate without an extra round-trip per arrow press. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,122 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.app.models import ImageRecord, Tag, TagKind
|
||||
from backend.app.models.tag import image_tag
|
||||
from backend.app.services.gallery_service import (
|
||||
GalleryService,
|
||||
decode_cursor,
|
||||
encode_cursor,
|
||||
)
|
||||
|
||||
|
||||
def _now():
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
async def _seed_images(db, count: int, sha_prefix: str = "0") -> list[ImageRecord]:
|
||||
base = _now()
|
||||
records = []
|
||||
for i in range(count):
|
||||
r = ImageRecord(
|
||||
path=f"/images/test/{i}.jpg",
|
||||
sha256=f"{sha_prefix}{i:063d}",
|
||||
size_bytes=1000,
|
||||
mime="image/jpeg",
|
||||
width=100,
|
||||
height=100,
|
||||
origin="imported_filesystem",
|
||||
integrity_status="unknown",
|
||||
)
|
||||
r.created_at = base - timedelta(minutes=i)
|
||||
db.add(r)
|
||||
records.append(r)
|
||||
await db.flush()
|
||||
return records
|
||||
|
||||
|
||||
def test_cursor_roundtrip():
|
||||
ts = datetime(2026, 5, 14, 12, 30, 0, tzinfo=timezone.utc)
|
||||
encoded = encode_cursor(ts, 42)
|
||||
back_ts, back_id = decode_cursor(encoded)
|
||||
assert back_ts == ts
|
||||
assert back_id == 42
|
||||
|
||||
|
||||
def test_decode_invalid_cursor_raises():
|
||||
with pytest.raises(ValueError):
|
||||
decode_cursor("not-base64!!!")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_returns_newest_first(db):
|
||||
await _seed_images(db, 5)
|
||||
svc = GalleryService(db)
|
||||
page = await svc.scroll(cursor=None, limit=10)
|
||||
assert len(page.images) == 5
|
||||
assert page.images[0].created_at > page.images[-1].created_at
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_pagination(db):
|
||||
await _seed_images(db, 5)
|
||||
svc = GalleryService(db)
|
||||
first = await svc.scroll(cursor=None, limit=2)
|
||||
assert len(first.images) == 2
|
||||
assert first.next_cursor is not None
|
||||
second = await svc.scroll(cursor=first.next_cursor, limit=2)
|
||||
assert len(second.images) == 2
|
||||
assert {i.id for i in first.images}.isdisjoint({i.id for i in second.images})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scroll_with_tag_filter(db):
|
||||
images = await _seed_images(db, 5)
|
||||
tag = Tag(name="filterme", kind=TagKind.general)
|
||||
db.add(tag)
|
||||
await db.flush()
|
||||
await db.execute(
|
||||
image_tag.insert().values(
|
||||
image_record_id=images[0].id, tag_id=tag.id, source="manual"
|
||||
)
|
||||
)
|
||||
|
||||
svc = GalleryService(db)
|
||||
page = await svc.scroll(cursor=None, limit=10, tag_id=tag.id)
|
||||
assert len(page.images) == 1
|
||||
assert page.images[0].id == images[0].id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timeline_groups_by_month(db):
|
||||
await _seed_images(db, 3)
|
||||
svc = GalleryService(db)
|
||||
buckets = await svc.timeline()
|
||||
assert sum(b.count for b in buckets) == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_date_groups_in_page(db):
|
||||
await _seed_images(db, 3)
|
||||
svc = GalleryService(db)
|
||||
page = await svc.scroll(cursor=None, limit=10)
|
||||
assert len(page.date_groups) == 1
|
||||
y, m, ids = page.date_groups[0]
|
||||
assert len(ids) == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_neighbors(db):
|
||||
images = await _seed_images(db, 3)
|
||||
svc = GalleryService(db)
|
||||
middle = images[1]
|
||||
payload = await svc.get_image_with_tags(middle.id)
|
||||
assert payload["neighbors"]["prev_id"] == images[0].id
|
||||
assert payload["neighbors"]["next_id"] == images[2].id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_image_with_tags_returns_none_for_missing(db):
|
||||
svc = GalleryService(db)
|
||||
assert await svc.get_image_with_tags(99999) is None
|
||||
Reference in New Issue
Block a user