"""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::` 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 == []