test_celery_tasks.md

Introduction

In my previous article I promised you to show you how to properly unit test Celery tasks. Since I keep my promises every time, this is the target of the following article.

You had better be familiar with the mocking technique in the unit test practices before dive into the article. If you are not that sure in your knowledge - check this article.

Repetition is the key to success

I'm going to expand my previous article so you had better take a quick look at it. If you already read it you can take a look at its demo.

Let's start

First of all, I want to give you a brief look of how I understand unittesting. It is a test technique that must that allow us to cover our system and give us the abillity to prove that our code works correct.

The most esential point is that you unittest which means you have to test only one specific functionality in one testcase. If you don't so you are doing some kind of integration testing which is way different and is not in the topic of this article.

Dive in

Once we revisioned our unittest knowledge we are ready to start with the main part - test the code from the previous article.

This is how our tests.py file looks like:

# in tasks.py
from third_party import ThirdPartyClient
from celery import shared_task, Task, chain

from django.db import transaction
from django.conf import settings

from .models import ThirdPartyDataStorage, AsyncActionReport
from .mixins import BaseErrorHandlerMixin


class ThirdPartyBaseTask(BaseErrorHandlerMixin, Task):
    pass


@shared_task(base=ThirdPartyBaseTask)
def fetch_data(**kwargs):
    """
    Expected kwargs: 'async_action_report_id': AsyncActionReport.id.

    This kwargs is going to be passed to the constructor of
    the ThirdPartyBaseTask so we can handle the exceptions and store it
    in the AsyncActionReport model.
    """
    client = ThirdPartyClient(api_key=settings.THIRD_PARTY_API_KEY)
    fetched_data = client.fetch_data_method()

    return fetched_data


@shared_task
@transaction.atomic
def store_data(fetched_data):
    container = ThirdPartyDataStorage(**fetched_data)
    container.save()

    return container.id


@shared_task
def fetch_data_and_store_it():
    async_action_report = AsyncActionReport()
    t1 = fetch_data.s(async_action_report_id=async_action_report.id)
    t2 = store_data.s()

    return chain(t1, t2).delay()

Plan it

Let's make a quick draft about how our tests are going to behave:

• Test fetch_data_and_store_it() task
• Test store_data() task
• Test fetch_data() task (here we should include exception raising tests)

We are ready to actually start the real testing part.

Mock everything

As you may noticed most of our tasks actually call other task. For example fetch_data_and_store_it() just chains two tasks.

Since we already know what mocking is we can use it to actually mock these two tasks and manipulate their return value.

In the other test cases we are going to make pretty much the same.

Before we start

Before we start we must knwo something - Django defaultly runs its tests with CELERY_ALWAYS_EAGER=True. What does that mean? It disables the asynchrony of the Celery and make it synchronous!

Start testing

Let's test our "mother" function. What it actually does is to call two other functions - so that's what we are going to test!