Модуль 32

Тестування Celery-задач

Це просто чудова тема! Вона знаходиться на перетині асинхронності, архітектури та надійності — саме те, що люблять розбирати на CS50.

Вмикайте свої редактори коду, наливайте каву, і поїхали! 🚀

🧠 Урок: Тестування Celery-задач

(або: Як перевірити відкладене, не чекаючи вічність)

1. 🔥 Вступ: Проблема та мотивація

Уявіть, що ви будуєте новий Amazon. Користувач натискає кнопку «Купити», і ви хочете відправити йому email-підтвердження. Але відправка email — це довго. Секунда, дві, іноді п'ять. Ви не хочете, щоб користувач дивився на кружечок завантаження, правда?

Ви вирішуєте: "О! Я використаю Celery!" Ви кидаєте задачу в чергу («відправити лист»), а користувачу миттєво кажете: «Дякуємо, все буде!».

Але тут виникає проблема, коли ви сідаєте писати тести. Ви запускаєте pytest, і... тиша. Або тест проходить миттєво, але лист не прийшов. Або ще гірше — під час тестів ви випадково відправляєте 1000 реальних листів на свою тестову пошту (або, борони Боже, реальним клієнтам).

Питання до вас: Як нам протестувати логіку задачі, яка створена для того, щоб запускатися десь там, колись потім і іншим процесом?

Чи потрібно нам піднімати реальний Redis та запускати окремий worker-процес для кожного юніт-тесту? Це звучить як жах для швидкості розробки, чи не так?

Сьогодні ми навчимося "зупиняти час" і змушувати Celery грати за нашими правилами.

2. 🧠 Теоретична база (без сухої академічності)

Щоб зрозуміти, як це тестувати, згадаймо, як працює Celery.

У нас є три гравці: 1. Producer (Ваш код): Каже "Зроби це". 2. Broker (Redis/RabbitMQ): Поштова скринька, де лежить задача. 3. Worker (Celery процес): Той, хто забирає задачу і виконує її.

У звичайному житті вони розділені. Але в тестах нам часто це заважає.

🔑 Два головні підходи до тестування:

  1. Режим "Eager" (Нетерплячий режим). Це — магія. Ми кажемо Celery: "Забудь про черги. Забудь про воркерів. Як тільки я викликаю .delay(), просто візьми і виконай цю функцію прямо тут і зараз, синхронно".

    • Аналогія: Замість того, щоб кидати лист у поштову скриньку і чекати листоношу, ви самі берете лист, біжите за адресою і вручаєте його.
    • Навіщо: Щоб перевірити логіку всередині задачі.

  2. Mocking (Імітація). Ми взагалі не виконуємо задачу. Ми просто перевіряємо, чи спробував наш код викликати цю задачу з правильними аргументами.

    • Аналогія: Ви перевіряєте, чи правильно написана адреса на конверті, але нікуди його не несете.

💡 Що треба запам'ятати залізно:

Unit-тести не повинні залежати від зовнішніх сервісів (Redis, RabbitMQ). Вони мають бути швидкими та ізольованими.

3. 🧪 Приклади (від простого до реального)

Давайте подивимось на код. Припустимо, у нас є проста задача в tasks.py:

# tasks.py
from celery import shared_task

@shared_task
def add(x, y):
    return x + y

Приклад 1: Наївна спроба (Провал)

# test_tasks.py
from tasks import add

def test_add_naive():
    result = add.delay(4, 4)
    # Що тут буде?
    assert result.get() == 8

Чому це погано? Якщо у вас не запущений локальний Redis і Celery worker під час тесту, цей код "зависне" або впаде з помилкою з’єднання. Тест залежить від інфраструктури. Це — поганий тон.

Приклад 2: Використання CELERY_TASK_ALWAYS_EAGER (Успіх)

Ми змушуємо Celery виконувати все синхронно.

# test_tasks_eager.py
import pytest
from tasks import add

# У Django це робиться через override_settings, 
# у чистому Celery — через конфігурацію app.conf.task_always_eager = True

def test_add_eager(celery_app, celery_worker):
    # Вмикаємо "нетерплячий" режим
    celery_app.conf.update(task_always_eager=True)

    # Викликаємо .delay(), але код виконується миттєво!
    result = add.delay(4, 4)

    # У Eager-режимі result.get() повертає результат відразу
    assert result.get() == 8

Тепер це звичайний виклик функції, просто трохи обгорнутий.

Приклад 3: Реальний кейс (Mocking)

Уявіть, що у вас є функція реєстрації, яка викликає задачу.

