Oleg Korshunov 5 days ago
committed by GitHub
parent
commit
2865774186
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. 69
      docs/en/docs/advanced/async-tests.md
  2. 24
      docs_src/async_tests/main.py
  3. 17
      docs_src/async_tests/test_main.py

69
docs/en/docs/advanced/async-tests.md

@ -1,15 +1,20 @@
# Async Tests
# Async Tests with Lifespan Trigger
You have already seen how to test your **FastAPI** applications using the provided `TestClient`. Up to now, you have only seen how to write synchronous tests, without using `async` functions.
You’ve already seen how to test your **FastAPI** applications using `TestClient`. So far, you’ve primarily written synchronous tests using def functions.
Being able to use asynchronous functions in your tests could be useful, for example, when you're querying your database asynchronously. Imagine you want to test sending requests to your FastAPI application and then verify that your backend successfully wrote the correct data in the database, while using an async database library.
Let's look at how we can make that work.
However, asynchronous test functions can be useful, especially when querying your database asynchronously. For instance, you may want to test that sending requests to your **FastAPI** app successfully writes correct data to the database using an async library. Let’s explore how to achieve this.
## pytest.mark.anyio
If we want to call asynchronous functions in our tests, our test functions have to be asynchronous. AnyIO provides a neat plugin for this, that allows us to specify that some test functions are to be called asynchronously.
/// note
You can also use <a href="https://pytest-asyncio.readthedocs.io/en/latest/" class="external-link" target="_blank">PyTest asyncio</a>.
///
## HTTPX
Even if your **FastAPI** application uses normal `def` functions instead of `async def`, it is still an `async` application underneath.
@ -25,9 +30,9 @@ For a simple example, let's consider a file structure similar to the one describ
```
.
├── app
   ├── __init__.py
   ├── main.py
   └── test_main.py
├── __init__.py
├── main.py
└── test_main.py
```
The file `main.py` would have:
@ -54,46 +59,40 @@ $ pytest
## In Detail
The marker `@pytest.mark.anyio` tells pytest that this test function should be called asynchronously:
{* ../../docs_src/async_tests/test_main.py hl[7] *}
The lifespan function demonstrates how to manage the lifecycle of application-wide resources. During the app's lifespan, we open a resource (`some_state_open`) at startup and clean it up (`some_state_close`) during shutdown.
/// tip
We use **ASGITransport** from **HTTPX** to interact directly with the **FastAPI** app in an async test environment.
Note that the test function is now `async def` instead of just `def` as before when using the `TestClient`.
When testing **FastAPI** apps with a custom lifespan, it's critical to manually trigger it in the test context to ensure proper setup and teardown of resources.
///
If you observe issues with state initialization or teardown in your tests, ensure that the lifespan is correctly invoked, and verify the app's state before and after requests.
Then we can create an `AsyncClient` with the app, and send async requests to it, using `await`.
{* ../../docs_src/async_tests/test_main.py hl[9:12] *}
## Other Asynchronous Function Calls
This is the equivalent to:
As the testing function is now asynchronous, you can now also call (and await) other async functions apart from sending requests to your FastAPI application in your tests, exactly as you would call them anywhere else in your code.
```Python
response = client.get('/')
```
//// tip
...that we used to make our requests with the `TestClient`.
If you encounter a `RuntimeError: Task attached to a different loop` when integrating asynchronous function calls in your tests, you can override the default pytest event loop using the following fixture:
/// tip
```python
import asyncio
import pytest
from collections.abc import Generator
Note that we're using async/await with the new `AsyncClient` - the request is asynchronous.
@pytest.fixture(scope="function")
def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
"""Overrides pytest default function scoped event loop using asyncio.Runner"""
with asyncio.Runner() as runner:
yield runner.get_loop()
///
```
/// warning
/// note
If your application relies on lifespan events, the `AsyncClient` won't trigger these events. To ensure they are triggered, use `LifespanManager` from <a href="https://github.com/florimondmanca/asgi-lifespan#usage" class="external-link" target="_blank">florimondmanca/asgi-lifespan</a>.
This fixture uses `scope="function"`, which ensures a fresh event loop is created for each test. While this is a robust approach, especially for isolating test cases, configuring a session-scoped loop (`scope="session"`) can simplify setup in some scenarios, such as when working with shared resources like database connections. Choose the scope based on the specific requirements and complexity of your tests.
///
## Other Asynchronous Function Calls
As the testing function is now asynchronous, you can now also call (and `await`) other `async` functions apart from sending requests to your FastAPI application in your tests, exactly as you would call them anywhere else in your code.
/// tip
If you encounter a `RuntimeError: Task attached to a different loop` when integrating asynchronous function calls in your tests (e.g. when using <a href="https://stackoverflow.com/questions/41584243/runtimeerror-task-attached-to-a-different-loop" class="external-link" target="_blank">MongoDB's MotorClient</a>), remember to instantiate objects that need an event loop only within async functions, e.g. an `'@app.on_event("startup")` callback.
///
////

24
docs_src/async_tests/main.py

@ -1,8 +1,24 @@
from fastapi import FastAPI
from contextlib import asynccontextmanager
app = FastAPI()
from fastapi import Depends, FastAPI, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize a shared state when the app starts
app.state.some_state = "some_state_open"
yield
# Cleanup the shared state when the app shuts down
app.state.some_state = "some_state_close"
async def get_some_state(request: Request):
return request.app.state.some_state
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def root():
return {"message": "Tomato"}
async def root(some_state=Depends(get_some_state)):
return {"message": some_state}

17
docs_src/async_tests/test_main.py

@ -1,14 +1,21 @@
import pytest
from httpx import ASGITransport, AsyncClient
from .main import app
from .main import app, lifespan
@pytest.mark.anyio
async def test_root():
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
response = await ac.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Tomato"}
) as client:
async with lifespan(app=app):
# Send a GET request to the root endpoint
response = await client.get("/")
# Verify that the response is successful
assert response.status_code == 200
# Verify that the app's state is initialized
assert response.json() == {"message": "some_state_open"}
# Verify that the app's state is cleaned up
assert app.state.some_state == "some_state_close"

Loading…
Cancel
Save