diff --git a/docs/en/docs/advanced/async-tests.md b/docs/en/docs/advanced/async-tests.md index 8d6929222..256ef92a5 100644 --- a/docs/en/docs/advanced/async-tests.md +++ b/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 PyTest asyncio. + +/// + + ## 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 florimondmanca/asgi-lifespan. +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 MongoDB's MotorClient), remember to instantiate objects that need an event loop only within async functions, e.g. an `'@app.on_event("startup")` callback. - -/// +//// diff --git a/docs_src/async_tests/main.py b/docs_src/async_tests/main.py index 9594f859c..bf9a7b392 100644 --- a/docs_src/async_tests/main.py +++ b/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} diff --git a/docs_src/async_tests/test_main.py b/docs_src/async_tests/test_main.py index a57a31f7d..8aea2fd6a 100644 --- a/docs_src/async_tests/test_main.py +++ b/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"