bvandeusen
7c4b24c80d
An image whose on-disk path contains '#' (post folders like 'BLUE#59') served its hash-named thumbnail fine but 404'd the original: the unencoded '#' in image_url was parsed by the browser as a URL fragment, so '#59/01_timelapse.jpg' never reached the /images route. Add a shared image_url(path) helper that percent-encodes the path (safe='/') and route the 3 raw builders (gallery detail + 2 in series) through it. Not a cleanup-tool deletion — the file is on disk; only the URL was wrong. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
336 lines
12 KiB
Python
336 lines
12 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
|
|
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,
|
|
image_url,
|
|
)
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def test_image_url_percent_encodes_special_chars():
|
|
# A '#' in a post folder ('BLUE#59') would otherwise be parsed as a URL
|
|
# fragment, dropping the rest of the path and 404'ing the original while the
|
|
# hash-named thumbnail still loads. Operator-flagged 2026-06-12.
|
|
url = image_url("/images/dismassd/patreon/2024-04-28_BLUE#59/01 a.jpg")
|
|
assert url == "/images/dismassd/patreon/2024-04-28_BLUE%2359/01%20a.jpg"
|
|
# '/' stays a separator; a plain path is unchanged.
|
|
assert image_url("/images/a/b/c.jpg") == "/images/a/b/c.jpg"
|
|
|
|
|
|
def _now():
|
|
return datetime.now(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)
|
|
# No post → effective_date == created_at (alembic 0035 denorm).
|
|
r.effective_date = r.created_at
|
|
db.add(r)
|
|
records.append(r)
|
|
await db.flush()
|
|
return records
|
|
|
|
|
|
def test_cursor_roundtrip():
|
|
ts = datetime(2026, 5, 14, 12, 30, 0, tzinfo=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_ids=[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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_image_with_tags_includes_integrity_status(db):
|
|
images = await _seed_images(db, 1)
|
|
img = images[0]
|
|
img.integrity_status = "ok"
|
|
await db.flush()
|
|
svc = GalleryService(db)
|
|
payload = await svc.get_image_with_tags(img.id)
|
|
assert payload["integrity_status"] == "ok"
|
|
|
|
|
|
async def _seed_image_with_post(
|
|
db, *, sha: str, image_created_at, post_date, artist_name="test-artist",
|
|
platform="patreon", external_post_id="42",
|
|
):
|
|
"""Helper: seed an Artist + Source + Post and one ImageRecord whose
|
|
primary_post_id points at that Post. Used for date-coalesce tests."""
|
|
from backend.app.models import Artist, Post, Source
|
|
artist = Artist(name=artist_name, slug=artist_name.lower().replace(" ", "-"))
|
|
db.add(artist)
|
|
await db.flush()
|
|
source = Source(
|
|
artist_id=artist.id, platform=platform,
|
|
url=f"https://www.{platform}.com/{artist.slug}",
|
|
)
|
|
db.add(source)
|
|
await db.flush()
|
|
post = Post(
|
|
source_id=source.id, artist_id=artist.id,
|
|
external_post_id=external_post_id,
|
|
post_title="A Post", post_date=post_date,
|
|
)
|
|
db.add(post)
|
|
await db.flush()
|
|
img = ImageRecord(
|
|
path=f"/images/test/{sha[:8]}.jpg",
|
|
sha256=sha, size_bytes=1000, mime="image/jpeg",
|
|
width=100, height=100,
|
|
origin="imported_filesystem", integrity_status="unknown",
|
|
primary_post_id=post.id,
|
|
)
|
|
img.created_at = image_created_at
|
|
# Mirror the importer's denorm (alembic 0035): a linked post with a date
|
|
# sets effective_date to that date, else it falls back to created_at.
|
|
img.effective_date = post_date or image_created_at
|
|
db.add(img)
|
|
await db.flush()
|
|
return img, post
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_sorts_by_post_date_when_available(db):
|
|
"""Operator-flagged 2026-05-25: ~57k IR images all imported in the
|
|
same week sort by image.created_at and pile up in one month bucket.
|
|
Once primary_post_id is wired (via tag_apply phase 4), the gallery
|
|
should sort by Post.post_date instead, spreading them across the
|
|
actual publish years."""
|
|
base_import = _now()
|
|
# Image A: imported NOW, but post was made 2 years ago.
|
|
img_a, _ = await _seed_image_with_post(
|
|
db, sha="a" * 64,
|
|
image_created_at=base_import,
|
|
post_date=base_import - timedelta(days=730),
|
|
artist_name="Aria", external_post_id="A-1",
|
|
)
|
|
# Image B: imported NOW (1 min later), post made YESTERDAY.
|
|
img_b, _ = await _seed_image_with_post(
|
|
db, sha="b" * 64,
|
|
image_created_at=base_import - timedelta(minutes=1),
|
|
post_date=base_import - timedelta(days=1),
|
|
artist_name="Bea", external_post_id="B-1",
|
|
)
|
|
# Image C: filesystem-imported, no primary_post_id, created 5 days ago.
|
|
img_c = ImageRecord(
|
|
path="/images/test/c.jpg", sha256="c" * 64,
|
|
size_bytes=1000, mime="image/jpeg",
|
|
width=100, height=100,
|
|
origin="imported_filesystem", integrity_status="unknown",
|
|
)
|
|
img_c.created_at = base_import - timedelta(days=5)
|
|
img_c.effective_date = img_c.created_at # no post → tracks created_at
|
|
db.add(img_c)
|
|
await db.flush()
|
|
|
|
svc = GalleryService(db)
|
|
page = await svc.scroll(cursor=None, limit=10)
|
|
# Effective-date order: B (yesterday) > C (5 days ago) > A (2 years ago)
|
|
assert [i.id for i in page.images] == [img_b.id, img_c.id, img_a.id]
|
|
|
|
# API exposes both fields explicitly so the UI can show "Posted X / Imported Y".
|
|
a_payload = next(i for i in page.images if i.id == img_a.id)
|
|
assert a_payload.posted_at is not None
|
|
assert a_payload.posted_at < a_payload.created_at
|
|
c_payload = next(i for i in page.images if i.id == img_c.id)
|
|
assert c_payload.posted_at is None
|
|
assert c_payload.effective_date == c_payload.created_at
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timeline_buckets_use_post_date_when_available(db):
|
|
"""Timeline group-by must follow the same effective_date rule so the
|
|
UI's year/month navigation surfaces publish-date buckets, not the
|
|
single FC-scan bucket all migrated images share."""
|
|
base = datetime(2026, 6, 15, 12, 0, tzinfo=UTC)
|
|
await _seed_image_with_post(
|
|
db, sha="1" * 64,
|
|
image_created_at=base,
|
|
post_date=datetime(2024, 3, 10, tzinfo=UTC),
|
|
artist_name="Carl", external_post_id="C-1",
|
|
)
|
|
await _seed_image_with_post(
|
|
db, sha="2" * 64,
|
|
image_created_at=base,
|
|
post_date=datetime(2024, 3, 11, tzinfo=UTC),
|
|
artist_name="Dee", external_post_id="D-1",
|
|
)
|
|
await _seed_image_with_post(
|
|
db, sha="3" * 64,
|
|
image_created_at=base,
|
|
post_date=datetime(2025, 9, 1, tzinfo=UTC),
|
|
artist_name="Eli", external_post_id="E-1",
|
|
)
|
|
svc = GalleryService(db)
|
|
buckets = await svc.timeline()
|
|
bucket_keys = {(b.year, b.month, b.count) for b in buckets}
|
|
# Two posts in 2024-03, one in 2025-09 — even though all imported in 2026-06.
|
|
assert (2024, 3, 2) in bucket_keys
|
|
assert (2025, 9, 1) in bucket_keys
|
|
# The FC-import bucket should NOT appear since all 3 images have post_date.
|
|
assert not any(b.year == 2026 and b.month == 6 for b in buckets)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_image_with_tags_includes_posted_at_when_present(db):
|
|
base = _now()
|
|
img, _ = await _seed_image_with_post(
|
|
db, sha="f" * 64,
|
|
image_created_at=base,
|
|
post_date=base - timedelta(days=365),
|
|
artist_name="Fred", external_post_id="F-1",
|
|
)
|
|
svc = GalleryService(db)
|
|
payload = await svc.get_image_with_tags(img.id)
|
|
assert payload["posted_at"] is not None
|
|
# Image's own created_at is still surfaced separately.
|
|
assert payload["created_at"] != payload["posted_at"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_multi_tag_and(db):
|
|
images = await _seed_images(db, 5)
|
|
a = Tag(name="aa", kind=TagKind.general)
|
|
b = Tag(name="bb", kind=TagKind.general)
|
|
db.add_all([a, b])
|
|
await db.flush()
|
|
# images[0] carries BOTH tags; images[1] carries only a.
|
|
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": b.id, "source": "manual"},
|
|
{"image_record_id": images[1].id, "tag_id": a.id, "source": "manual"},
|
|
]))
|
|
svc = GalleryService(db)
|
|
both = await svc.scroll(cursor=None, limit=10, tag_ids=[a.id, b.id])
|
|
assert [i.id for i in both.images] == [images[0].id] # AND: only the both-tagged
|
|
just_a = await svc.scroll(cursor=None, limit=10, tag_ids=[a.id])
|
|
assert {i.id for i in just_a.images} == {images[0].id, images[1].id}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_media_filter(db):
|
|
imgs = await _seed_images(db, 2) # image/jpeg
|
|
vid = ImageRecord(
|
|
path="/images/test/v.mp4", sha256="v" * 64, size_bytes=1,
|
|
mime="video/mp4", width=1, height=1,
|
|
origin="imported_filesystem", integrity_status="unknown",
|
|
)
|
|
vid.created_at = _now()
|
|
vid.effective_date = vid.created_at
|
|
db.add(vid)
|
|
await db.flush()
|
|
svc = GalleryService(db)
|
|
vids = await svc.scroll(cursor=None, limit=10, media_type="video")
|
|
assert [i.id for i in vids.images] == [vid.id]
|
|
pics = await svc.scroll(cursor=None, limit=10, media_type="image")
|
|
assert {i.id for i in pics.images} == {i.id for i in imgs}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_sort_oldest_reverses_order(db):
|
|
images = await _seed_images(db, 4) # images[0] newest ... images[3] oldest
|
|
svc = GalleryService(db)
|
|
newest = await svc.scroll(cursor=None, limit=10)
|
|
oldest = await svc.scroll(cursor=None, limit=10, sort="oldest")
|
|
newest_ids = [i.id for i in newest.images]
|
|
assert newest_ids[0] == images[0].id
|
|
assert [i.id for i in oldest.images] == list(reversed(newest_ids))
|