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