FabledCurator/tests/test_migration_0008.py

"""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, artist_id=a.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, artist_id=a.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