Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Merge branch 'kipanshi/R104_test_runner_storing_results_in_test_db'

  • Loading branch information...
commit e98ce282b2fef7b0325cf7e34c4f33dd03b2034e 2 parents cb13865 + 8e3d3f1
@ask ask authored
Showing with 92 additions and 3 deletions.
  1. +62 −0 djcelery/contrib/test_runner.py
  2. +30 −3 docs/cookbook/unit-testing.rst
View
62 djcelery/contrib/test_runner.py
@@ -1,8 +1,15 @@
from __future__ import absolute_import
+from uuid import uuid4
+from datetime import datetime
+
from django.conf import settings
from django.test.simple import DjangoTestSuiteRunner
+from celery.task import Task
+from djcelery.models import TaskState
+
+
USAGE = """\
Custom test runner to allow testing of celery delayed tasks.
"""
@@ -22,3 +29,58 @@ def setup_test_environment(self, **kwargs):
super(CeleryTestSuiteRunner, self).setup_test_environment(**kwargs)
settings.CELERY_ALWAYS_EAGER = True
settings.CELERY_EAGER_PROPAGATES_EXCEPTIONS = True # Issue #75
+
+
+class CeleryTestSuiteRunnerStoringResult(DjangoTestSuiteRunner):
+ """This custom test suite runner make some preliminary
+ monkey-patching allowing storing result of Celery task execution
+ in ``djcelery.models.TaskState`` model. Tasks run eagerly.
+
+ Exceptions is turned on. If you need to test ``on_failure``
+ behavior, you should monkey-patch in your test:
+ ``settings.CELERY_EAGER_PROPAGATES_EXCEPTIONS = False``
+
+ USAGE:
+ In ``settings.py``:
+ TEST_RUNNER = 'djcelery.contrib.test_runner.' \
+ 'CeleryTestSuiteRunnerStoringResult'
+
+ In ``tests.py``:
+ from djcelery.models import TaskState
+ TaskState.object.filter(state='SUCCESS', args__contains='test')
+
+ """
+ def setup_test_environment(self, **kwargs):
+ """ Setting up test environment. """
+
+ # Monkey-patch Task.on_success() method
+ def on_success_patched(self, retval, task_id, args, kwargs):
+
+ TaskState.objects.create(task_id=uuid4().hex,
+ state="SUCCESS",
+ name=self.name,
+ result=retval,
+ args=args,
+ kwargs=kwargs,
+ tstamp=datetime.now())
+ Task.on_success = classmethod(on_success_patched)
+
+ # Monkey-patch Task.on_failure() method
+ def on_failure_patched(self, exc, task_id, args, kwargs, einfo):
+
+ TaskState.objects.create(task_id=uuid4().hex,
+ state="FAILURE",
+ name=self.name,
+ result=einfo,
+ args=args,
+ kwargs=kwargs,
+ tstamp=datetime.now())
+ Task.on_failure = classmethod(on_failure_patched)
+
+ # Call parent's version
+ super(CeleryTestSuiteRunnerStoringResult,
+ self).setup_test_environment(**kwargs)
+
+ # Tell celery run tasks synchronously
+ settings.CELERY_ALWAYS_EAGER = True
+ settings.CELERY_EAGER_PROPAGATES_EXCEPTIONS = True # Issue #75
View
33 docs/cookbook/unit-testing.rst
@@ -20,10 +20,16 @@ Using a custom test runner to test with celery
If you're going the ``CELERY_ALWAYS_EAGER`` route, which is probably better than
just never testing some parts of your app, a custom Django test runner does the
-trick. Celery provides a simple test runner, but it's easy enough to roll your
-own if you have other things that need to be done.
+trick. Celery provides two simple test runner classes, but it's easy enough
+to roll your own if you have other things that need to be done.
http://docs.djangoproject.com/en/dev/topics/testing/#defining-a-test-runner
+``CeleryTestSuiteRunner`` eagerly runs all tasks in tests, but result is not stored anywhere.
+``CeleryTestSuiteRunnerStoringResult`` --- this test runner in addition stores
+result of task execution or failure in ``djcelery.models.TaskState`` model,
+likewise django-celery does during normal operation when workers
+and ``celerycam`` are launched.
+
For this example, we'll use the ``djcelery.contrib.test_runner`` to test the
``add`` task from the `User Guide: Tasks`_ examples in the Celery
documentation.
@@ -34,13 +40,15 @@ To enable the test runner, set the following settings:
.. code-block:: python
- TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'
+ TEST_RUNNER = 'djcelery.contrib.test_runner.' \
+ 'CeleryTestSuiteRunnerStoringResult'
Then we can put the tests in a ``tests.py`` somewhere:
.. code-block:: python
from django.test import TestCase
+ from djcelery.models import TaskState
from myapp.tasks import add
class AddTestCase(TestCase):
@@ -53,6 +61,25 @@ Then we can put the tests in a ``tests.py`` somewhere:
self.assertEquals(result.get(), 16)
self.assertTrue(result.successful())
+ # Run another task
+ add.delay(4, 4)
+
+ # Assert we have 2 task results in the test database
+ self.assertEqual(TaskState.objects.count(), 2)
+
+ # Assert results
+ self.assertEqual([task_state.result for task_state
+ in TaskState.objects.all()], [16, 8])
This test assumes that you put your example ``add`` task in ``maypp.tasks``
so adjust the import for wherever you put the class.
+
+If you're going to use
+``'djcelery.contrib.test_runner.CeleryTestSuiteRunnerStoringResult``
+then if your task will raise exception it will propagate through.
+If you need to test ``on_failure`` behavior of your task,
+set ``settings.CELERY_EAGER_PROPAGATES_EXCEPTIONS`` to ``False``:
+
+.. code-block:: python
+
+ settings.CELERY_EAGER_PROPAGATES_EXCEPTIONS = False
Please sign in to comment.
Something went wrong with that request. Please try again.