Fastapi: AsyncClient ignores the startup and shutdown events

Created on 2 Sep 2020  Â·  6Comments  Â·  Source: tiangolo/fastapi

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

question

Most helpful comment

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

All 6 comments

If a minimal example is provided, others can check. :tada:

@Kludex pls take a look one more time

@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

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

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

Was this page helpful?
0 / 5 - 0 ratings