Асинхронные тесты
Вы уже видели, как тестировать приложения FastAPI с помощью предоставленного TestClient. До сих пор вы видели только, как писать синхронные тесты, не используя функции async.
Возможность использовать асинхронные функции в тестах может быть полезна, например, когда вы асинхронно выполняете запросы к базе данных. Представьте, что вы хотите протестировать отправку запросов к приложению FastAPI, а затем убедиться, что серверная часть успешно записала нужные данные в базу данных, используя асинхронную библиотеку для работы с базой данных.
Давайте посмотрим, как это можно сделать.
pytest.mark.anyio
Если мы хотим вызывать асинхронные функции в тестах, тестовые функции тоже должны быть асинхронными. AnyIO предоставляет удобный плагин, позволяющий указать, что некоторые тестовые функции нужно вызывать асинхронно.
HTTPX
Даже если ваше приложение FastAPI использует обычные функции def вместо async def, внутри оно всё равно остаётся приложением async.
TestClient выполняет некоторые действия, чтобы вызывать асинхронное приложение FastAPI из обычных синхронных тестовых функций def с помощью стандартного pytest. Но этот механизм перестаёт работать, когда мы используем его внутри асинхронных функций. Запуская тесты асинхронно, мы больше не можем использовать TestClient внутри тестовых функций.
TestClient основан на HTTPX, и, к счастью, мы можем использовать его напрямую для тестирования API.
Пример
Для простого примера рассмотрим структуру файлов, похожую на описанную в разделах Большие приложения и Тестирование:
.
├── app
│ ├── __init__.py
│ ├── main.py
│ └── test_main.py
Файл main.py будет содержать:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Tomato"}
Файл test_main.py будет содержать тесты для main.py. Теперь он может выглядеть так:
import pytest
from httpx import ASGITransport, AsyncClient
from .main import app
@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"}
Запуск
Вы можете запустить тесты обычным способом:
$ uv run pytest
---> 100%
Подробнее
Маркер @pytest.mark.anyio сообщает pytest, что эту тестовую функцию нужно вызывать асинхронно:
import pytest
from httpx import ASGITransport, AsyncClient
from .main import app
@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 def, а не просто def, как раньше при использовании TestClient.
Затем мы можем создать AsyncClient с приложением и отправлять ему асинхронные запросы с помощью await.
import pytest
from httpx import ASGITransport, AsyncClient
from .main import app
@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"}
Это эквивалентно следующему:
response = client.get('/')
...что мы использовали для отправки запросов с помощью TestClient.
Совет
Обратите внимание, что с новым AsyncClient мы используем async/await — запрос выполняется асинхронно.
Предупреждение
Если ваше приложение зависит от событий жизненного цикла, AsyncClient не вызовет эти события. Чтобы они были вызваны, используйте LifespanManager из florimondmanca/asgi-lifespan.
Другие вызовы асинхронных функций
Поскольку тестовая функция теперь асинхронная, вы также можете вызывать (и await) другие функции async в тестах, помимо отправки запросов к приложению FastAPI, точно так же, как вы вызывали бы их в любом другом месте кода.
Совет
Если при добавлении вызовов асинхронных функций в тесты вы столкнётесь с RuntimeError: Task attached to a different loop (например, при использовании MotorClient для MongoDB), не забывайте создавать объекты, которым нужен цикл событий, только внутри асинхронных функций, например в callback-функции @app.on_event("startup").
© 2018 Sebastián Ramírez
Licensed under the MIT License.
https://fastapi.tiangolo.com/advanced/async-tests/