Why I can't use client: AsyncClinet in pytest

[NightSkySK]

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

[seifertm]

@NightSkySK Your issue is that you're using pytest-asyncio in strict mode, but you annotate your async fixtures with @pytest.fixture, rather than @pytest_asyncio.fixture.

Here is the reproducer I used to track down your issue:

from contextlib import asynccontextmanager
import pytest
import pytest_asyncio

class Client:
    async def get(self, s):
        print(s)


@asynccontextmanager
async def create_client():
    yield Client()

@pytest.mark.asyncio
async def test_create_client_inline():
    async with create_client() as client:
        await client.get("Hello")


@pytest_asyncio.fixture # works
#@pytest.fixture # fails
async def client():
    async with create_client() as c:
        yield c

@pytest.mark.asyncio
async def test_create_client_fixture(client):
    await client.get("Hello")

[seifertm]

See also pytest-dev/pytest#10839

[NightSkySK]

@seifertm Thank you for you advice and reply.

I've managed to solve it with a bit different approach, However, I'm not sure which one is better. Your solution code looks very clean and it's probably the correct way do it.

import os
import sys
from typing import Any, AsyncGenerator, Generator

import pytest
from apis.base import api_router
from db.session import get_async_session
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from webapp.base import api_router as web_app_router

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

TEST_DATABASE_URL = "sqlite+aiosqlite:///./test_app.sqlite"


def start_application():
    app = FastAPI()
    app.include_router(api_router)
    app.include_router(web_app_router)
    return app


async def get_db_test() -> AsyncGenerator[AsyncSession, None]:
    try:
        engine_test = create_async_engine(TEST_DATABASE_URL, echo=True)
        async_session_maker = sessionmaker(
            engine_test, class_=AsyncSession, expire_on_commit=False
        )
        test_session = async_session_maker()
        yield test_session
    finally:
        await test_session.close()

@pytest.fixture(scope="function")
async def client() -> Generator[TestClient, Any, None]:
    app = start_application()
    app.dependency_overrides[get_async_session] = get_db_test
    with TestClient(app) as client:
        yield client

and than test is like this:

import pytest
import logging


DEBUG = True

if DEBUG:
    logger = logging.getLogger(__name__)


@pytest.mark.asyncio
async def test_about(client):
    async for ac in client:
        response = ac.get("/about")

    if DEBUG:
        logger.info(response)
    assert response.status_code == 200