模拟芹菜任务方法 apply_async

Question

我创建了一个依赖于其他模块的 celery 任务。我在模型保存方法中延迟了芹菜任务，我想在没有芹菜的情况下通过所有测试用例。下面是我的代码。

def save(self, force_insert=False, force_update=False, using=None,
             update_fields=None):

        super(Offers, self).save(force_insert=False, force_update=False, using=None,
             update_fields=None)

        change_offer_status.apply_async(args=[self.id], eta=self.valid_to, queue='credr_core_task',
                                               routing_key='credr_core_task')

Test.py

class OfferTests(APITestCase):
    authenticated_client = APIClient()

    def setUp(self):
        data = {
            "username": "vivek",
            "first_name": "Vivek",
            "last_name": "Dogra",
            "email": "vivk@credr.com",
            "contact_number": "9834982602",
            "password": "easy"
        }
        self.user = User.objects.create(data)
        token = Token.objects.get(user__username='vivek')
        self.authenticated_client.credentials(HTTP_AUTHORIZATION='Token ' + token.key)
        mock_task = Mock()
        mock_task.get = Mock(return_value={'success': True})

        print mock_task.get() # outputs {'success': True}

        with patch('offers.tasks.change_offer_status.apply_async', new=mock_task) as mocked_task:
            mocked_task.return_value = True

    @override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
                   CELERY_ALWAYS_EAGER=True, )
    def test_add_offer(self):
        """
        add user address
        """
        start_date = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S")
        end_date = datetime.strftime(datetime.now() + timedelta(days=3), "%Y-%m-%d %H:%M:%S")
        data = {
            "type": OfferTypeEnum.FLAT._value_,
            "name": "CredR Offer",
            "code": "FLAT100",
            "status": OfferStatusEnum.ACTIVE._value_,
            "value": "1004",
            "discount_value": 1004,
            "valid_from": start_date,
            "valid_to": end_date,
            "message": "GET FLAT 100"
        }
        response = self.authenticated_client.post(URL, data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

我在 setUp 方法中创建了一个模拟对象。但是我收到 rabbitMq 的连接被拒绝错误。

Answer 1

i want to pass all test cases without celery

当运行你的测试时设置 CELERY_ALWAYS_EAGER = True。

[...] tasks will be executed locally instead of being sent to the queue.

阅读这里的其他配置选项可能也有好处：http://docs.celeryproject.org/en/latest/configuration.html#celery-always-eager

Answer 2

从你的话“...我想在没有芹菜的情况下通过所有测试用例。”和“...但是我收到 rabbitMq 的连接被拒绝错误”，我猜你正在尝试运行测试而不触发真正的芹菜任务 。

如果那是真的，你不想嘲笑 apply_async。真正的需求是防止芹菜发送消息到队列触发任务。所以你可能想模拟你的芹菜应用程序send_task。

表示您在 offer 模块中有 celeryapp 个实例。您可以尝试使用 patch('offers.celeryapp.send_task', new=mock_task)

替换

patch('offers.tasks.change_offer_status.apply_async', new=mock_task).

那么你的 celery 应用程序不应该发送消息。

如果你使用的是Celery 4+（具体版本忘记了），CELERY_ALWAYS_EAGER需要改为CELERY_TASK_ALWAYS_EAGER

模拟芹菜任务方法 apply_async

Mock Celery Task method apply_async

python

django

unit-testing

mocking

celery