Pytest with django celery

I have celery task which create models entries in database. Celery uses Redis as a broker.

Celery config in settings.py:

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'django-db'

from .models import Message
from itertools import islice
from celery import shared_task

@shared_task(ignore_result=True)
def create_entries(data: list):
    batch_size = 100
    obj_iterator = (Message(**obj) for obj in data)
    while True:
        batch = list(islice(obj_iterator, batch_size))
        if not batch:
            break
        Message.objects.bulk_create(batch, batch_size)

In application everything works fine, but tests don't work. Test method code:

import tasks
import pytest
from .models import Message
from celery.result import AsyncResult

@pytest.mark.django_db
@pytest.mark.celery
def test_create_entries():
    message_data = [
        {
            text: 'hello bro',
            client: 'Nick'
        },
    ]
    assert Message.objects.count() == 0
    task = tasks.create_entries.delay(message_data)
    result = AsyncResult(task.task_id)
    assert result.status == 'SUCCESS' # here is a fail, status = PENDING
    assert Message.objects.count() == 1 # here is a fail, count = 0

But if run server: python manage.py runserver and run tasks, then entries are created successfully.

What could be the reason that tasks don't work in tests?

Back to Top