From 72181745f6316f76420b347e2b6f366c15eea2f3 Mon Sep 17 00:00:00 2001 From: Zyntro Date: Sun, 30 Aug 2026 00:06:34 +0700 Subject: [PATCH] Create conftest.py Signed-off-by: Zyntro --- conftest.py | 53 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 conftest.py diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..e1ade75 --- /dev/null +++ b/conftest.py @@ -0,0 +1,53 @@ +import asyncio +from typing import AsyncGenerator +import pytest +import pytest_asyncio +from httpx import ASGITransport, AsyncClient +from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker +from sqlmodel import SQLModel +from sqlmodel.ext.asyncio.session import AsyncSession + +from app.main import app +from app.core.database import get_db_session + +# SQLite in-memory engine for fast local testing (or test PostgreSQL instance) +TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:" + +test_engine = create_async_engine(TEST_DATABASE_URL, echo=False) +TestingSessionLocal = async_sessionmaker( + test_engine, class_=AsyncSession, expire_on_commit=False +) + +@pytest.fixture(scope="session") +def event_loop(): + """Create an instance of the default event loop for the test session.""" + loop = asyncio.get_event_loop_policy().new_event_loop() + yield loop + loop.close() + +@pytest_asyncio.fixture(scope="function") +async def db_session() -> AsyncGenerator[AsyncSession, None]: + """Provides a clean database session per test function.""" + async with test_engine.begin() as conn: + await conn.run_sync(SQLModel.metadata.create_all) + + async with TestingSessionLocal() as session: + yield session + + async with test_engine.begin() as conn: + await conn.run_sync(SQLModel.metadata.drop_all) + +@pytest_asyncio.fixture(scope="function") +async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]: + """Provides an AsyncClient bound to the FastAPI app with DB dependency override.""" + async def _get_test_db(): + yield db_session + + app.dependency_overrides[get_db_session] = _get_test_db + + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://testserver" + ) as ac: + yield ac + + app.dependency_overrides.clear()