Adds testing user guide

docs/userguide/index.rst

@@ -24,5 +24,6 @@
     debugging
     concurrency/index
     signals
+    testing
     extending
     configuration

docs/userguide/testing.rst

@@ -0,0 +1,264 @@
+.. _testing:
+
+================================================================
+ Testing with Celery
+================================================================
+
+Tasks and unit tests
+====================
+
+To test task behavior in unit tests the preferred method is mocking.
+
+.. admonition:: Eager mode
+
+    The eager mode enabled by the :setting:`task_always_eager` setting
+    is by definition not suitable for unit tests.
+
+    When testing with eager mode you are only testing an emulation
+    of what happens in a worker, and there are many discrepancies
+    between the emulation and what happens in reality.
+
+A Celery task is much like a web view, in that it should only
+define how to perform the action in the context of being called as a task.
+
+This means optimally tasks only handle things like serialization, message headers,
+retries, and so on, with the actual logic implemented elsewhere.
+
+Say we had a task like this:
+
+.. code-block:: python
+
+    from .models import Product
+
+
+    @app.task(bind=True)
+    def send_order(self, product_pk, quantity, price):
+        price = Decimal(price)  # json serializes this to string.
+
+        # models are passed by id, not serialized.
+        product = Product.objects.get(product_pk)
+
+        try:
+            product.order(quantity, price)
+        except OperationalError as exc:
+            raise self.retry(exc=exc)
+
+
+You could write unit tests for this task, using mocking like
+in this example:
+
+.. code-block:: python
+
+    from pytest import raises
+
+    from celery.exceptions import Retry
+
+    # for python 2: use mock.patch from `pip install mock`.
+    from unittest.mock import patch
+
+    from proj.models import Product
+    from proj.tasks import send_order
+
+    class test_send_order:
+
+        @patch('proj.tasks.Product.order')  # < patching Product in module above
+        def test_success(self, product_order):
+            product = Product.objects.create(
+                name='Foo',
+            )
+            send_order(product.pk, 3, Decimal(30.3))
+            product_order.assert_called_with(3, Decimal(30.3))
+
+        @patch('proj.tasks.Product.order')
+        @patch('proj.tasks.send_order.retry')
+        def test_failure(send_order_retry, product_order):
+            product = Product.objects.create(
+                name='Foo',
+            )
+
+            # set a side effect on the patched method
+            # so that it raises the error we want.
+            product_order.side_effect = OperationalError()
+
+            with raises(Retry):
+                send_order(product.pk, 3, Decimal(30.6))
+
+Py.test
+=======
+
+.. versionadded:: 4.0
+
+Celery is also a :pypi:`pytest` plugin that adds fixtures that you can
+use in your integration (or unit) test suites.
+
+
+Fixtures
+--------
+
+Function scope
+~~~~~~~~~~~~~~
+
+``celery_app`` - Celery app used for testing.
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This fixture returns a Celery app you can use for testing.
+
+Example:
+
+.. code-block:: python
+
+    def test_create_task(celery_app, celery_worker):
+        @celery_app.task
+        def mul(x, y):
+            return x * y
+
+        assert mul.delay(4, 4).get(timeout=10) == 16
+
+``celery_worker`` - Embed live worker.
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This fixture starts a Celery worker instance that you can use
+for integration tests.  The worker will be started in a *separate thread*
+and will be shutdown as soon as the test returns.
+
+Example:
+
+.. code-block:: python
+
+    # Put this in your confttest.py
+    @pytest.fixture(scope='session')
+    def celery_config():
+        return {
+            'broker_url': 'amqp://',
+            'result_backend': 'redis://'
+        }
+
+    def test_add(celery_worker):
+        mytask.delay()
+
+Session scope
+~~~~~~~~~~~~~
+
+``celery_config`` - Override to setup Celery test app configuration.
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+You can redefine this fixture to configure the test Celery app.
+
+The config returned by your fixture will then be used
+to configure the :func:`celery_app`, and :func:`celery_session_app` fixtures.
+
+Example:
+
+.. code-block:: python
+
+    @pytest.fixture(scope='session')
+    def celery_config():
+        return {
+            'broker_url': 'amqp://',
+            'result_backend': 'rpc',
+        }
+
+``celery_enable_logging`` - Override to enable logging in embedded workers.
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This is a fixture you can override to enable logging in embedded workers.
+
+Example:
+
+.. code-block:: python
+
+    @pytest.fixture(scope='session')
+    def celery_enable_logging():
+        return True
+
+``celery_includes`` - Add additional imports for embedded workers.
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+You can override fixture to include modules when an embedded worker starts.
+
+You can have this return a list of module names to import,
+which can be task modules, modules registering signals, and so on.
+
+Example:
+
+.. code-block:: python
+
+    @pytest.fixture(scope='session')
+    def celery_includes():
+        return [
+            'proj.tests.tasks',
+            'proj.tests.celery_signal_handlers',
+        ]
+
+``celery_worker_pool`` - Override the pool used for embedded workers.
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+You can override fixture to configure the execution pool used for embedded
+workers.
+
+Example:
+
+.. code-block:: python
+
+    @pytest.fixture(scope='session')
+    def celery_worker_pool():
+        return 'prefork'
+
+.. warning::
+
+    You cannot use the gevent/eventlet pools, that is unless your whole test
+    suite is running with the monkeypatches enabled.
+
+``celery_session_worker`` - Embedded worker that lives throughout the session.
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This fixture starts a worker that lives throughout the testing session
+(it won't be started/stopped for every test).
+
+Example:
+
+.. code-block:: python
+
+    # Add this to your conftest.py
+    @pytest.fixture(scope='session')
+    def celery_config():
+        return {
+            'broker_url': 'amqp://',
+            'result_backend': 'rpc',
+        }
+
+    # Do this in your tests.
+    def test_add_task(celery_session_worker):
+        assert add.delay(2, 2) == 4
+
+.. warning::
+
+    It's probably a bad idea to mix session and ephemeral workers...
+
+``celery_session_app`` - Celery app used for testing (session scope).
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This can be used by other session scoped fixtures when they need to refer
+to a Celery app instance.
+
+``use_celery_app_trap`` - Raise exception on falling back to default app.
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This is a fixture you can override in your ``conftest.py``, to enable the "app trap":
+if something tries to access the default or current_app, an exception
+is raised.
+
+Example:
+
+.. code-block:: python
+
+    @pytest.fixture(scope='session')
+    def use_celery_app_trap():
+        return True
+
+
+If a test wants to access the default app, you would have to mark it using
+the ``depends_on_current_app`` fixture:
+
+.. code-block::
+
+    @pytest.mark.usefixtures('depends_on_current_app')
+    def test_something():
+        something()