AsyncClient ignores the startup and shutdown events

[zhukovgreen]

This is very close to #1072, but for AsyncClient.

The problem is that AsyncClient ignores the startup and shutdown events.

To Reproduce

Steps to reproduce the behavior with a minimum self-contained file.

Replace each part with your own scenario:

• Create a main.py file with:

from fastapi import FastAPI, Request

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}


async def add_test_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Test-Header"] = 'Hello'
    return response


@app.on_event("startup")
def setup():
    # example code that runs on startup
    global add_test_header
    print('executing startup!!')
    add_test_header = app.middleware("http")(add_test_header)

• Create a test_startup.py file with:

import pytest

from httpx import AsyncClient

from main import app


@pytest.fixture()
async def client():
    async with AsyncClient(app=app, base_url="http://test") as 
        yield client


@pytest.mark.asyncio
async def test_read_main(client):
    response = await client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}
    assert 'X-Test-Header' in response.headers

Run pytest -s -v test_startup.py
You will see an AssertionError for the X-Test-Header not being there
The test should pass

So far I am using a hack to make it work:

@pytest.fixture()
async def client():
    """Test client pytest fixture.

    Example:
        >>> from httpx import Response
        >>>
        >>>
        >>> @pytest.mark.asyncio
        >>> async def test_health_check(client):
        >>>    resp: Response = await client.get("/health_check")
        >>>    assert resp.status_code == 200

    """
    app = build_app()
    async with AsyncClient(app=app, base_url="http://test") as client:
        await connect_to_db(app)
        yield client
    await db_teardown(app)

but probably some solution or a note in the docs would compliment the project

[Kludex]

If a minimal example is provided, others can check. 🎉

[zhukovgreen]

@Kludex pls take a look one more time

[ArcLightSlavik]

@zhukovgreen
You're using httpx and it's AsyncClient which i don't believe supports test?
You can use pip install async_asgi_testclient

Which will look like this:

import pytest

from main import app
from async_asgi_testclient import TestClient


@pytest.fixture
async def client():
    async with TestClient(app) as client:
        yield client


@pytest.mark.asyncio
async def test_read_main(client):
    response = await client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}
    assert 'X-Test-Header' in response.headers

[Kludex]

@ArcLightSlavik It's the recommended solution on our documentation.

[ArcLightSlavik]

Ah, didn't see 🙂
What i did can be used as another workaround i guess ¯_(ツ)_/¯

[GriffGeorge]

More info in: encode/httpx#350

[zhukovgreen]

@ArcLightSlavik thank you. I believe async test client should support app lifecycle.

[blazewicz]

It's all discussed in encode/httpx#350 and related issues, but I'm gonna leave here a short summary.

As described in encode/httpx#1441, app lifecycle managemenent won't be added to HTTPX's AsyncClient because its considered out of its scope.

Suggested solution for original problem is to use LifespanManager from asgi-lifespan in pair with AsyncClient.

So test_startup.py could be changed to:

import pytest

from asgi_lifespan import LifespanManager
from httpx import AsyncClient

from main import app


@pytest.fixture()
async def client():
    async with AsyncClient(app=app, base_url="http://test") as client, LifespanManager(app):
        yield client


@pytest.mark.asyncio
async def test_read_main(client):
    response = await client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}
    assert 'X-Test-Header' in response.headers

I think it should be at least mentioned in "Testing Events: startup - shutdown" section of FastAPI docs.

[Kludex]

I agree with @blazewicz . Does someone want to give it a try?

[kkolur]

@blazewicz Do you happen to have an example of how to access the application instance using AsyncClient and testing it?

For example, this is my startup event

def startup_handler(app: FastAPI) -> Callable:
    def startup() -> None:
        logger.info("Running startup handler.")
        app.state.model = Model()

    return startup

app.add_event_handler("startup", startup_handler(app))

And I am trying to test like so but AsyncClient does not store the app instance like FastAPI's TestClient:

@pytest.mark.asyncio
class TestEvents:
    async def test_startup(self, test_client: AsyncClient) -> None:
        res = await test_client.get("/health")
        assert res.status_code == 200
        assert isinstance(test_client.app.state.model, Model)

[j-carson]

@ArcLightSlavik It's the recommended solution on our documentation.

Since AsyncClient is the documented async test procedure, and it is missing this commonly-used function, would it be possible to document this and maybe suggest a preferred work-around? Thanks.

[j-carson]

Now that my tests are running, I will suggest that the so-called hack suggested by the original poster may in fact be desired best practice. This is because, once I got the first test running under pytest-asyncio I hit another round of problems trying to get multiple tests to run until I realized something else was going on besides just startup/shutdown confusion:

Like all of pytest, pytest-asyncio is built around the concept of running individually isolated unit tests. Thus, it creates a new event loop for each unit test. This can be a good thing, you'll be able to isolate xyz was never awaited errors to one unit test. But it can also lead to confusing results if your test tries to reuse the app variable as in from main import app: I was trying to run a test that was marked with parametrize and no matter what I did, only one iteration would run and then I'd started getting weird errors about event loops. To be fully isolated unit tests, you have to create a new FastAPI instance each time as the OP did with their build_app function.

