运行一个受监控的芹菜工人进行依赖于芹菜任务的集成测试
celery-testutils的Python项目详细描述
fork of cellerytest，修复了小错误
芹菜测试-芹菜集成测试
编写（集成）测试依赖于芹菜任务是有问题的。 当你手动运行芹菜工人和你的测试，它运行 在一个单独的过程中，没有干净的方法来处理对象 以芹菜为目标。当你使用单独的测试时 数据库（例如django），必须复制 配置代码，以便芹菜工人访问相同的数据库。
芹菜测试提供在后台运行芹菜工人的能力 从你的测试中。它还允许您的测试监视工作者和 暂停，直到芹菜任务完成。
使用芹菜试验
要在单独的线程中启动芹菜工作线程，请使用：
app=Celery()# your Celery appworker=start_celery_worker(app)# configure the app for our celery worker
要等待工作人员完成执行任务，请使用：
result=some_celery_task.delay()worker.idle.wait()# optionally specify time-out
django
要通过django芹菜将此应用程序与django应用程序一起使用，请将您的应用程序设置为 例如：
fromdjcelery.appimportappworker=start_celery_worker(app)
测试用例
如果您想在unittest测试用例中使用它，可以使用 CeleryTestCaseMixin。如果您编写的单元测试依赖于 不过，芹菜工人，你做错了。对于单元测试，您将 想模仿你的芹菜方法，并分别测试它们。你可以用 CelyyTestCasmixIn编写芹菜任务的集成测试， 不过。
fromunittestimportTestCasefromcelerytest.testcaseimportCeleryTestCaseMixin,setup_celery_workerimporttimeapp=Celery()setup_celery_worker(app)# need to setup worker outsideclassSomeTestCase(CeleryTestCaseMixin,TestCase):celery_app=appcelery_concurrency=4deftest_something(self):result=multiply.delay(2,3)self.worker.idle.wait()self.assertEqual(result.get(),6)
生菜
在运行 生菜集成测试套件，添加到terrain.py：
# my_celery_app.pyapp=Celery('my_celery_app',broker='amqp://')# terrain.pyfromlettuceimport*fromcelerytestimportstart_celery_worker# replace this with an import of your actual appfrommy_celery_appimportapp@before.harvestdefinitial_setup(server):# memory transport may not work hereworld.celery=start_celery_worker(app,config="amqp")@after.harvestdefcleanup(server):world.celery.stop()@after.each_stepdefafter_step(step):# make sure we've received any scheduled tasksworld.celery.active.wait(.05)# allow tasks to completeworld.celery.idle.wait(5)
安装
从pypi安装最新版本的celerytest：
$ pip install celerytest
或者，从github克隆最新版本的celerytest，然后运行 设置：
$ git clone git://github.com/RentMethod/celerytest.git $ cd celerytest $ ./setup.py install # as root
欢迎加入QQ群-->： 979659372
