python-3.x - Celery 在任务中模拟网络调用

Question

我正在使用 Celery 4.3.0 和 pytest 我能够使用 celery_worker fixture 运行我的 celery 任务以进行单元测试。

然而，在我的 celery 任务中，我想模拟网络调用。

@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

在上面的例子中，我想 mock.patch lkl_core.newsletter.tasks.MailChimp 位于 mailchimp_subscription 任务中。模拟操作不适用于该任务，因为我相信它在不同的线程上运行。

关于我如何实现这一点有什么建议吗？

谢谢

Answer 1

尝试用 mock 包装 celery_app fixture:

@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):
    ...