Why I can't use client: AsyncClinet in pytest?

136 Views Asked by At

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
0

There are 0 best solutions below