FastApi

1
import asyncio
2
from typing import AsyncGenerator
3

4
import pytest
5
from fastapi.testclient import TestClient
6
from httpx import AsyncClient
7
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
8
from sqlalchemy.orm import sessionmaker
9
from sqlalchemy.pool import NullPool
10

11
from database import get_async_session
12
from src import metadata
13
from src.config import (DB_HOST_TEST, DB_NAME_TEST, DB_PASS_TEST, DB_PORT_TEST,
14
                        DB_USER_TEST)
15
from src.main import app
16

17
# DATABASE
18
DATABASE_URL_TEST = f"postgresql+asyncpg://{DB_USER_TEST}:{DB_PASS_TEST}@{DB_HOST_TEST}:{DB_PORT_TEST}/{DB_NAME_TEST}"
19

20
engine_test = create_async_engine(DATABASE_URL_TEST, poolclass=NullPool)
21
async_session_maker = sessionmaker(engine_test, class_=AsyncSession, expire_on_commit=False)
22
metadata.bind = engine_test
23

24
async def override_get_async_session() -> AsyncGenerator[AsyncSession, None]:
25
    async with async_session_maker() as session:
26
        yield session
27

28
app.dependency_overrides[get_async_session] = override_get_async_session
29

30
@pytest.fixture(autouse=True, scope='session')
31
async def prepare_database():
32
    async with engine_test.begin() as conn:
33
        await conn.run_sync(metadata.create_all)
34
    yield
35
    async with engine_test.begin() as conn:
36
        await conn.run_sync(metadata.drop_all)
37

38
# SETUP
39
@pytest.fixture(scope='session')
40
def event_loop(request):
41
    """Create an instance of the default event loop for each test case."""
42
    loop = asyncio.get_event_loop_policy().new_event_loop()
43
    yield loop
44
    loop.close()
45

46
client = TestClient(app)
47

48
@pytest.fixture(scope="session")
49
async def ac() -> AsyncGenerator[AsyncClient, None]:
50
    async with AsyncClient(app=app, base_url="http://test") as ac:
51
        yield ac
52