django - Django Celery how to ignore task?

Question

I want to perform functional testing in Django.

How can I just ignore the Celery tasks during my tests ?

class TestsFunctional(TestCase):

    def test_ignore_task(self):
        response = my_method()
        self.assertEqual(201, response)



def my_method():
    #just want to ignore tasks
    from celery import chain
    chain(tasks.long_task.s(), tasks.another_task.s()).apply_async()
    return 201


@task(default_retry_delay=10, max_retries=None)
def long_task():
    try:
        #infinite on localhost
    except socket.error:
        logger.warning("Service not reachable")
        long_task.retry()
    except Exception as e:
        logger.exception(e)

In my settings.py

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

Answer 1

要扩展Kevin Stone 的答案，您可以使用补丁装饰器和MagicMock模拟测试的 Celery 部分，如下所示：

测试代码

from unittest.mock import MagicMock, patch

@patch('reference.to.your.long_task', new=MagicMock())
class TestsFunctional(TestCase):

  def test_ignore_task(self):
      response = my_method()
      self.assertEqual(201, response) 

申请代码

def my_method():
  from celery import chain
  chain(tasks.long_task.s(), tasks.another_task.s()).apply_async()
  return 201

@task(default_retry_delay=10, max_retries=None)
def long_task():
  # Long running process

(YMMV)

Answer 2

我写了一个装饰器，我将它添加到我想在单元测试期间跳过的函数中。

https://gist.github.com/kevinastone/7295567

你的另一个选择是使用mock和模拟任务。