bvandeusen
f6c7117231
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>
76 lines
2.1 KiB
Python
76 lines
2.1 KiB
Python
"""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()
|