1
# Copyright 2012 Canonical Ltd. This software is licensed under the
2
# GNU Affero General Public License version 3 (see the file LICENSE).
4
"""A fixture to make Celery run tasks in a synchronous fashion."""
6
from __future__ import (
21
from fixtures import Fixture
22
from testtools.monkey import MonkeyPatcher
25
class CeleryFixture(Fixture):
26
"""This fixture will make Celery run tasks in a synchronous fashion.
28
This fixture can be used directly::
30
>>> class CeleryTest1(TestCase):
33
>>> super(CeleryTest1, self).setUp()
34
>>> self.useFixture(CeleryFixture())
36
It can also be converted into a FixtureResource::
38
>>> from testresources import FixtureResource
40
>>> class CeleryTest2(TestCase):
43
>>> ("celery", FixtureResource(CeleryFixture())),
49
super(CeleryFixture, self).setUp()
54
patcher = MonkeyPatcher()
55
patcher.add_patch(current_app.conf, 'CELERY_ALWAYS_EAGER', True)
57
current_app.conf, 'CELERY_EAGER_PROPAGATES_EXCEPTIONS', True)
58
self.addCleanup(patcher.restore)
61
def record_tasks(self):
64
def on_task_postrun(**kwargs):
65
self.tasks.append(kwargs)
66
signals.task_postrun.connect(on_task_postrun, weak=False)
67
self.addCleanup(lambda: self.cleanup_tasks())
69
def cleanup_tasks(self):