130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
"""FC-5: ir_ingest unit tests."""
|
|
import json
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models import Tag
|
|
from backend.app.services.migrators import ir_ingest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _ir_export(tags=None, artist_assignments=None, tag_associations=None, series_pages=None):
|
|
return {
|
|
"source_app": "imagerepo",
|
|
"schema_version": 1,
|
|
"exported_at": datetime.now(UTC).isoformat(),
|
|
"tags": tags or [],
|
|
"image_artist_assignments": artist_assignments or [],
|
|
"image_tag_associations": tag_associations or [],
|
|
"series_pages": series_pages or [],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_general_tag_created(db, tmp_path):
|
|
data = _ir_export(tags=[
|
|
{"name": "blue_eyes", "kind": "general", "fandom_name": None},
|
|
])
|
|
await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False)
|
|
|
|
tag = (await db.execute(
|
|
select(Tag).where(Tag.name == "blue_eyes", Tag.kind == "general")
|
|
)).scalar_one()
|
|
assert tag.fandom_id is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_character_tag_resolves_fandom_name(db, tmp_path):
|
|
data = _ir_export(tags=[
|
|
{"name": "Wonderland", "kind": "fandom", "fandom_name": None},
|
|
{"name": "Alice", "kind": "character", "fandom_name": "Wonderland"},
|
|
])
|
|
await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False)
|
|
|
|
fandom = (await db.execute(
|
|
select(Tag).where(Tag.name == "Wonderland", Tag.kind == "fandom")
|
|
)).scalar_one()
|
|
alice = (await db.execute(
|
|
select(Tag).where(Tag.name == "Alice", Tag.kind == "character")
|
|
)).scalar_one()
|
|
assert alice.fandom_id == fandom.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artist_and_post_kinds_skipped(db, tmp_path):
|
|
data = _ir_export(tags=[
|
|
{"name": "BobArtist", "kind": "artist", "fandom_name": None},
|
|
{"name": "PostXYZ", "kind": "post", "fandom_name": None},
|
|
{"name": "general_one", "kind": "general", "fandom_name": None},
|
|
])
|
|
counts = await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False)
|
|
assert counts["rows_skipped"] >= 2
|
|
|
|
bob = (await db.execute(
|
|
select(Tag).where(Tag.name == "BobArtist")
|
|
)).scalar_one_or_none()
|
|
post = (await db.execute(
|
|
select(Tag).where(Tag.name == "PostXYZ")
|
|
)).scalar_one_or_none()
|
|
assert bob is None
|
|
assert post is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manifest_written_to_disk(db, tmp_path):
|
|
data = _ir_export(
|
|
artist_assignments=[{"sha256": "a" * 64, "artist_name": "Alice"}],
|
|
tag_associations=[
|
|
{"sha256": "a" * 64, "tag_name": "blue_eyes", "tag_kind": "general", "fandom_name": None},
|
|
],
|
|
series_pages=[
|
|
{"sha256": "a" * 64, "series_tag_name": "MySeries", "page_number": 1},
|
|
],
|
|
)
|
|
await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False)
|
|
|
|
manifest_file = tmp_path / "_migration_state" / "ir_tag_manifest.json"
|
|
assert manifest_file.exists()
|
|
manifest = json.loads(manifest_file.read_text())
|
|
assert manifest["schema_version"] == 1
|
|
assert len(manifest["image_artist_assignments"]) == 1
|
|
assert len(manifest["image_tag_associations"]) == 1
|
|
assert len(manifest["series_pages"]) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dry_run_writes_no_manifest_and_no_tags(db, tmp_path):
|
|
data = _ir_export(
|
|
tags=[{"name": "dry", "kind": "general", "fandom_name": None}],
|
|
artist_assignments=[{"sha256": "a" * 64, "artist_name": "DryArtist"}],
|
|
)
|
|
await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=True)
|
|
|
|
tag = (await db.execute(
|
|
select(Tag).where(Tag.name == "dry")
|
|
)).scalar_one_or_none()
|
|
assert tag is None
|
|
manifest_file = tmp_path / "_migration_state" / "ir_tag_manifest.json"
|
|
assert not manifest_file.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_idempotent_on_rerun(db, tmp_path):
|
|
data = _ir_export(tags=[
|
|
{"name": "rerun", "kind": "general", "fandom_name": None},
|
|
])
|
|
await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False)
|
|
counts2 = await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False)
|
|
assert counts2["rows_skipped"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_wrong_source_app(db, tmp_path):
|
|
bad = {"source_app": "elsewhere", "schema_version": 1, "tags": [],
|
|
"image_artist_assignments": [], "image_tag_associations": [], "series_pages": []}
|
|
with pytest.raises(ValueError):
|
|
await ir_ingest.migrate_async(db, data=bad, images_root=tmp_path, dry_run=False)
|