Інтеграційні тести на Python з використанням pytest та FastAPI
Вітаю! Мене звуть Євген, і я Software Engineer в українській компанії. На проєкті я постійно маю справу з інтеграціями партнерських API.
Обмін між API відбувається за допомогу REST-запитів, тож ми отримуємо та відправляємо багато JSONчиків — все як у всіх. Для підтвердження працездатності функціоналу мені доводиться писати багато різних тестів, зокрема інтеграційних.
Тож хочу поділитися з вами власним досвідом у написанні саме інтеграційних тестів. Це перша частина матеріалу, так би мовити, знайомство, сподіваюсь найближчим часом я закінчу другу частину. Тож до справи.
Вступ
Інтеграційні тести — це різновид тестування, що передбачає перевірку працездатності взаємопов’язаних модулів програмного забезпечення.
Відмінність від Unit-тестів: Unit-тести призначені для того, аби перевіряти найменшу, ізольовану частину коду та доводити її працездатність і відповідність заявленим вимогам.
Логічно, що для перевірки того, як між собою працюють модулі, нам потрібно як мінімум два модулі. Тож я накидав тестовий застосунок для того, аби ви не витрачати час. Посилання на ґіт тут.
Опис застосунку
Застосунок для обслуговування онлайн-магазину з продажу авокадо. Доступний функціонал:
- подивитися список замовлень, статус, вартість;
- зробити замовлення;
- скасувати замовлення.
Список використаних технологій:
- FastAPI;
- MongoDB;
- pytest.
Ознайомимося трохи зі структурою застосунку та тестів.
Для того, аби робити запити, нам знадобиться тестовий клієнт. На щастя, FastAPI вже має такий, ним ми й будемо послуговуватися. Але все ж необхідно внести деякі зміни:
@pytest.fixture(scope="session")
def test_client(database):
with TestClient(app) as client:
app.dependency_overrides[getDB] = override_get_db
yield client
reset_database(database)
app.dependency_overrides = {}
- Додаємо декоратор
fixtureз аргументом scope-рівним session. Це дозволить нам користуватися тестовим клієнтом в межах тестової сесії. Клієнт буде створюватися один раз (без додаткових витрат часу на його створення для кожного тесту) і буде поширений між усіма інтеграційними* тестами. У цьому випадку це виглядає занадто, але з власного досвіду скажу, що на зрізі дві та більше тисячі тестів це дає свої переваги. А на 10+ тисячах це просто необхідно. Детальніше про сферу застосування (scope) тут. - Для того, щоб користуватися тестовою базою даних, ми маємо перевизначити виклик бази даних, у цьому нам допоможе
dependency_overridesфункціонал, уже вбудований у застосунок FastAPI**. - Після виконання тесту обов’язково необхідно очистити тестову базу від даних, які ми туди записали, аби наступний тест не мав проблем.
- З тих же причин очищуємо
dependency_overrides.
*документація pytest радить зберігати окремо фікстури для Unit- та Integration-тестів, тому кожен вид тестів буде мати окремий conftest-файл.
**FastAPI пропонує використовувати механізм ін’єкції залежностей, тому що це має низку переваг (детальніше про це тут). Але простими словами: метод чи функція, що працюють з об’єктом, нічого не знають про цей об’єкт, а тому не несуть відповідальності за його створення.
Саме тому для встановлення з’єднання з базою даних у звичайних умовах (не тестування застосунку) ми використовуємо таке:
def get_db() -> Database:
client = MongoClient()
return client.get_database("config")
Цей же метод get_db додаємо як залежність до кожного ендпоінту, де нам знадобиться доступ до бази даних. Це не просто, але, сподіваюсь, буде зрозуміло.
@orders_router.get("/orders", status_code=HTTPStatus.OK)
def get_orders(db=Depends(get_db)) -> list:
orders = db.orders.find()
return orders_entity(orders)
@orders_router.post("/orders", status_code=HTTPStatus.CREATED)
def process_order(order: dict, db=Depends(get_db)) -> Order:
parsed_order = Order.parse_obj(order)
db.orders.insert_one(parsed_order.dict())
return parsed_order
@orders_router.patch("/order-status-update", status_code=HTTPStatus.NO_CONTENT)
def order_status_update(update: dict, db=Depends(get_db)):
order_id: str = update.get("id", "")
order_status: str = update.get("status", "")
db.orders.update_one({"id": order_id}, {"$set": {"status": order_status}})
return {"massage": "Order status has been updated"}
Таким чином застосунок під час тесту має серед залежностей функцію get_db, яку ми замінюємо на override_get_db, аби встановити з’єднання з тестовою базою даних.
Список замовлень
Маємо необхідний мінімум, аби написати перший тест. Ідея його буде в тому, аби зробити запит на /orders та отримати унаслідок цього статус «код 200» та порожній список, бо ж у нашій тестовій базі даних поки нічого немає.
def test_get_orders(test_client: TestClient) -> None: response = test_client.get(url="/orders") assert response.status_code == HTTPStatus.OK assert not len(response.json())
- Для того, аби pytest розрізняв наш тест як власне тест, назва класу/функції має починатися з test_ або test, більше про це тут.
- Тестовий клієнт, у нашому випадку
test_clientфікстура, який буде використовуватись для опрацювання запитів, має бути додана як аргумент для того, аби мати до нього доступ під час виконання тесту. - Тестовий клієнт має реалізацію всіх http-методів запиту, ми лише хочемо подивитись, а не додавати/змінювати/видаляти, тому
get, в якості url вказуємо/orders— список замовлень. - Після того, як запит надіслано, перевіряємо статус код та дані, які нам віддав сервер.
Що насправді відбувається під час тесту:
Створюється екземпляр TestClient(app) на основі нашого ж застосунку з файлу main.py. Так усі роути застосунку будуть відомі тестовому клієнту. Виклик до робочої бази даних замінюється на виклик до тестової бази. Тестовий клієнт робить HTTP запит на /orders.
Враховуючи те, що наразі тестова база пуста, умовою успішного виконання тесту має бути статус «код 200» та порожній список. Це ми й бачимо нижче.
Запуск тесту
Для того, аби запустити тест, є кілька варіантів. Найпростіший варіант — це натиснути зелену кнопку Play у вашій IDE. Також тести можна запустити з терміналу за допомогою відповідної команди pytest, але займатися налагодженням (debugging) не дуже зручно. Тут вибір за вами.

