| 123456789101112131415161718192021222324252627 |
- from __future__ import annotations
- from collections.abc import AsyncIterator
- from pathlib import Path
- import pytest_asyncio
- from sqlalchemy.ext.asyncio import (
- AsyncEngine,
- AsyncSession,
- async_sessionmaker,
- create_async_engine,
- )
- from script_build_host.infrastructure.legacy_tables import legacy_metadata
- from script_build_host.infrastructure.tables import metadata
- @pytest_asyncio.fixture
- async def database(
- tmp_path: Path,
- ) -> AsyncIterator[tuple[AsyncEngine, async_sessionmaker[AsyncSession]]]:
- engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
- async with engine.begin() as connection:
- await connection.run_sync(legacy_metadata.create_all)
- await connection.run_sync(metadata.create_all)
- yield engine, async_sessionmaker(engine, expire_on_commit=False)
- await engine.dispose()
|