Перейти к содержанию

Тестирование

Благодаря Starlette, тестировать приложения FastAPI легко и приятно.

Тестирование основано на библиотеке HTTPX, которая в свою очередь основана на библиотеке Requests, так что все действия знакомы и интуитивно понятны.

Используя эти инструменты, Вы можете напрямую задействовать pytest с FastAPI.

Использование класса TestClient

Информация

Для использования класса TestClient необходимо установить библиотеку httpx.

Например, так: pip install httpx.

Импортируйте TestClient.

Создайте объект TestClient, передав ему в качестве параметра Ваше приложение FastAPI.

Создайте функцию, название которой должно начинаться с test_ (это стандарт из соглашений pytest).

Используйте объект TestClient так же, как Вы используете httpx.

Напишите простое утверждение с assert дабы проверить истинность Python-выражения (это тоже стандарт pytest).

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}


client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

Подсказка

Обратите внимание, что тестирующая функция является обычной def, а не асинхронной async def.

И вызов клиента также осуществляется без await.

Это позволяет вам использовать pytest без лишних усложнений.

Технические детали

Также можно написать from starlette.testclient import TestClient.

FastAPI предоставляет тот же самый starlette.testclient как fastapi.testclient. Это всего лишь небольшое удобство для Вас, как разработчика.

Подсказка

Если для тестирования Вам, помимо запросов к приложению FastAPI, необходимо вызывать асинхронные функции (например, для подключения к базе данных с помощью асинхронного драйвера), то ознакомьтесь со страницей Асинхронное тестирование в расширенном руководстве.

Разделение тестов и приложения

В реальном приложении Вы, вероятно, разместите тесты в отдельном файле.

Кроме того, Ваше приложение FastAPI может состоять из нескольких файлов, модулей и т.п.

Файл приложения FastAPI

Допустим, структура файлов Вашего приложения похожа на ту, что описана на странице Более крупные приложения:

.
├── app
│   ├── __init__.py
│   └── main.py

Здесь файл main.py является "точкой входа" в Ваше приложение и содержит инициализацию Вашего приложения FastAPI:

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

Файл тестов

Также у Вас может быть файл test_main.py содержащий тесты. Можно разместить тестовый файл и файл приложения в одной директории (в директориях для Python-кода желательно размещать и файл __init__.py):

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

Так как оба файла находятся в одной директории, для импорта объекта приложения из файла main в файл test_main Вы можете использовать относительный импорт:

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

...и писать дальше тесты, как и раньше.

Тестирование: расширенный пример

Теперь давайте расширим наш пример и добавим деталей, чтоб посмотреть, как тестировать различные части приложения.

Расширенный файл приложения FastAPI

Мы продолжим работу с той же файловой структурой, что и ранее:

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

Предположим, что в файле main.py с приложением FastAPI есть несколько операций пути.

В нём описана операция GET, которая может вернуть ошибку.

Ещё есть операция POST и она тоже может вернуть ошибку.

Обе операции пути требуют наличия в запросе заголовка X-Token.

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=400, detail="Item already exists")
    fake_db[item.id] = item
    return item
from typing import Annotated, Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=400, detail="Item already exists")
    fake_db[item.id] = item
    return item
from typing import Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from typing_extensions import Annotated

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=400, detail="Item already exists")
    fake_db[item.id] = item
    return item

Подсказка

По возможности используйте версию с Annotated.

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

Подсказка

По возможности используйте версию с Annotated.

from typing import Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

Расширенный файл тестов

Теперь обновим файл test_main.py, добавив в него тестов:

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_item():
    response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 200
    assert response.json() == {
        "id": "foo",
        "title": "Foo",
        "description": "There goes my hero",
    }


def test_read_item_bad_token():
    response = client.get("/items/foo", headers={"X-Token": "hailhydra"})
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_read_inexistent_item():
    response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}


def test_create_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
    )
    assert response.status_code == 200
    assert response.json() == {
        "id": "foobar",
        "title": "Foo Bar",
        "description": "The Foo Barters",
    }


def test_create_item_bad_token():
    response = client.post(
        "/items/",
        headers={"X-Token": "hailhydra"},
        json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},
    )
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_create_existing_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={
            "id": "foo",
            "title": "The Foo ID Stealers",
            "description": "There goes my stealer",
        },
    )
    assert response.status_code == 409
    assert response.json() == {"detail": "Item already exists"}

Если Вы не знаете, как передать информацию в запросе, можете воспользоваться поисковиком (погуглить) и задать вопрос: "Как передать информацию в запросе с помощью httpx", можно даже спросить: "Как передать информацию в запросе с помощью requests", поскольку дизайн HTTPX основан на дизайне Requests.

Затем Вы просто применяете найденные ответы в тестах.

Например:

  • Передаёте path-параметры или query-параметры, вписав их непосредственно в строку URL.
  • Передаёте JSON в теле запроса, передав Python-объект (например: dict) через именованный параметр json.
  • Если же Вам необходимо отправить форму с данными вместо JSON, то используйте параметр data вместо json.
  • Для передачи заголовков, передайте объект dict через параметр headers.
  • Для передачи cookies также передайте dict, но через параметр cookies.

Для получения дополнительной информации о передаче данных на бэкенд с помощью httpx или TestClient ознакомьтесь с документацией HTTPX.

Информация

Обратите внимание, что TestClient принимает данные, которые можно конвертировать в JSON, но не модели Pydantic.

Если в Ваших тестах есть модели Pydantic и Вы хотите отправить их в тестируемое приложение, то можете использовать функцию jsonable_encoder, описанную на странице Кодировщик совместимый с JSON.

Запуск тестов

Далее Вам нужно установить pytest:

$ pip install pytest

---> 100%

Он автоматически найдёт все файлы и тесты, выполнит их и предоставит Вам отчёт о результатах тестирования.

Запустите тесты командой pytest и увидите результат:

$ pytest

================ test session starts ================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /home/user/code/superawesome-cli/app
plugins: forked-1.1.3, xdist-1.31.0, cov-2.8.1
collected 6 items

---> 100%

test_main.py <span style="color: green; white-space: pre;">......                            [100%]</span>

<span style="color: green;">================= 1 passed in 0.03s =================</span>