conftest.py
python
| 1 | """Pytest configuration and fixtures.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import logging |
| 5 | from collections.abc import AsyncGenerator, Generator |
| 6 | import pytest |
| 7 | import pytest_asyncio |
| 8 | from httpx import AsyncClient, ASGITransport |
| 9 | from sqlalchemy.ext.asyncio import ( |
| 10 | AsyncSession, |
| 11 | async_sessionmaker, |
| 12 | create_async_engine, |
| 13 | ) |
| 14 | from sqlalchemy.pool import StaticPool |
| 15 | |
| 16 | from musehub.db import database |
| 17 | from musehub.db.database import Base, get_db |
| 18 | from musehub.db.models import User |
| 19 | from musehub.main import app |
| 20 | |
| 21 | |
| 22 | def pytest_configure(config: pytest.Config) -> None: |
| 23 | |
| 24 | """Ensure asyncio_mode is auto so async fixtures work (e.g. in Docker when pyproject not in cwd).""" |
| 25 | if hasattr(config.option, "asyncio_mode") and config.option.asyncio_mode is None: |
| 26 | config.option.asyncio_mode = "auto" |
| 27 | logging.getLogger("httpcore").setLevel(logging.CRITICAL) |
| 28 | |
| 29 | |
| 30 | @pytest.fixture |
| 31 | def anyio_backend() -> str: |
| 32 | return "asyncio" |
| 33 | |
| 34 | |
| 35 | @pytest.fixture(autouse=True) |
| 36 | def _reset_variation_store() -> Generator[None, None, None]: |
| 37 | """Reset the singleton VariationStore between tests to prevent cross-test pollution.""" |
| 38 | yield |
| 39 | from musehub.variation.storage.variation_store import reset_variation_store |
| 40 | reset_variation_store() |
| 41 | |
| 42 | |
| 43 | @pytest_asyncio.fixture |
| 44 | async def db_session() -> AsyncGenerator[AsyncSession, None]: |
| 45 | """Create an in-memory test database session.""" |
| 46 | engine = create_async_engine( |
| 47 | "sqlite+aiosqlite:///:memory:", |
| 48 | connect_args={"check_same_thread": False}, |
| 49 | poolclass=StaticPool, |
| 50 | ) |
| 51 | async with engine.begin() as conn: |
| 52 | await conn.run_sync(Base.metadata.create_all) |
| 53 | |
| 54 | async_session_factory = async_sessionmaker( |
| 55 | bind=engine, |
| 56 | class_=AsyncSession, |
| 57 | expire_on_commit=False, |
| 58 | ) |
| 59 | old_engine = database._engine |
| 60 | old_factory = database._async_session_factory |
| 61 | database._engine = engine |
| 62 | database._async_session_factory = async_session_factory |
| 63 | try: |
| 64 | async with async_session_factory() as session: |
| 65 | async def override_get_db() -> AsyncGenerator[AsyncSession, None]: |
| 66 | yield session |
| 67 | app.dependency_overrides[get_db] = override_get_db |
| 68 | yield session |
| 69 | app.dependency_overrides.clear() |
| 70 | finally: |
| 71 | database._engine = old_engine |
| 72 | database._async_session_factory = old_factory |
| 73 | async with engine.begin() as conn: |
| 74 | await conn.run_sync(Base.metadata.drop_all) |
| 75 | await engine.dispose() |
| 76 | |
| 77 | |
| 78 | @pytest_asyncio.fixture |
| 79 | async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]: |
| 80 | |
| 81 | """Create an async test client. Depends on db_session so auth revocation check uses test DB.""" |
| 82 | transport = ASGITransport(app=app) |
| 83 | async with AsyncClient(transport=transport, base_url="http://test") as ac: |
| 84 | yield ac |
| 85 | |
| 86 | |
| 87 | # ----------------------------------------------------------------------------- |
| 88 | # Auth fixtures for API contract and integration tests |
| 89 | # ----------------------------------------------------------------------------- |
| 90 | |
| 91 | @pytest_asyncio.fixture |
| 92 | async def test_user(db_session: AsyncSession) -> User: |
| 93 | |
| 94 | """Create a test user (for authenticated route tests).""" |
| 95 | user = User( |
| 96 | id="550e8400-e29b-41d4-a716-446655440000", |
| 97 | ) |
| 98 | db_session.add(user) |
| 99 | await db_session.commit() |
| 100 | await db_session.refresh(user) |
| 101 | return user |
| 102 | |
| 103 | |
| 104 | @pytest.fixture |
| 105 | def auth_token(test_user: User) -> str: |
| 106 | |
| 107 | """JWT for test_user (1 hour).""" |
| 108 | from musehub.auth.tokens import create_access_token |
| 109 | return create_access_token(user_id=test_user.id, expires_hours=1) |
| 110 | |
| 111 | |
| 112 | @pytest.fixture |
| 113 | def auth_headers(auth_token: str) -> dict[str, str]: |
| 114 | |
| 115 | """Headers with Bearer token and JSON content type.""" |
| 116 | return { |
| 117 | "Authorization": f"Bearer {auth_token}", |
| 118 | "Content-Type": "application/json", |
| 119 | } |