Sergey Vanyushkin
f9fff0302c
Some checks failed
ci/woodpecker/pr/pipeline Pipeline failed
- Rewrite setup_database fixture to use sync engine with alembic stamp - Fix E2E CI race condition (test-e2e depends on test-integration) - Remove ruff I rule to resolve isort conflict
56 lines
1.6 KiB
Python
56 lines
1.6 KiB
Python
import asyncio
|
|
import os
|
|
from collections.abc import AsyncGenerator
|
|
from typing import Generator
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from app.infrastructure.config import settings
|
|
from app.infrastructure.database.models import Base
|
|
|
|
|
|
def _sync_url(db_url: str) -> str:
|
|
return db_url.replace("+aiosqlite", "").replace("+asyncpg", "")
|
|
|
|
|
|
def _build_alembic_config(db_url: str) -> Config:
|
|
alembic_cfg = Config("alembic.ini")
|
|
alembic_cfg.set_main_option("sqlalchemy.url", db_url)
|
|
return alembic_cfg
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop() -> Generator[asyncio.AbstractEventLoop]:
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def setup_database() -> Generator[None]:
|
|
db_url = os.environ.get("DB_URL", settings.database_url)
|
|
sync_engine = create_engine(_sync_url(db_url))
|
|
Base.metadata.drop_all(sync_engine)
|
|
Base.metadata.create_all(sync_engine)
|
|
sync_engine.dispose()
|
|
|
|
alembic_cfg = _build_alembic_config(db_url)
|
|
command.stamp(alembic_cfg, "head")
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_session() -> AsyncGenerator[AsyncSession]:
|
|
db_url = os.environ.get("DB_URL", settings.database_url)
|
|
test_engine = create_async_engine(db_url)
|
|
session_factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
async with session_factory() as session:
|
|
yield session
|
|
|
|
await test_engine.dispose()
|