fc-3g-ext: IR Post/Provenance restore (tag_apply phase 4) + modal artist fallback

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/test_tag_apply.py

@@ -2,9 +2,19 @@
 import json
 
 import pytest
-from sqlalchemy import select
+from sqlalchemy import func, select
 
-from backend.app.models import Artist, ImageRecord, SeriesPage, Tag, TagKind, image_tag
+from backend.app.models import (
+    Artist,
+    ImageProvenance,
+    ImageRecord,
+    Post,
+    SeriesPage,
+    Source,
+    Tag,
+    TagKind,
+    image_tag,
+)
 from backend.app.services.migrators import tag_apply
 
 pytestmark = pytest.mark.integration
@@ -21,14 +31,19 @@ async def _seed_image(db, sha: str, *, suffix="x"):
     return img
 
 
-def _write_manifest(tmp_path, *, artist_assignments=None, tag_associations=None, series_pages=None):
+def _write_manifest(
+    tmp_path, *,
+    artist_assignments=None, tag_associations=None,
+    series_pages=None, image_posts=None, schema_version=2,
+):
     d = tmp_path / "_migration_state"
     d.mkdir(parents=True, exist_ok=True)
     (d / "ir_tag_manifest.json").write_text(json.dumps({
-        "schema_version": 1,
+        "schema_version": schema_version,
         "image_artist_assignments": artist_assignments or [],
         "image_tag_associations": tag_associations or [],
         "series_pages": series_pages or [],
+        "image_posts": image_posts or [],
     }))
 
 
@@ -154,3 +169,130 @@ async def test_dry_run_makes_no_changes(db, tmp_path):
         select(image_tag.c.tag_id).where(image_tag.c.image_record_id == img_id)
     )).all()
     assert len(rows) == 0
+
+
+# --- Phase 4: image_posts (schema v2) -----------------------------------
+
+
+_POST_ENTRY = {
+    "platform": "patreon",
+    "post_id": "10001",
+    "artist": "Maewix",
+    "title": "Test Post Title",
+    "description": "<p>HTML description</p>",
+    "source_url": "https://www.patreon.com/posts/test-10001",
+    "attachment_count": 1,
+    "published_at": "2026-01-15T12:00:00+00:00",
+}
+
+
+@pytest.mark.asyncio
+async def test_image_posts_creates_source_post_provenance(db, tmp_path):
+    sha = "d" * 64
+    await _seed_image(db, sha, suffix="d")
+    await db.commit()
+
+    _write_manifest(tmp_path, image_posts=[
+        {**_POST_ENTRY, "image_sha256s": [sha]},
+    ])
+
+    result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    assert result["counts"]["rows_inserted"] >= 1
+
+    # Async Core-DML assertions — column selects, not ORM attribute access.
+    artist_count = (await db.execute(
+        select(func.count(Artist.id)).where(Artist.slug == "maewix")
+    )).scalar_one()
+    assert artist_count == 1
+
+    source_count = (await db.execute(
+        select(func.count(Source.id)).where(
+            Source.platform == "patreon",
+            Source.url == "https://www.patreon.com/maewix",
+        )
+    )).scalar_one()
+    assert source_count == 1
+
+    post_count = (await db.execute(
+        select(func.count(Post.id)).where(Post.external_post_id == "10001")
+    )).scalar_one()
+    assert post_count == 1
+
+    img_id = (await db.execute(
+        select(ImageRecord.id).where(ImageRecord.sha256 == sha)
+    )).scalar_one()
+    prov_count = (await db.execute(
+        select(func.count(ImageProvenance.id))
+        .where(ImageProvenance.image_record_id == img_id)
+    )).scalar_one()
+    assert prov_count == 1
+
+
+@pytest.mark.asyncio
+async def test_image_posts_idempotent_on_rerun(db, tmp_path):
+    sha = "e" * 64
+    await _seed_image(db, sha, suffix="e")
+    await db.commit()
+
+    _write_manifest(tmp_path, image_posts=[
+        {**_POST_ENTRY, "image_sha256s": [sha]},
+    ])
+
+    await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    result2 = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+
+    # Re-run: nothing should be newly inserted; all skipped.
+    assert result2["counts"]["rows_inserted"] == 0
+    assert result2["counts"]["rows_skipped"] >= 1
+
+    prov_count = (await db.execute(
+        select(func.count(ImageProvenance.id))
+    )).scalar_one()
+    assert prov_count == 1
+
+
+@pytest.mark.asyncio
+async def test_image_posts_unknown_sha_unmatched(db, tmp_path):
+    # No image seeded; sha doesn't exist.
+    _write_manifest(tmp_path, image_posts=[
+        {**_POST_ENTRY, "image_sha256s": ["f" * 64]},
+    ])
+    result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    unmatched_kinds = [u["kind"] for u in result["unmatched"]]
+    assert "post" in unmatched_kinds
+
+
+@pytest.mark.asyncio
+async def test_image_posts_unknown_platform_skipped(db, tmp_path):
+    sha = "1" * 64
+    await _seed_image(db, sha, suffix="1")
+    await db.commit()
+
+    _write_manifest(tmp_path, image_posts=[
+        {**_POST_ENTRY, "platform": "tumblr", "image_sha256s": [sha]},
+    ])
+    result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    assert result["counts"]["rows_skipped"] >= 1
+
+    # No Post / Source / ImageProvenance should have been created.
+    src_count = (await db.execute(
+        select(func.count(Source.id)).where(Source.platform == "tumblr")
+    )).scalar_one()
+    assert src_count == 0
+
+
+@pytest.mark.asyncio
+async def test_image_posts_dry_run_makes_no_writes(db, tmp_path):
+    sha = "2" * 64
+    await _seed_image(db, sha, suffix="2")
+    await db.commit()
+
+    _write_manifest(tmp_path, image_posts=[
+        {**_POST_ENTRY, "image_sha256s": [sha]},
+    ])
+
+    await tag_apply.apply_async(db, images_root=tmp_path, dry_run=True)
+    prov_count = (await db.execute(
+        select(func.count(ImageProvenance.id))
+    )).scalar_one()
+    assert prov_count == 0