Spaces:
Sleeping
Sleeping
File size: 2,586 Bytes
868b252 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 |
from typing import Any, AsyncGenerator
import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from reworkd_platform.db.dependencies import get_db_session
from reworkd_platform.db.utils import create_database, drop_database
from reworkd_platform.settings import settings
from reworkd_platform.web.application import get_app
@pytest.fixture(scope="session")
def anyio_backend() -> str:
"""
Backend for anyio pytest plugin.
:return: backend name.
"""
return "asyncio"
@pytest.fixture(scope="session")
async def _engine() -> AsyncGenerator[AsyncEngine, None]:
"""
Create engine and databases.
:yield: new engine.
"""
from reworkd_platform.db.meta import meta # noqa: WPS433
from reworkd_platform.db.models import load_all_models # noqa: WPS433
load_all_models()
await create_database()
engine = create_async_engine(str(settings.db_url))
async with engine.begin() as conn:
await conn.run_sync(meta.create_all)
try:
yield engine
finally:
await engine.dispose()
await drop_database()
@pytest.fixture
async def dbsession(
_engine: AsyncEngine,
) -> AsyncGenerator[AsyncSession, None]:
"""
Get session to database.
Fixture that returns a SQLAlchemy session with a SAVEPOINT, and the rollback to it
after the test completes.
:param _engine: current engine.
:yields: async session.
"""
connection = await _engine.connect()
trans = await connection.begin()
session_maker = async_sessionmaker(
connection,
expire_on_commit=False,
)
session = session_maker()
try:
yield session
finally:
await session.close()
await trans.rollback()
await connection.close()
@pytest.fixture
def fastapi_app(dbsession: AsyncSession) -> FastAPI:
"""
Fixture for creating FastAPI app.
:return: fastapi app with mocked dependencies.
"""
application = get_app()
application.dependency_overrides[get_db_session] = lambda: dbsession
return application # noqa: WPS331
@pytest.fixture
async def client(
fastapi_app: FastAPI, anyio_backend: Any
) -> AsyncGenerator[AsyncClient, None]:
"""
Fixture that creates client for requesting server.
:param fastapi_app: the application.
:yield: client for the app.
"""
async with AsyncClient(app=fastapi_app, base_url="http://test") as ac:
yield ac
|