| """ |
| Pytest configuration and global fixtures. |
| Initializes a temporary in-memory/file-based SQLite database for unit & integration tests. |
| """ |
| from __future__ import annotations |
|
|
| import asyncio |
| from typing import AsyncGenerator, Generator |
| import pytest |
| from httpx import ASGITransport, AsyncClient |
| from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker |
|
|
| import sys |
| import os |
|
|
| |
| base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) |
| sys.path.insert(0, os.path.join(base_dir, "backend")) |
| sys.path.insert(0, base_dir) |
| for extra_path in ["/app", "/agents", "/services"]: |
| if os.path.exists(extra_path) and extra_path not in sys.path: |
| sys.path.insert(0, extra_path) |
|
|
| from app.core.config import settings |
| from app.core.database import Base, get_db as db_get_db |
| from app.core.dependencies import get_db as dep_get_db |
| from app.main import app |
|
|
| |
| TEST_DB_URL = "sqlite+aiosqlite:///:memory:" |
|
|
| test_engine = create_async_engine( |
| TEST_DB_URL, |
| connect_args={"check_same_thread": False}, |
| ) |
|
|
| TestAsyncSessionLocal = async_sessionmaker( |
| bind=test_engine, |
| class_=AsyncSession, |
| expire_on_commit=False, |
| autocommit=False, |
| autoflush=False, |
| ) |
|
|
|
|
| @pytest.fixture(scope="session", autouse=True) |
| async def init_test_db(): |
| """Create all tables in the test database on session startup.""" |
| async with test_engine.begin() as conn: |
| await conn.run_sync(Base.metadata.create_all) |
| yield |
| async with test_engine.begin() as conn: |
| await conn.run_sync(Base.metadata.drop_all) |
| await test_engine.dispose() |
|
|
|
|
| @pytest.fixture |
| async def db_session() -> AsyncGenerator[AsyncSession, None]: |
| """Provide an isolated database session per test with rollback safety.""" |
| async with TestAsyncSessionLocal() as session: |
| try: |
| yield session |
| finally: |
| await session.rollback() |
| await session.close() |
|
|
|
|
| @pytest.fixture |
| async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]: |
| """Provide an HTTPX AsyncClient with database session overridden.""" |
| async def override_get_db(): |
| yield db_session |
|
|
| app.dependency_overrides[db_get_db] = override_get_db |
| app.dependency_overrides[dep_get_db] = override_get_db |
| transport = ASGITransport(app=app) |
| async with AsyncClient(transport=transport, base_url="http://test") as ac: |
| yield ac |
| app.dependency_overrides.clear() |
|
|
