Celery mocking network calls inside task

user_145277 0
Viewed 86

I am using Celery 4.3.0 with pytest am able to run my celery tasks for unit testing using celery_worker fixture.

Yet, inside my celery task, I want to mock network calls.

@pytest.mark.django_db(transaction=True)
@mock.patch("lkl_core.newsletter.tasks.MailChimp")
def test_celery_task_mail_chimp_subscription(mock_mc_client, celery_app, celery_worker):
    instance = SubscriptionFactory()
    assert mailchimp_subscription.apply_async(
            [instance.id,]
        ).wait(3) is None

In the above example, I want to mock.patch lkl_core.newsletter.tasks.MailChimp which lives inside mailchimp_subscription task. The mock operation is not applied to the task since I believe it runs on a different thread.

Any advice on how can I achieve that?

Thanks

pytest celery python-3.x unit-testing
1 Answers

Try wrapping the celery_app fixture with the mock:

@pytest.fixture
def mailchimp_mock(celery_app):
    with patch('lkl_core.newsletter.tasks.MailChimp') as mock:
        yield mock


@pytest.mark.django_db(transaction=True)
def test_celery_task_mail_chimp_subscription(mailchimp_mock, celery_worker):
    ...
user_14517380
Related