From fe39ca1caab808edb0d1e65856bd1e2b5306b36b Mon Sep 17 00:00:00 2001 From: "oleg.korshunov" Date: Tue, 7 Jan 2025 13:57:44 +0300 Subject: [PATCH 1/4] update pytest asyncio test added lifespan and update docs --- docs/en/docs/advanced/async-tests.md | 67 ++++++++++++---------------- docs_src/async_tests/main.py | 24 ++++++++-- docs_src/async_tests/test_main.py | 19 +++++--- 3 files changed, 61 insertions(+), 49 deletions(-) diff --git a/docs/en/docs/advanced/async-tests.md b/docs/en/docs/advanced/async-tests.md index 8d6929222..108bdf5c6 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,32 @@ $ pytest ## In Detail -The marker `@pytest.mark.anyio` tells pytest that this test function should be called asynchronously: +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. -{* ../../docs_src/async_tests/test_main.py hl[7] *} +We use **ASGITransport** from **HTTPX** to interact directly with the **FastAPI** app in an async test environment. -/// tip - -Note that the test function is now `async def` instead of just `def` as before when using the `TestClient`. - -/// - -Then we can create an `AsyncClient` with the app, and send async requests to it, using `await`. +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. -{* ../../docs_src/async_tests/test_main.py hl[9:12] *} +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. -This is the equivalent to: -```Python -response = client.get('/') -``` +## Other Asynchronous Function Calls -...that we used to make our requests with the `TestClient`. +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 -Note that we're using async/await with the new `AsyncClient` - the request is asynchronous. - -/// - -/// warning +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: -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. +```python +@pytest.fixture(scope="session") +def event_loop() -> Generator[AbstractEventLoop, None, None]: + """Overrides pytest default function scoped event loop""" + policy = asyncio.get_event_loop_policy() + loop = policy.new_event_loop() + yield loop + loop.close() +``` /// -## 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..5815d9dd3 100644 --- a/docs_src/async_tests/test_main.py +++ b/docs_src/async_tests/test_main.py @@ -1,14 +1,19 @@ 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"} + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") 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" From 3f8448fc17dfcd24e905db260d91938c38a2f326 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Tue, 7 Jan 2025 11:08:08 +0000 Subject: [PATCH 2/4] =?UTF-8?q?=F0=9F=8E=A8=20[pre-commit.ci]=20Auto=20for?= =?UTF-8?q?mat=20from=20pre-commit.com=20hooks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/en/docs/advanced/async-tests.md | 7 +++---- docs_src/async_tests/test_main.py | 4 +++- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/docs/en/docs/advanced/async-tests.md b/docs/en/docs/advanced/async-tests.md index 108bdf5c6..a44449f18 100644 --- a/docs/en/docs/advanced/async-tests.md +++ b/docs/en/docs/advanced/async-tests.md @@ -59,13 +59,13 @@ $ pytest ## In Detail -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. +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. We use **ASGITransport** from **HTTPX** to interact directly with the **FastAPI** app in an async test environment. -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. +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. +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. ## Other Asynchronous Function Calls @@ -87,4 +87,3 @@ def event_loop() -> Generator[AbstractEventLoop, None, None]: ``` /// - diff --git a/docs_src/async_tests/test_main.py b/docs_src/async_tests/test_main.py index 5815d9dd3..8aea2fd6a 100644 --- a/docs_src/async_tests/test_main.py +++ b/docs_src/async_tests/test_main.py @@ -6,7 +6,9 @@ from .main import app, lifespan @pytest.mark.anyio async def test_root(): - async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: async with lifespan(app=app): # Send a GET request to the root endpoint response = await client.get("/") From 33a6afc5cbc838fe438c95d4d016db5830ddd3b2 Mon Sep 17 00:00:00 2001 From: "oleg.korshunov" Date: Thu, 9 Jan 2025 23:27:54 +0300 Subject: [PATCH 3/4] Returned the old description, corrected it to asyncio.Runner --- docs/en/docs/advanced/async-tests.md | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/docs/en/docs/advanced/async-tests.md b/docs/en/docs/advanced/async-tests.md index a44449f18..df1b5d2d9 100644 --- a/docs/en/docs/advanced/async-tests.md +++ b/docs/en/docs/advanced/async-tests.md @@ -77,13 +77,16 @@ As the testing function is now asynchronous, you can now also call (and await) o 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: ```python +import asyncio +import pytest +from collections.abc import Generator + @pytest.fixture(scope="session") -def event_loop() -> Generator[AbstractEventLoop, None, None]: - """Overrides pytest default function scoped event loop""" - policy = asyncio.get_event_loop_policy() - loop = policy.new_event_loop() - yield loop - loop.close() +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() + ``` /// From 122a9663a6de3d29cdf7393886e874875d8f4074 Mon Sep 17 00:00:00 2001 From: "oleg.korshunov" Date: Fri, 10 Jan 2025 00:54:42 +0300 Subject: [PATCH 4/4] added note about isolating test cases --- docs/en/docs/advanced/async-tests.md | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/docs/en/docs/advanced/async-tests.md b/docs/en/docs/advanced/async-tests.md index df1b5d2d9..256ef92a5 100644 --- a/docs/en/docs/advanced/async-tests.md +++ b/docs/en/docs/advanced/async-tests.md @@ -72,7 +72,7 @@ If you observe issues with state initialization or teardown in your tests, ensur 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 +//// tip 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: @@ -81,7 +81,7 @@ import asyncio import pytest from collections.abc import Generator -@pytest.fixture(scope="session") +@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: @@ -89,4 +89,10 @@ def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]: ``` +/// note + +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. + /// + +////