Асинхронні тести¶

🌐 Переклад ШІ та людьми

Цей переклад виконано ШІ під керівництвом людей. 🤝

Можливі помилки через неправильне розуміння початкового змісту або неприродні формулювання тощо. 🤖

Ви можете покращити цей переклад, допомігши нам краще спрямовувати AI LLM.

Ви вже бачили, як тестувати ваші застосунки 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 міститиме:

Python 3.10+

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Tomato"}

🤓 Other versions and variants

Python 3.9+

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Tomato"}

Файл test_main.py міститиме тести для main.py, тепер це може виглядати так:

Python 3.10+

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"}

🤓 Other versions and variants

Python 3.9+

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"}

Запуск¶

Ви можете запустити тести як зазвичай:

$ pytest

---> 100%

Докладно¶

Маркер @pytest.mark.anyio повідомляє pytest, що цю тестову функцію слід викликати асинхронно:

Python 3.10+

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"}

🤓 Other versions and variants

Python 3.9+

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.

Python 3.10+

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"}

🤓 Other versions and variants

Python 3.9+

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.

Порада

Зауважте, що ми використовуємо async/await із новим AsyncClient - запит є асинхронним.

Попередження

Якщо ваш застосунок залежить від подій тривалості життя, AsyncClient не ініціюватиме ці події. Щоб гарантувати їх ініціалізацію, використовуйте LifespanManager з florimondmanca/asgi-lifespan.

Інші асинхронні виклики функцій¶

Оскільки тестова функція тепер асинхронна, ви також можете викликати (і await) інші async-функції окрім надсилання запитів до вашого застосунку FastAPI у тестах - так само, як ви робили б це будь-де у вашому коді.

Порада

Якщо ви натрапили на RuntimeError: Task attached to a different loop під час інтеграції асинхронних викликів у ваші тести (наприклад, при використанні MongoDB's MotorClient), пам'ятайте створювати об'єкти, яким потрібен цикл подій, лише всередині асинхронних функцій, наприклад, у зворотному виклику @app.on_event("startup").