181 lines
6.2 KiB
Python
181 lines
6.2 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from backend.app.models import (
|
|
Artist,
|
|
ImageProvenance,
|
|
ImageRecord,
|
|
Post,
|
|
PostAttachment,
|
|
Source,
|
|
)
|
|
from backend.app.services.provenance_service import ProvenanceService
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
async def _seed_image(db, sha="a" + "0" * 63) -> ImageRecord:
|
|
rec = ImageRecord(
|
|
path=f"/images/test/{sha}.jpg",
|
|
sha256=sha,
|
|
size_bytes=1, mime="image/jpeg", width=1, height=1,
|
|
origin="imported_filesystem", integrity_status="unknown",
|
|
)
|
|
db.add(rec)
|
|
await db.flush()
|
|
return rec
|
|
|
|
|
|
async def _seed_post(db, *, artist_name, slug, platform, ext_id,
|
|
title=None, desc=None, count=None) -> tuple:
|
|
artist = Artist(name=artist_name, slug=slug)
|
|
db.add(artist)
|
|
await db.flush()
|
|
source = Source(artist_id=artist.id, platform=platform,
|
|
url=f"https://{platform}.test/{slug}")
|
|
db.add(source)
|
|
await db.flush()
|
|
post = Post(
|
|
source_id=source.id, external_post_id=ext_id,
|
|
post_url=f"https://{platform}.test/p/{ext_id}",
|
|
post_title=title, post_date=datetime(2023, 8, 1, tzinfo=UTC),
|
|
description=desc, attachment_count=count,
|
|
)
|
|
db.add(post)
|
|
await db.flush()
|
|
return artist, source, post
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_image_missing_returns_none(db):
|
|
svc = ProvenanceService(db)
|
|
assert await svc.for_image(999999) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_image_no_provenance_returns_empty_list(db):
|
|
rec = await _seed_image(db)
|
|
svc = ProvenanceService(db)
|
|
payload = await svc.for_image(rec.id)
|
|
assert payload == {
|
|
"image_id": rec.id, "provenance": [], "attachments": []
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_image_single_provenance_full_shape(db):
|
|
rec = await _seed_image(db)
|
|
artist, source, post = await _seed_post(
|
|
db, artist_name="Alice", slug="alice", platform="patreon",
|
|
ext_id="555", title="Set 1", desc="<p>hi</p><script>x</script>",
|
|
count=2,
|
|
)
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
|
|
source_id=source.id))
|
|
await db.flush()
|
|
|
|
svc = ProvenanceService(db)
|
|
payload = await svc.for_image(rec.id)
|
|
assert payload["image_id"] == rec.id
|
|
assert len(payload["provenance"]) == 1
|
|
e = payload["provenance"][0]
|
|
assert e["post"]["id"] == post.id
|
|
assert e["post"]["external_post_id"] == "555"
|
|
assert e["post"]["title"] == "Set 1"
|
|
assert e["post"]["attachment_count"] == 2
|
|
assert e["post"]["description_html"] == "<p>hi</p>" # script removed
|
|
assert e["post"]["url"] == "https://patreon.test/p/555"
|
|
assert e["post"]["date"].startswith("2023-08-01")
|
|
assert e["source"] == {"id": source.id, "platform": "patreon",
|
|
"url": source.url}
|
|
assert e["artist"] == {"id": artist.id, "name": "Alice",
|
|
"slug": "alice"}
|
|
assert e["provenance_id"] is not None
|
|
assert e["captured_at"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_image_multiple_provenance_peer_ordering(db):
|
|
rec = await _seed_image(db)
|
|
_, s1, p1 = await _seed_post(db, artist_name="A1", slug="a1",
|
|
platform="patreon", ext_id="1")
|
|
_, s2, p2 = await _seed_post(db, artist_name="A2", slug="a2",
|
|
platform="fanbox", ext_id="2")
|
|
ip1 = ImageProvenance(image_record_id=rec.id, post_id=p1.id,
|
|
source_id=s1.id)
|
|
ip2 = ImageProvenance(image_record_id=rec.id, post_id=p2.id,
|
|
source_id=s2.id)
|
|
db.add(ip1)
|
|
await db.flush()
|
|
db.add(ip2)
|
|
await db.flush()
|
|
|
|
svc = ProvenanceService(db)
|
|
payload = await svc.for_image(rec.id)
|
|
ids = [e["provenance_id"] for e in payload["provenance"]]
|
|
assert ids == sorted(ids) # captured_at, id ascending → insertion order
|
|
assert len(payload["provenance"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_image_null_post_fields_serialize_null(db):
|
|
rec = await _seed_image(db)
|
|
_, source, post = await _seed_post(
|
|
db, artist_name="Bob", slug="bob", platform="x", ext_id="9",
|
|
) # title/desc/count/post_url default-ish
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
|
|
source_id=source.id))
|
|
await db.flush()
|
|
svc = ProvenanceService(db)
|
|
e = (await svc.for_image(rec.id))["provenance"][0]
|
|
assert e["post"]["title"] is None
|
|
assert e["post"]["description_html"] is None
|
|
assert e["post"]["attachment_count"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_post_missing_returns_none(db):
|
|
svc = ProvenanceService(db)
|
|
assert await svc.for_post(999999) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_post_returns_post_source_artist(db):
|
|
artist, source, post = await _seed_post(
|
|
db, artist_name="Carol", slug="carol", platform="patreon",
|
|
ext_id="77", title="T", desc="<p>d</p>", count=3,
|
|
)
|
|
svc = ProvenanceService(db)
|
|
payload = await svc.for_post(post.id)
|
|
assert payload["post"]["id"] == post.id
|
|
assert payload["post"]["title"] == "T"
|
|
assert payload["post"]["description_html"] == "<p>d</p>"
|
|
assert payload["post"]["attachment_count"] == 3
|
|
assert payload["source"] == {"id": source.id, "platform": "patreon",
|
|
"url": source.url}
|
|
assert payload["artist"] == {"id": artist.id, "name": "Carol",
|
|
"slug": "carol"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_post_includes_attachments(db):
|
|
artist, source, post = await _seed_post(
|
|
db, artist_name="Att", slug="att", platform="patreon",
|
|
ext_id="55",
|
|
)
|
|
db.add(PostAttachment(
|
|
post_id=post.id, artist_id=artist.id, sha256="t" + "0" * 63,
|
|
path="/images/attachments/t00/t.zip", original_filename="t.zip",
|
|
ext=".zip", mime="application/zip", size_bytes=9,
|
|
))
|
|
await db.flush()
|
|
svc = ProvenanceService(db)
|
|
payload = await svc.for_post(post.id)
|
|
assert len(payload["attachments"]) == 1
|
|
att = payload["attachments"][0]
|
|
assert att["original_filename"] == "t.zip"
|
|
assert att["download_url"].endswith(
|
|
f"/api/attachments/{att['id']}/download"
|
|
)
|