# A Practical Guide to Unit Testing Celery Tasks

While you might get away with *not* writing unit tests for very simple Rest API endpoints, doing the same for celery tasks is a recipe for frustration (and disaster).

Celery tasks are asynchronous by design and therefore a lot harder to get a grip on using a "development-driven development" approach.

Test Driven Development (TDD) might not have taken us to the promised land we had hoped for. But when it comes to celery tasks, it most definitely is essential to a sane, effective and efficient development process - and having that peace of mind when releasing your code into production.

## **A Celery task**

Let's have a look at this simple celery task:

```python
import requests
import os

from datetime import datetime
from worker import app

@app.task(bind=True, name='fetch_data')
def fetch_data(self, url):
    response = requests.get(url)
    path = './data'
    if response.ok:
        if not os.path.exists(path):
            os.makedirs(path)
        slug = datetime.utcnow().strftime('%Y%m%dT%H%M%S%f')
        with open(os.path.join(path, slug), 'w') as f:
            f.write(response.text)
    else:
        raise ValueError('Unexpected response')
```

This celery task executes a GET request against the argument `url` and saves the response body to the file system.  
There are several strategies to test this Celery task.

## **Strategy 1: Wait for the task to finish**

[Some authors](https://medium.com/@solankiarpit1997/testing-celery-tasks-a8a3568d0213?ref=celery-school.ghost.io){:target="\_blank"} recommend  
calling the Celery task asynchronously and then making the code wait until the task is ready to fetch the  
result and evaluate the test assertions.

```python
def test_fetch_data(self):
    task = fetch_data.s(url='...').delay()
    result = task.get()
    self.assertEqual(task.status, 'SUCCESS')
    ...
```

**Pros**

* Tests the Celery stuff
    
* Testcase and the real world are nearly identical
    
* Very close to the real environment
    

**Cons**

* Dependency on message broker
    
* Requires a celery worker
    
* More of an integration than a unit test
    

## **Strategy 2: Just test the method**

The [Celery docs](http://docs.celeryproject.org/en/latest/userguide/testing.html?ref=celery-school.ghost.io) suggest Celery tasks should just be tested like any other Python method.

```python
def test_fetch_data(self):
    fetch_data(url='...')
    self.assertEqual(...)
    ...
```

**Pros**

* Very simple
    
* No dependency on the message broker
    
* No celery worker required
    
* An isolated unit test
    

**Cons**

* Does not test the Celery stuff
    
* Testcase and the real world differ
    

## **Strategy 3: Call the task synchronously**

This strategy combines the best of both worlds. We call the Celery task in *nearly* the same way we do in real life, but synchronously (no need to wait) and locally (in the same process).

```python
def test_fetch_data(self):
    task = fetch_data.s(url='...').apply()
    self.assertEqual(task.result, 'SUCCESS')
    ...
```

**Pros**

* Very simple
    
* No dependency on the message broker
    
* No celery worker required
    
* An isolated unit test
    
* Tests the Celery stuff
    
* Testcase and the real world are very close
    

**Cons**

* Not sure if there are any
    

## **How to apply this**

Invoking your Celery tasks inside your tests with the `apply()` method executes the task synchronously and locally. This allows you to write tests that look and feel very similar to the ones for your API endpoints. Next time, I will look at how to test Celery chains.