Testing Celery tasks: mocking .delay and capturing on_commit callbacks

Backend Live Coding November 13, 2026 python

The task is a function: I call it directly (task.run(...) or task(...)) in the unit test.

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# ======================================================================
# Q: How do you test Celery tasks?
#
# The task is a function: I call it directly (task.run(...) or task(...)) in the
# unit test. To check that it was scheduled correctly, I mock .delay. I avoid
# CELERY_TASK_ALWAYS_EAGER as a global setting because it hides problems
# (serialization, on_commit, retries).
# ======================================================================

@pytest.mark.django_db
def test_confirm_upload_enqueues_processing(api_client, ws_user, mocker,
                                            django_capture_on_commit_callbacks):
    media = MediaFactory(workspace=ws_user.workspace, status='pending')
    delay = mocker.patch('media.views.process_media.delay')
    api_client.force_authenticate(ws_user)

    with django_capture_on_commit_callbacks(execute=True):
        r = api_client.post(f'/api/v2/media/{media.pk}/confirm/')

    assert r.status_code == 202
    delay.assert_called_once_with(media.pk)

@pytest.mark.django_db
def test_other_workspace_cannot_confirm(api_client, ws_user):
    media = MediaFactory()                     # different workspace
    api_client.force_authenticate(ws_user)
    r = api_client.post(f'/api/v2/media/{media.pk}/confirm/')
    assert r.status_code == 404