Starlette: [Question] Async test client

[ryankask]

I'm developing an app that requires async setup and teardown during testing. For example, I insert data into a Redis database using aioredis. I'm using pytest-asyncio to mark async test functions.

The test client uses the synchronous requests API and an ASGI adapter to call the Starlette app. The setup I described above doesn't work with the adapter because it doesn't expect an event loop to be running.

I haven't worked with asyncio prior to using Starlette so I may be missing something obvious but do you have any suggestions for how to deal with this? I thought about more explicitly managing the event loop during testing but this seemed a bit tedious.

Another option was an async test client. I had a look at aiohttp and if a custom Connector could be used in place of the ASGI requests adapter but I didn't spend much time on it.

For now, I cobbled together an async test client that combines the request setup from requests.Session and starlette.testclient. _ASGIAdapter.

By the way, if there's a better place to ask questions, please let me know 😄

[tomchristie]

Okay, so...

The "lifespan" events are useful here. See https://www.starlette.io/events/

One option would be for you to add the setup teardown events to the app when in test mode, and then use with LifespanContext(app) in the test cases.

Alternatively if you don't want to register the events on the app itself you can use just the lifespan handler by itself, so...

lifespan = LifespanHandler()

# We should probably add a regular function-based interface too. `lifespan.add_event_handler('startup', func)
@lifespan.on_event('startup')  
async def startup():
    ...

@lifespan.on_event('cleanup')  
async def cleanup():
    ...

You're then able to use something like this in your test cases:

with LifetimeContext(lifetime):
    ...

It's a regular sync interface at that point, but it'll marshall bridging to the async calls for you.
It'd be interesting if we could look at how those would correspond to pytest fixtures.

[tomchristie]

Also, we may want to reorganise the docs here, so that LifespanContext gets included in the test section.

[ryankask]

Thanks for the tips. I can see how it's possible to use these features along with Starlette's test client.

Since I'm already using pytest-asyncio to test other parts of the app, I'm going to continue using the async test client to keep things consistent.

I'll close this issue since you've answered the original question.

[jared-mackey]

@ryankask Would you be willing to setup a package for your async test client code? I'd love to take a look and see if we could use it as well.

[ryankask]

@mackeyja92 Here's a gist: https://gist.github.com/ryankask/90c0dc155d3f95a52a1eaa0ac4ef63d0.

I needed to make it async all the way down to this call loop.run_until_complete(connection(receive, send)) deep in the adapter so a nested loop wasn't started.

To do this, I combined the ASGI adapter Tom wrote with requests' Session API. I removed some of the Session-related code that I didn't think was relevant but I may have removed or left in too much.

I use pytest with pytest-asyncio doing something like this

@pytest.fixture
def client():
    return StarletteTestClient(starlette_app)

@pytest.mark.asyncio
async def test_update_company(client):
    company = await CompanyStore.create(name="Great Software")
    data = {
        "name": "Awesome Software",
    }

    response = await client.patch(f"/companies/{company.id}", json=data)

    assert response.status_code == 200
    updated_company = await CompanyStore.get(company.id)
    assert updated_company.name == "Awesome Software"

All my app code is async so to test in same way I've always done (arrange, act, assert, this time asynchronously), I think I need an async test client. This way pytest-asyncio can manage the loop and I can use pytest naturally. Maybe I need to just change the way I test so hopefully someone points out a better way.

I thought it would be better to use aiohttp's equivalent of a requests' adapter but I didn't spend much time figuring it out.

[gjeusel]

For somebody else passing by in search of an AsyncTestClient: async-asgi-testclient made the trick for me.

[euri10]

For somebody else passing by in search of an AsyncTestClient: async-asgi-testclient made the trick for me.

you can also use httpx and the asgi-lifespan by @florimondmanca !

[gjeusel]

For somebody else passing by in search of an AsyncTestClient: async-asgi-testclient made the trick for me.

you can also use httpx and the asgi-lifespan by @florimondmanca !

httpx is much better ! Thx

[and-semakin]

I'll just put links here for future strangers.

• https://github.com/florimondmanca/asgi-lifespan#usage (see usage example here)
• https://pypi.org/project/asgi-lifespan/

Also awesome explanation about the async test client: https://fastapi.tiangolo.com/advanced/async-tests/ (it's about FastAPI but it still should be applicable to pure Starlette).

My example of async test client fixture:

# conftest.py
from asgi_lifespan import LifespanManager
from httpx import AsyncClient


@pytest.fixture()
async def client() -> Iterator[AsyncClient]:
    async with LifespanManager(app):
        async with AsyncClient(app=app, base_url="http://test") as ac:
            yield ac


# test_app.py
@pytest.mark.asyncio
async def test_endpoint(client: AsyncClient) -> None:
    response = await client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}