FabledCurator/tests/test_importer_upsert_helpers.py

"""Tests for Importer._find_or_create_source / _find_or_create_post —
the race-safe savepoint-based helpers that replaced the previous
check-then-insert pattern.

Operator-flagged 2026-05-26: concurrent workers processing different
files in the same post both found no existing Source row, then both
INSERTed, tripping uq_source_artist_platform_url and poisoning the
session with `psycopg.errors.UniqueViolation`. The new helpers wrap
the INSERT in a savepoint and recover from IntegrityError by
re-selecting the row that the concurrent op committed.

Tests cover:
  - idempotent return: same (artist_id, platform, url) → same Source row
  - idempotent return for Post: same (source_id, external_post_id) → same Post
  - IntegrityError recovery: monkeypatched flush raises once, helper
    finds the row a concurrent op committed
"""

from pathlib import Path

import pytest
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from backend.app.models import Artist, ImportSettings, Post, Source
from backend.app.services.importer import Importer
from backend.app.services.thumbnailer import Thumbnailer

pytestmark = pytest.mark.integration


@pytest.fixture
def importer(db_sync, tmp_path):
    import_root = tmp_path / "import"
    images_root = tmp_path / "images"
    import_root.mkdir()
    images_root.mkdir()
    settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    return Importer(
        session=db_sync,
        images_root=images_root,
        import_root=import_root,
        thumbnailer=Thumbnailer(images_root=images_root),
        settings=settings,
    )


@pytest.fixture
def artist_row(db_sync):
    a = Artist(name="TestArtist", slug="testartist")
    db_sync.add(a)
    db_sync.flush()
    return a


def test_find_or_create_source_creates_then_returns_existing(importer, artist_row):
    s1 = importer._find_or_create_source(
        artist_id=artist_row.id, platform="patreon",
        url="https://www.patreon.com/posts/test-1",
    )
    s2 = importer._find_or_create_source(
        artist_id=artist_row.id, platform="patreon",
        url="https://www.patreon.com/posts/test-1",
    )
    assert s1.id == s2.id


def test_find_or_create_source_distinct_urls_yield_distinct_rows(
    importer, artist_row,
):
    a = importer._find_or_create_source(
        artist_id=artist_row.id, platform="patreon",
        url="https://www.patreon.com/posts/a",
    )
    b = importer._find_or_create_source(
        artist_id=artist_row.id, platform="patreon",
        url="https://www.patreon.com/posts/b",
    )
    assert a.id != b.id


def test_find_or_create_post_idempotent(importer, artist_row, db_sync):
    src = importer._find_or_create_source(
        artist_id=artist_row.id, platform="patreon",
        url="https://www.patreon.com/posts/post-test",
    )
    p1 = importer._find_or_create_post(
        source_id=src.id, external_post_id="ext-001",
    )
    p2 = importer._find_or_create_post(
        source_id=src.id, external_post_id="ext-001",
    )
    assert p1.id == p2.id


def test_find_or_create_source_recovers_from_integrity_error(
    importer, artist_row, db_sync, monkeypatch,
):
    """Simulate the race: another worker has already inserted a Source row
    matching our (artist_id, platform, url) just before our flush would
    have. Our flush raises IntegrityError; the helper rolls back the
    savepoint and re-selects, returning the row the concurrent op created.
    """
    canonical_url = "https://www.patreon.com/posts/race-141226276"
    pre_existing = Source(
        artist_id=artist_row.id, platform="patreon", url=canonical_url,
    )
    db_sync.add(pre_existing)
    db_sync.flush()

    # Force a fresh select within the helper to MISS the existing row by
    # detaching it from the identity map; SQLAlchemy's first-level cache
    # would otherwise return the pre_existing row immediately.
    # Easier: monkeypatch the FIRST select inside the helper to return
    # None on first call, real result on subsequent. We do that by
    # patching session.execute with a single-shot wrapper.
    real_execute = db_sync.execute
    skip_count = [0]

    def execute_with_first_select_miss(stmt, *args, **kwargs):
        # Strip-down heuristic: the first SELECT issued by the helper is
        # the existence check. Force it to return a "no row" result.
        result = real_execute(stmt, *args, **kwargs)
        if skip_count[0] == 0:
            skip_count[0] += 1
            # Wrap result so .scalar_one_or_none() returns None for this
            # one call, then unwrap on subsequent uses.
            class _ForcedMiss:
                def scalar_one_or_none(self):
                    return None

                def scalar_one(self):
                    return result.scalar_one()

                def __getattr__(self, name):
                    return getattr(result, name)
            return _ForcedMiss()
        return result

    monkeypatch.setattr(db_sync, "execute", execute_with_first_select_miss)

    recovered = importer._find_or_create_source(
        artist_id=artist_row.id, platform="patreon", url=canonical_url,
    )
    assert recovered.id == pre_existing.id


def test_source_for_sidecar_reuses_existing_subscription(
    importer, artist_row, db_sync,
):
    """The filesystem-import sidecar resolver should attach to whatever
    Source already exists for (artist, platform) — the canonical subscription
    Source — regardless of its URL. Without this, every imported post
    spawned its own Source row.
    """
    canonical = Source(
        artist_id=artist_row.id, platform="patreon",
        url="https://www.patreon.com/cw/testartist", enabled=True,
    )
    db_sync.add(canonical)
    db_sync.flush()

    resolved = importer._source_for_sidecar(
        artist_id=artist_row.id, platform="patreon",
        artist_slug=artist_row.slug,
    )
    assert resolved.id == canonical.id


def test_source_for_sidecar_creates_synthetic_anchor_when_none_exists(
    importer, artist_row, db_sync,
):
    """No subscription Source for this (artist, platform) yet. The helper
    creates one synthetic anchor (enabled=False, url='sidecar:<plat>:<slug>')
    so subsequent imports reuse it instead of spawning per-post Sources.
    """
    resolved = importer._source_for_sidecar(
        artist_id=artist_row.id, platform="pixiv",
        artist_slug=artist_row.slug,
    )
    assert resolved.url == f"sidecar:pixiv:{artist_row.slug}"
    assert resolved.enabled is False
    assert resolved.artist_id == artist_row.id
    assert resolved.platform == "pixiv"

    # Second call returns the same row (no new Source spawned).
    again = importer._source_for_sidecar(
        artist_id=artist_row.id, platform="pixiv",
        artist_slug=artist_row.slug,
    )
    assert again.id == resolved.id


def test_source_for_sidecar_distinct_platforms_distinct_anchors(
    importer, artist_row, db_sync,
):
    """One synthetic anchor per (artist, platform). Different platforms get
    different anchors even when no campaign Source exists for either.
    """
    p = importer._source_for_sidecar(
        artist_id=artist_row.id, platform="patreon",
        artist_slug=artist_row.slug,
    )
    x = importer._source_for_sidecar(
        artist_id=artist_row.id, platform="pixiv",
        artist_slug=artist_row.slug,
    )
    assert p.id != x.id
    assert p.platform == "patreon"
    assert x.platform == "pixiv"