Here is my client function definition in conftest.py:
from typing import Any, AsyncGenerator, Callable
import os
import sys
import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# this is to include backend dir in sys.path so that we can import from db,main.py
from apis.base import api_router
from db.session import get_async_session
from db.base_class import Base
from core.config import settings
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test_app.db"
def start_application():
app = FastAPI()
app.include_router(api_router)
return app
@pytest.fixture(scope="session")
async def _engine() -> AsyncGenerator[AsyncEngine, None]:
"""
Create engine and databases.
:yield: new engine.
"""
engine = create_async_engine(
TEST_DATABASE_URL, connect_args={"check_same_thread": False}
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield engine
finally:
await engine.dispose()
await conn.run_sync(Base.metadata.drop_all)
@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 = start_application()
application.dependency_overrides[get_async_session] = lambda: dbsession
return application # noqa: WPS331
@pytest.fixture
async def client(fastapi_app: FastAPI) -> 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
Once I try to use client fixture I get following error msg:
@pytest.mark.asyncio
async def test_about(client):
# for ac in client:
> response = await client.get("/about")
E AttributeError: 'async_generator' object has no attribute 'get'
However once I call following test all works correctly:
import pytest
import logging
from httpx import AsyncClient
logger = logging.getLogger(__name__)
@pytest.mark.order(index=1)
@pytest.mark.asyncio
async def test_about(fastapi_app):
async with AsyncClient(app=fastapi_app, base_url="http://test") as ac:
response = await ac.get("/about")
logger.info(response)
assert response.status_code == 200
platform win32 -- Python 3.9.2, pytest-7.4.0, pluggy-1.3.0
rootdir: C:\project\api-www
configfile: pytest.ini
plugins: anyio-3.7.1, asyncio-0.21.1, order-1.1.0
asyncio: mode=strict