# services.py
from tasks import send_welcome_email

def register_user(email):
    # ... логіка створення юзера в БД ...
    send_welcome_email.delay(email)
    return "User created"

Ми хочемо перевірити register_user. Нам байдуже, чи працює send_welcome_email (ми перевіримо її окремо). Нам важливо, чи викликали ми її.

# test_services.py
from unittest.mock import patch
from services import register_user

@patch('services.send_welcome_email.delay')
def test_register_triggers_email(mock_task):
    email = "david@cs50.harvard.edu"

    register_user(email)

    # Перевіряємо: "Чи викликали ми задачу з цим email?"
    mock_task.assert_called_once_with(email)

Результат: Миттєвий тест, жодних Redis, повна впевненість, що потік управління правильний.

4. 🛠 Практична частина

А тепер ваша черга! Відкривайте IDE або блокнот.

Завдання 1: Класика

Напишіть тест для задачі multiply(x, y), використовуючи режим task_always_eager. Переконайтеся, що multiply.delay(3, 3) повертає 9.

Завдання 2: Обробка помилок

Створіть задачу divide(x, y). Напишіть тест, який передає y=0. Питання: Як поводиться тест у Eager-режимі? Чи викидає він ZeroDivisionError прямо в тесті? (Спробуйте і побачите, що так!). Огорніть це в pytest.raises.

Завдання 3: "Шпигун" (Spy)

У вас є задача, яка нічого не повертає, але робить print("Hello"). Використайте unittest.mock.patch на builtins.print і запустіть задачу в Eager-режимі. Перевірте, чи був викликаний print.

Завдання 4: Реальний виклик (Міні-кейс)

Задача: process_order(order_id). Вона має: 1. Знайти замовлення в БД (можна замокати). 2. Якщо статус "Paid" — нічого не робити. 3. Якщо "New" — змінити на "Processing". Завдання: Напишіть тест, який перевіряє логіку зміни статусу, запустивши задачу синхронно (без .delay(), просто викликавши функцію як process_order(1) — так, так теж можна тестувати задачі!).

Завдання 5: А що, якщо... Retry?

Задача налаштована на повтор (autoretry_for=(ConnectionError,)). Спробуйте написати тест, де ви мокаєте внутрішній виклик так, щоб він викликав ConnectionError. Перевірте, чи Celery (в eager режимі) намагається повторити задачу або викидає Retry exception.

5. 💡 Мислення як у розробника

Ось де новачки часто помиляються, а сеньйори — посміхаються.

1. Пастка "Інтеграційного пекла"

Новачок намагається підняти Docker-контейнери з Redis для кожного прогону тестів. Як думає профі: "Я довіряю розробникам Redis і Celery, що їхній софт працює. Мені треба перевірити мій код. task_always_eager покриває 95% моїх потреб. Для решти 5% я запущу один великий end-to-end тест перед релізом".

2. Пастка аргументів

Celery серіалізує дані (перетворює в JSON/Pickle). Помилка: Передавати в задачу складний об'єкт (наприклад, об'єкт моделі Django або з'єднання з БД). У тестах в Eager-режимі це спрацює! А на проді — впаде, бо JSON не вміє зберігати об'єкти Python. Порада: Завжди передавайте в задачі лише прості ID (int, string). В тесті теж передавайте ID.

3. Тестування .delay() vs тестування функції

Ви можете імпортувати задачу і викликати її без .delay() як звичайну функцію: my_task(arg). Це абсолютно нормально для перевірки внутрішньої "кухні" задачі.

6. 🧩 Підсумок

Отже, що ми сьогодні зробили? Ми взяли асинхронного звіра Celery і навчили його ходити по струнці.

Тепер ви вмієте: 1. Тестувати логіку задач миттєво за допомогою task_always_eager (синхронний режим). 2. Використовувати mocks, щоб перевірити сам факт відправки задачі, не виконуючи її. 3. Розумієте різницю між перевіркою інфраструктури (чи працює черга) і бізнес-логіки (чи правильно рахує задача).

Ти справді розібрався з цим! Тепер твої тести не будуть відправляти спам клієнтам, а CI/CD проходитиме за хвилини, а не години.

Далі в курсі: Ми навчилися запускати задачі. Але що робити, коли задач стає мільйон, і Redis починає "захлинатися"? Наступного разу поговоримо про Моніторинг черг та Flower. Готуйтеся, будемо дивитися на графіки! 📈

А поки що — щасливого кодингу!

← До списку уроків Наступний урок →