从另一个模块检查芹菜任务状态

我试图测试一个芹菜任务是否在请求到我的Django服务器后启动。我有这样的事情：从另一个模块检查芹菜任务状态

# tasks.py 
def add(x, y): 
    return x + y 

# views.py 
def home(request): # respond to request at root url 
    tasks.add.delay(1,2) 
    return HttpResponse('hello world') 

# tests.py 
class MyTest(TestCase): 
    def test_task_triggered(self): 
     self.client.get('/') 

     # XXXX HOW TO TEST THAT TASK HAS BEEN TRIGGERED?

我该如何测试任务是否在我的单元测试中开始？很明显，我没有直接访问任务ID，否则like this would work。

更一般地说，如何检测从各个函数，类或模块触发的芹菜任务？

感谢您的帮助。

来源

2012-03-01 dino

您可以使用Mock（单元测试时）。补充您的任务中的delay属性，然后检查模拟的called属性。在你的榜样，你可以试试：

# tests.py 
from mock import Mock, patch 

class MyTest(TestCase): 
    def test_task_triggered(self): 
     with patch('views.tasks.add.delay') as patch_mock: 
      self.client.get('/') 
     self.assertTrue(patch_mock.called)

需要注意的是，一旦你做到了这一点，tasks.add实际上不会在这个测试用例调用，因为我们已经有一个模拟取而代之。这个测试只是断言它被调用（并且你可以检查patch_mock.call_args_list来声明它被用正确的参数调用）

如果这是一个交易断路器 - 你想在一个测试案例中断言一个任务被调用，以及任务的影响/副作用 - 你可以用wraps属性mock，这可能工作实验：

# tests.py 
from mock import Mock, patch 

class MyTest(TestCase): 
    def test_task_triggered(self): 
     add_mock = Mock(wraps=tasks.add) 
     with patch('views.tasks.add.delay', new=add_mock): 
      self.client.get('/') 
     self.assertTrue(add_mock.called)

但是其他的方式可能会更好，因为它更好地隔离开什么正在测试。

来源

2012-03-04 04:52:55 Dan

我还发现'patch_mock.call_args'可用于解包传递给任务的'args'和'kwargs'。 – Matt 2015-03-07 01:27:42

Celery不允许这种交互，但您可以使用缓存或甚至会话来监视它的状态。当然，您需要从外部传递一些ID以进一步获得该价值。

但是，如果您仅需要此测试，则可以使用CELERY_ALWAYS_EAGER设置，该设置告诉Celery以简单的程序执行所有任务，而无需运行celeryd。

来源

2012-03-01 01:35:21 ilvar

不是必须有一种方法来访问不同模块之间的芹菜任务ID？毕竟，celeryctl可以列出当前正在运行的任务的任务ID。 – dino 2012-03-01 22:26:33

我发现他们有两种方法，可以使任务同步运行在我的测试

1.使用CELERY_ALWAYS_EAGER =真。

在您的测试文件的顶部，你可以将它设置这样

from django.conf import settings 
settings.CELERY_ALWAYS_EAGER = True 

from django.test.client import Client 

class MyTest(unittest.TestCase): 
    def setUp(self): 
     self.client = Client() 

    def test_sample(self): 
     self.client = client.get('/')

2.芹菜2.5，我使用mock嘲笑delay方法

这里是我的榜样

import mock 

def mock_delay(task, *args, **kwargs): 
    task.run(args, kwargs) 

@mock.patch('celery.app.task.BaseTask.delay', mock_delay) 
class MyTest(unittest.TestCase): 
    def setUp(self): 
     self.client = Client() 

    def test_sample(self): 
     self.client = client.get('/')

该解决方案适用于celery2.5.3，但是当我升级到cel ery3.0.12，它不再工作。

看起来像芹菜2.5.3改变class BaseTask到class Task在celery.app.task模块。

我修改我的代码到，但无法解决它。仍然试图找出它

来源

2012-11-10 04:05:42 kiennt

从另一个模块检查芹菜任务状态

回答

相关问题