运行一个受监控的芹菜工人进行依赖于芹菜任务的集成测试
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