conftest.py 864 B

123456789101112131415161718192021222324252627
  1. from __future__ import annotations
  2. from collections.abc import AsyncIterator
  3. from pathlib import Path
  4. import pytest_asyncio
  5. from sqlalchemy.ext.asyncio import (
  6. AsyncEngine,
  7. AsyncSession,
  8. async_sessionmaker,
  9. create_async_engine,
  10. )
  11. from script_build_host.infrastructure.legacy_tables import legacy_metadata
  12. from script_build_host.infrastructure.tables import metadata
  13. @pytest_asyncio.fixture
  14. async def database(
  15. tmp_path: Path,
  16. ) -> AsyncIterator[tuple[AsyncEngine, async_sessionmaker[AsyncSession]]]:
  17. engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
  18. async with engine.begin() as connection:
  19. await connection.run_sync(legacy_metadata.create_all)
  20. await connection.run_sync(metadata.create_all)
  21. yield engine, async_sessionmaker(engine, expire_on_commit=False)
  22. await engine.dispose()