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",
        artist_id=artist_row.id,
    )
    p2 = importer._find_or_create_post(
        source_id=src.id, external_post_id="ext-001",
        artist_id=artist_row.id,
    )
    assert p1.id == p2.id
    assert p1.artist_id == artist_row.id


def test_find_or_create_post_idempotent_with_null_source(
    importer, artist_row, db_sync,
):
    """Post.source_id is nullable since alembic 0030 — filesystem-imported
    posts with no live subscription have NULL source_id. The partial
    unique index `uq_post_artist_external_id_null_source` on
    (artist_id, external_post_id) WHERE source_id IS NULL guards the
    dedup; the helper matches on the same (artist_id, external_post_id)
    keys when source_id is None."""
    p1 = importer._find_or_create_post(
        source_id=None, external_post_id="fs-001",
        artist_id=artist_row.id,
    )
    p2 = importer._find_or_create_post(
        source_id=None, external_post_id="fs-001",
        artist_id=artist_row.id,
    )
    assert p1.id == p2.id
    assert p1.source_id is None
    assert p1.artist_id == artist_row.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_lookup_source_for_sidecar_returns_existing_subscription(
    importer, artist_row, db_sync,
):
    """The sidecar-import Source lookup returns the existing Source for
    (artist, platform) when one exists — letting filesystem-imported
    content attach to the artist's real subscription instead of being
    orphaned.
    """
    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._lookup_source_for_sidecar(
        artist_id=artist_row.id, platform="patreon",
    )
    assert resolved is not None
    assert resolved.id == canonical.id


def test_lookup_source_for_sidecar_returns_none_when_no_subscription(
    importer, artist_row, db_sync,
):
    """Alembic 0030 made Post.source_id nullable; the importer no
    longer creates synthetic `sidecar:<platform>:<slug>` anchor rows
    when no real subscription exists. The lookup returns None and the
    caller carries that through as a NULL source_id on the new Post.
    """
    resolved = importer._lookup_source_for_sidecar(
        artist_id=artist_row.id, platform="pixiv",
    )
    assert resolved is None

    # Verify no Source row was created as a side effect.
    count = db_sync.execute(
        select(Source).where(
            Source.artist_id == artist_row.id, Source.platform == "pixiv",
        )
    ).all()
    assert count == []