The task is a function: I call it directly (task.run(...) or task(...)) in the unit test.
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 |