22 lines
680 B
Python
22 lines
680 B
Python
from collections.abc import AsyncIterator
|
|
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from app.db.base import Base
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def session_factory(tmp_path) -> AsyncIterator[async_sessionmaker[AsyncSession]]:
|
|
database_path = tmp_path / "test.db"
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{database_path}")
|
|
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(Base.metadata.create_all)
|
|
|
|
factory = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
|
|
try:
|
|
yield factory
|
|
finally:
|
|
await engine.dispose()
|