bvandeusen
6959e1220c
This reverts 2529b51. Not a retreat — a reordering, on the operator's
call, and the better sequence.
The squash's acceptance test (run 4971) found ~130 places where the ORM
models do not describe the deployed schema (#3275), including a
unique=True the database never had and two UNIQUE indexes that exist
only in migrations. Collapsing now would have baked all of that into the
one file a public installer starts from.
So: fix the drift first as ordinary migrations on the intact chain, let
the operator deploy so their database moves to the corrected head, and
only then collapse. The baseline is then generated from reconciled
models and reproduces a schema worth reproducing.
Nothing is lost by reverting. The baseline was never deployed, and
regenerating it after the fixes is strictly better than patching this
copy — it will come out of autogenerate correct rather than needing the
same hand-finishing twice.
138 lines
3.8 KiB
Python
138 lines
3.8 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, 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
|