feat(fc2a): add TagService — find-or-create, autocomplete, image association
Validates kind/fandom rules at the service layer: fandom_id only allowed for kind='character', and must reference an existing kind='fandom' tag. Autocomplete ranking: exact match > prefix match > substring, tie-broken by image_count descending. Image-tag association uses INSERT ON CONFLICT DO NOTHING for idempotent re-tagging. conftest.py adds a transactional AsyncSession fixture; each test rolls back so they don't pollute each other. Also includes a sync Session fixture (db_sync) for the Importer tests in Task 5. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,75 @@
|
||||
"""Shared pytest fixtures.
|
||||
|
||||
The async db fixture provides an AsyncSession bound to a transaction that
|
||||
gets rolled back after each test. CI provisions a real Postgres + pgvector
|
||||
(see .forgejo/workflows/ci.yml), so tests exercise the actual schema and
|
||||
migration code paths.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.asyncio import (
|
||||
AsyncSession,
|
||||
async_sessionmaker,
|
||||
create_async_engine,
|
||||
)
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
|
||||
def _async_database_url() -> str:
|
||||
user = os.environ.get("DB_USER", "fabledcurator")
|
||||
password = os.environ["DB_PASSWORD"]
|
||||
host = os.environ.get("DB_HOST", "postgres")
|
||||
port = os.environ.get("DB_PORT", "5432")
|
||||
name = os.environ.get("DB_NAME", "fabledcurator_test")
|
||||
return f"postgresql+asyncpg://{user}:{password}@{host}:{port}/{name}"
|
||||
|
||||
|
||||
def _sync_database_url() -> str:
|
||||
user = os.environ.get("DB_USER", "fabledcurator")
|
||||
password = os.environ["DB_PASSWORD"]
|
||||
host = os.environ.get("DB_HOST", "postgres")
|
||||
port = os.environ.get("DB_PORT", "5432")
|
||||
name = os.environ.get("DB_NAME", "fabledcurator_test")
|
||||
return f"postgresql+psycopg://{user}:{password}@{host}:{port}/{name}"
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def engine():
|
||||
e = create_async_engine(_async_database_url(), future=True)
|
||||
yield e
|
||||
await e.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db(engine) -> AsyncSession:
|
||||
Session = async_sessionmaker(engine, expire_on_commit=False)
|
||||
async with Session() as session:
|
||||
await session.begin()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
await session.rollback()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sync_engine():
|
||||
e = create_engine(_sync_database_url(), future=True)
|
||||
yield e
|
||||
e.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_sync(sync_engine):
|
||||
"""Synchronous Session bound to a savepoint — used by Importer tests
|
||||
(the Importer is sync-only by design)."""
|
||||
SyncSession = sessionmaker(sync_engine, expire_on_commit=False)
|
||||
with SyncSession() as session:
|
||||
session.begin()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.rollback()
|
||||
Reference in New Issue
Block a user