Встановимо breakpoint на рядку 12 та поглянемо, що ми отримуємо у результаті запиту:
status_code 200;
content b’[]’ (що після перетворення дасть нам порожній список).
А отже вітаю: перший інтеграційний тест написано. А що найприємніше, то це те, що він успішний.
Нове замовлення
Перевірятимемо, чи можна створити нове замовлення, відправивши запит типу POST на /orders та отримати status_code 201. Це означатиме, що замовлення успішно створене.
Також, знаючи номер замовлення (order_id), перевіримо тестову базу даних на наявність замовлення з такий номером.
Відповідно до наших потреб напишемо тест:
def test_make_order(test_client: TestClient, database: Database, test_order: dict) -> None:
order_id = str(uuid1())
test_order["id"] = order_id
response = test_client.post("/orders", json=test_order)
assert response.status_code == HTTPStatus.CREATED
assert database.orders.find_one({"id": order_id})
Поглянемо докладніше на структуру тесту:
- Для того, аби мати доступ до тестової бази даних, додаємо її фікстуру до аргументів.
- Унікальний для кожного тесту
order_idстворюємо окремо, аби потім використати його для запиту до бази даних. - Вміст замовлення test_order*.
- Запит типу POST до
/ordersзtest_orderу якості корисного навантаження. - Перевірка отриманого статус коду.
- Перевірка того, що база даних тепер містить замовлення з відповідним
order_id.
Фікстура, додана як аргумент до тесту, не потребує імпорту чи окремого застосування. Фікстура, додана у conftest-файл відповідного типу тестів, буде доступна для використання на всьому рівні.