Here is how I refactored my code:

# This code block is my refactored main.py
def app_factory():
   myapp = FastAPI()
   myapp.include_router(module_a.router)
   myapp.include_router(module_b.router)
   return myapp

async def app_startup(app):
    pass # startup goes here

async def app_shutdown(app):
    pass # shutdown goes here

app = app_factory()
@app.on_event("startup")
async def startup():
    await app_startup(app)

@app.on_event("shutdown")
async def shutdown():
    await app_shutdown(app)

Then, in my tests, I have a fixture that creates a brand new FastAPI instance for each test function so that nothing gets confused by all the new event loops being created and destroyed under the covers by pytest-asyncio.

# this code block goes in conftest.py and simultaneously solves both the startup/shutdown doesn't run problem
# as well as my issues with running parametrized tests

from main import app_factory, app_startup, app_shutdown

@pytest.fixture
async def app():
     app = app_factory()
     await app_startup(app)
     yield app
     await app_shutdown(app)

And here's a stub for a test:

@pytest.mark.parametrize("param_a,param_b", [[1,2],[3,4],[5,6]])
@pytest.mark.asyncio
async def test_module_a(
    app,
    param_a,
    param_b
):
    async with AsyncClient(app=app, base_url="http://test") as ac:
        params = {}
        params["param_a"] = param_a
        params["param_b"] = param_b
        response = await ac.get("/module_a/some_api", params=params)
        assert response.status_code == 200  

Edit to add: If you don't mind having your event_loop stick around between tests, this issue shows how to prevent pytest-asyncio from creating new loops during the test: #2006

Edited to patch up the pseudo-code a bit more.

[andrew-chang-dewitt]

I agree with @blazewicz . Does someone want to give it a try?

I just created #4167 to begin addressing this in the docs. Anyone wanna take a look and give me some feedback on how I could improve the warning I added? I wrote the warning assuming that the LifespanManager context manager is the recommended solution.

[j-carson]

I agree with @blazewicz . Does someone want to give it a try?

I just created #4167 to begin addressing this in the docs. Anyone wanna take a look and give me some feedback on how I could improve the warning I added? I wrote the warning assuming that the LifespanManager context manager is the recommended solution.

Rather than just a warning, make it a documentation section with code snippets. That would make it much easier to give it a try.

[PidgeyBE]

For me, the following construction works:

    async with AsyncClient(app=app) as client:
        await app.router.startup()
         ...
        await app.router.shutdown()

[j-carson]

For me, the following construction works:

    async with AsyncClient(app=app) as client:
        await app.router.startup()
         ...
        await app.router.shutdown()

The problem with that is your test requires more intimate knowledge of where the critical startup shutdown code is. By isolating app setup in a fixture, I don't have to put those "awaits" in each test when I create the AsyncClient. Furthermore, by making the fixture call the app_startup and app_shutdown functions defined in my application, the fixture itself is further isolated from the details of the implementation.

[PidgeyBE]

For me, the following construction works:

    async with AsyncClient(app=app) as client:
        await app.router.startup()
         ...
        await app.router.shutdown()

The problem with that is your test requires more intimate knowledge of where the critical startup shutdown code is. By isolating app setup in a fixture, I don't have to put those "awaits" in each test when I create the AsyncClient. Furthermore, by making the fixture call the app_startup and app_shutdown functions defined in my application, the fixture itself is further isolated from the details of the implementation.

Yes, the code is isolated in a fixture :)

@pytest.fixture(scope="session")
@pytest.mark.asyncio
async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
    async with AsyncClient(app=hub_app, base_url="http://test") as client:
        await app.router.startup()
        yield client  # type: ignore
        await app.router.shutdown()

app.router.startup() is the same function that's being called if your run FastAPI in production. Same for app.router.shutdown().

[j-carson]

Inside a fixture makes more sense! But does your solution only work with scope="session"? If I can use it for smaller scopes as well, then it seems to be a simpler solution than what I did.

[PidgeyBE]

Inside a fixture makes more sense! But does your solution only work with scope="session"? If I can use it for smaller scopes as well, then it seems to be a simpler solution than what I did.

The scope doesn't matter... I prefer bigger scopes because it makes tests faster (in this case using session instead of function makes my tests run almost 10x faster...).

[Scowley4]

@j-carson: Then, in my tests, I have a fixture that creates a brand new FastAPI instance for each test function so that nothing gets confused by all the new event loops being created and destroyed under the covers by pytest-asyncio.

Was getting the error

TypeError: 'async_generator' object is not callable

Depending on your options with pytest-asyncio, you will want to declare those fixtures as @pytest_asyncio.fixture.
https://github.com/pytest-dev/pytest-asyncio#async-fixtures

You don't need to worry about this if you have set asuyncio_mode = auto in pytest.ini or similar.