feat(fc3a): migration 0010 — unique(source.artist_id, platform, url)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,32 @@
|
|||||||
|
"""fc3a: unique(source.artist_id, source.platform, source.url)
|
||||||
|
|
||||||
|
Revision ID: 0010
|
||||||
|
Revises: 0009
|
||||||
|
Create Date: 2026-05-20
|
||||||
|
|
||||||
|
Enforces FC-3a's dedup invariant at the DB level. No backfill — no
|
||||||
|
existing rows are expected to collide; if they do the migration will
|
||||||
|
fail loudly (intended).
|
||||||
|
"""
|
||||||
|
from typing import Sequence, Union
|
||||||
|
|
||||||
|
from alembic import op
|
||||||
|
|
||||||
|
revision: str = "0010"
|
||||||
|
down_revision: Union[str, None] = "0009"
|
||||||
|
branch_labels: Union[str, Sequence[str], None] = None
|
||||||
|
depends_on: Union[str, Sequence[str], None] = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade() -> None:
|
||||||
|
op.create_unique_constraint(
|
||||||
|
"uq_source_artist_platform_url",
|
||||||
|
"source",
|
||||||
|
["artist_id", "platform", "url"],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade() -> None:
|
||||||
|
op.drop_constraint(
|
||||||
|
"uq_source_artist_platform_url", "source", type_="unique"
|
||||||
|
)
|
||||||
@@ -0,0 +1,35 @@
|
|||||||
|
import pytest
|
||||||
|
from sqlalchemy.exc import IntegrityError
|
||||||
|
|
||||||
|
from backend.app.models import Artist, Source
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.integration
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_duplicate_artist_platform_url_rejected(db):
|
||||||
|
artist = Artist(name="Alice", slug="alice")
|
||||||
|
db.add(artist)
|
||||||
|
await db.flush()
|
||||||
|
db.add(Source(
|
||||||
|
artist_id=artist.id, platform="patreon",
|
||||||
|
url="https://patreon.com/alice", enabled=True,
|
||||||
|
))
|
||||||
|
await db.flush()
|
||||||
|
db.add(Source(
|
||||||
|
artist_id=artist.id, platform="patreon",
|
||||||
|
url="https://patreon.com/alice", enabled=True,
|
||||||
|
))
|
||||||
|
with pytest.raises(IntegrityError):
|
||||||
|
await db.flush()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_same_url_under_different_artist_ok(db):
|
||||||
|
a = Artist(name="A", slug="a")
|
||||||
|
b = Artist(name="B", slug="b")
|
||||||
|
db.add_all([a, b])
|
||||||
|
await db.flush()
|
||||||
|
db.add(Source(artist_id=a.id, platform="patreon", url="https://x/y", enabled=True))
|
||||||
|
db.add(Source(artist_id=b.id, platform="patreon", url="https://x/y", enabled=True))
|
||||||
|
await db.flush() # must NOT raise
|
||||||
Reference in New Issue
Block a user