Наприклад, фікстура test_order інтеграційних тестів буде доступна для них усіх.
Запустивши тест, бачимо, що він завершився успішно (ну або має завершитися успішно), як і в попередньому тесті поставимо breakpoint на 22 рядку, аби поглянути не тільки на об’єкт response, але цього разу також подивимось, що в нас у базі даних.
Як бачимо, тест успішний, і такі показники як status_code та content відповідають нашим очікуванням.

Не завершуючи тесту, погляньмо, що розташовується в тестовій базі даних. Як і передбачалося до завершення тесту, база містить наше замовлення.

Завершимо тест та звернемось до бази даних ще раз. Як бачимо, база порожня — і це дуже добре, бо означає, що фікстура database після виконання тестом була очищена та в наступному тесті не матиме ніяких side effects.

Оновлення статусу наявного замовлення
Останнім тестом в межах цієї статті буде перевірка успішності оновлення статусу для вже наявного замовлення, яке ми попередньо створили, опрацювали та зберегли в базі даних.
Критерієм успішного виконання тесту буде підтвердження того, що замовлення пройшло всі необхідні кроки, та статуси впродовж опрацювання були змінені успішно.
def test_order_status_update(test_client: TestClient, database: Database, test_order: dict) -> None:
# prepare test order
order_id = str(uuid1())
test_order["id"] = order_id
response = test_client.post("/orders", json=test_order)
# check that order was created
assert response.status_code == HTTPStatus.CREATED
assert database.orders.find_one({"id": order_id})
status_update_payload = {
"id": order_id,
"status": OrderStatus.PROCESSING,
}
# status transition from NEW to PROCESSING
response = test_client.patch("/order-status-update", json=status_update_payload)
assert response.status_code == HTTPStatus.NO_CONTENT
order = database.orders.find_one({"id": order_id})
assert order.get("status") == OrderStatus.PROCESSING
status_update_payload = {
"id": order_id,
"status": OrderStatus.DONE,
}
# status transition from PROCESSING to DONE
response = test_client.patch("/order-status-update", json=status_update_payload)
assert response.status_code == HTTPStatus.NO_CONTENT
order = database.orders.find_one({"id": order_id})
assert order.get("status") == OrderStatus.DONE
Погляньмо докладніше на структуру тесту:
- Початок — це копія попереднього тесту Список замовлень.
- Друга частина — це відправка
status_update_payloadдля оновлення конкретного замовлення. Для цього ми беремо вже наявнийorder_idта новий статус замовлення PROCESSING. - Після відправки запиту дістаємо замовлення з бази та перевіряємо, що статус замовлення оновлено з NEW до PROCESSING.
- Повторимо пункт другий, тільки тепер змінимо
status_update_payloadтак, аби надіслати статус DONE. - Знову перевіримо статус, він змінився з PROCESSING на DONE.
Висновки
Вітаю, завдання виконано!
Основною ідеєю цієї статті було показати, як написати інтеграційні тести для простого застосунку з замовлення авокадо, що нам успішно вдалося.
Тож загалом ми зрозуміли:
- як, використовуючи тестовий клієнт, наданий FastAPI, надсилати замовлення на свій же застосунок;
- як встановити з’єднання з тестовою базою даних під час тесту та очистити її після його виконання.
Здавалося б і не так багато, але якщо придивитися уважно, можна винести багато чого корисного. Якщо ваш застосунок має публічне або приватне API і ви хочете зрозуміти, як він буде поводитись під час того чи іншого запиту, цей матеріал буде корисним.
Модифікуючи ці тести, можна також перевірити автентифікацію, модифікатори доступу (якщо у вас є кілька ролей з різними обмеженнями доступу та декораторами типу authenticated) та багато інших, не менш важливих деталей, які ви можете використовувати у своєму застосунку.
Також у мене є бажання одного дня розповісти про дуже круту бібліотеку responses у цьому контексті, але це вже якось іншим разом!
Окрема подяка за допомогу та редагування технічної частини моєму другу і наставнику — Владиславу Максимову.
Дякую за увагу, бережіть себе!

Немає коментарів
Додати коментар Підписатись на коментаріВідписатись від коментарів