CODEWITHSUNDEEP
pythondjangocelery
jeldzinski
jeldzinski

Reputation: 37

Celery task behave strangely while testing

I'm trying to test that my Celery task updates Django model. It works fine, but behaves strangely during testing.

# someapp/models.py

class SomeModel(models.Model):
    ...
    hidden = models.BooleanField(default=False)

# someapp/tasks.py

@shared_task()
def my_task(model_id):
    model_instance = someapp.models.SomeModel.objects.get(id=model_id)
    model_instance.hidden = True
    model_instance.save()
    logger.info(f'Uncovered model instance with id {model_id]')

To test this I've implemented following workflow:

The code below

# someapp/tests.py

@pytest.mark.django_db
@pytest.mark.celery(task_always_eager=True)
def test_celery_task_uncovers_model_instance() -> None:

    SomeModelFactory.create(hidden=False)
    some_model = someapp.models.SomeModel.objects.first()
    assert some_model.hidden is True
    my_task.apply((some_model.id, ))

    assert some_model.hidden is True

raises at the last line. Then I assert:

assert (model_instance.pk, model_instance.hidden) == (someapp.models.SomeModel.objects.first().pk,
                                                      someapp.models.SomeModel.objects.first().hidden)

It raises:

E       assert (1, True) == (1, False)
E         At index 1 diff: True != False

Finally, I want to check ids:

assert id(model_instance) == id(authapp.models.SomeModel.objects.first())

And it raises something like this:

E       AssertionError: assert 139938217188656 == 139938219885184
E        +  where 139938217188656 = id(<SomeModel:  - 2022-02-01>)
E        +  and   139938219885184 = id(<SomeModel:  - 2022-02-01>)

Why does not the task update the some_model object in my test?

Upvotes: 0

Views: 475

Answers (1)

Iain Shelvington
Iain Shelvington

Reputation: 32244

The instance was updated in the task but not in your test, you need to refresh the instance's data from the DB using Model.refresh_from_db() after calling the task

@pytest.mark.django_db
@pytest.mark.celery(task_always_eager=True)
def test_celery_task_uncovers_model_instance() -> None:

    SomeModelFactory.create(hidden=False)
    some_model = someapp.models.SomeModel.objects.first()

    assert some_model.hidden is False

    my_task.apply((some_model.id, ))
    some_model.refresh_from_db()  # <<<

    assert some_model.hidden is True

Upvotes: 3

Related Questions