138 lines
3.7 KiB
Python
138 lines
3.7 KiB
Python
"""FC-2d-vii-c: image_record.artist_id + backfill + artist-tag delete."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select, text
|
|
|
|
from backend.app.models import (
|
|
Artist,
|
|
ImageProvenance,
|
|
ImageRecord,
|
|
Post,
|
|
Source,
|
|
Tag,
|
|
TagKind,
|
|
)
|
|
from backend.app.models.tag import image_tag
|
|
from backend.app.utils.artist_backfill import (
|
|
BACKFILL_PRIMARY_SQL,
|
|
BACKFILL_PROVENANCE_SQL,
|
|
BACKFILL_TAG_SQL,
|
|
DELETE_ARTIST_TAGS_SQL,
|
|
)
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def test_image_record_has_artist_id_column():
|
|
assert "artist_id" in {c.name for c in ImageRecord.__table__.columns}
|
|
|
|
|
|
async def _img(db, n):
|
|
rec = ImageRecord(
|
|
path=f"/images/bf/{n}.jpg", sha256=f"bf{n:062d}",
|
|
size_bytes=1, mime="image/jpeg", width=1, height=1,
|
|
origin="imported_filesystem", integrity_status="unknown",
|
|
)
|
|
rec.created_at = datetime.now(UTC) - timedelta(minutes=n)
|
|
db.add(rec)
|
|
await db.flush()
|
|
return rec
|
|
|
|
|
|
async def _artist_source(db, name, slug):
|
|
a = Artist(name=name, slug=slug)
|
|
db.add(a)
|
|
await db.flush()
|
|
s = Source(artist_id=a.id, platform="patreon",
|
|
url=f"https://p.test/{slug}")
|
|
db.add(s)
|
|
await db.flush()
|
|
return a, s
|
|
|
|
|
|
async def _run_backfill(db):
|
|
await db.execute(text(BACKFILL_PRIMARY_SQL))
|
|
await db.execute(text(BACKFILL_PROVENANCE_SQL))
|
|
await db.execute(text(BACKFILL_TAG_SQL))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_primary_post(db):
|
|
rec = await _img(db, 1)
|
|
a, s = await _artist_source(db, "Alice", "alice")
|
|
post = Post(source_id=s.id, external_post_id="1")
|
|
db.add(post)
|
|
await db.flush()
|
|
rec.primary_post_id = post.id
|
|
await db.flush()
|
|
await _run_backfill(db)
|
|
got = await db.scalar(
|
|
select(ImageRecord.artist_id).where(ImageRecord.id == rec.id)
|
|
)
|
|
assert got == a.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_provenance_fallback(db):
|
|
rec = await _img(db, 1)
|
|
a, s = await _artist_source(db, "Bob", "bob")
|
|
post = Post(source_id=s.id, external_post_id="2")
|
|
db.add(post)
|
|
await db.flush()
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
|
|
source_id=s.id))
|
|
await db.flush()
|
|
await _run_backfill(db)
|
|
got = await db.scalar(
|
|
select(ImageRecord.artist_id).where(ImageRecord.id == rec.id)
|
|
)
|
|
assert got == a.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_artist_tag_by_name(db):
|
|
rec = await _img(db, 1)
|
|
a = Artist(name="Carol", slug="carol")
|
|
db.add(a)
|
|
await db.flush()
|
|
tag = Tag(name="Carol", kind=TagKind.artist)
|
|
db.add(tag)
|
|
await db.flush()
|
|
await db.execute(image_tag.insert().values(
|
|
image_record_id=rec.id, tag_id=tag.id, source="auto"))
|
|
await db.flush()
|
|
await _run_backfill(db)
|
|
got = await db.scalar(
|
|
select(ImageRecord.artist_id).where(ImageRecord.id == rec.id)
|
|
)
|
|
assert got == a.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_signal_stays_null(db):
|
|
rec = await _img(db, 1)
|
|
await _run_backfill(db)
|
|
got = await db.scalar(
|
|
select(ImageRecord.artist_id).where(ImageRecord.id == rec.id)
|
|
)
|
|
assert got is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_removes_only_artist_tags(db):
|
|
artist_tag = Tag(name="Dave", kind=TagKind.artist)
|
|
general_tag = Tag(name="forest", kind=TagKind.general)
|
|
db.add_all([artist_tag, general_tag])
|
|
await db.flush()
|
|
await db.execute(text(DELETE_ARTIST_TAGS_SQL))
|
|
remaining = await db.scalar(
|
|
select(func.count()).select_from(Tag).where(Tag.kind == TagKind.artist)
|
|
)
|
|
assert remaining == 0
|
|
survived = await db.scalar(
|
|
select(func.count()).select_from(Tag).where(Tag.kind == TagKind.general)
|
|
)
|
|
assert survived >= 1
|