fixtures: add django_capture_on_commit_callbacks fixture

Similar to Django's `TestCase.captureOnCommitCallbacks`. Documentation
is cribbed from there.

Fixes #752.

Author: bluetech

docs/helpers.rst

@@ -425,6 +425,51 @@ Example usage::
             Item.objects.create('foo')
             Item.objects.create('bar')
 
+
+.. fixture:: django_capture_on_commit_callbacks
+
+``django_capture_on_commit_callbacks``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. py:function:: django_capture_on_commit_callbacks(*, using=DEFAULT_DB_ALIAS, execute=False)
+
+  :param using:
+    The alias of the database connection to capture callbacks for.
+  :param execute:
+    If True, all the callbacks will be called as the context manager exits, if
+    no exception occurred. This emulates a commit after the wrapped block of
+    code.
+
+.. versionadded:: 4.4
+
+Returns a context manager that captures
+:func:`transaction.on_commit() <django.db.transaction.on_commit>` callbacks for
+the given database connection. It returns a list that contains, on exit of the
+context, the captured callback functions. From this list you can make assertions
+on the callbacks or call them to invoke their side effects, emulating a commit.
+
+Avoid this fixture in tests using ``transaction=True``; you are not likely to
+get useful results.
+
+This fixture is based on Django's :meth:`django.test.TestCase.captureOnCommitCallbacks`
+helper.
+
+Example usage::
+
+    def test_on_commit(client, mailoutbox, django_capture_on_commit_callbacks):
+        with django_capture_on_commit_callbacks(execute=True) as callbacks:
+            response = client.post(
+                '/contact/',
+                {'message': 'I like your site'},
+            )
+
+        assert response.status_code == 200
+        assert len(callbacks) == 1
+        assert len(mailoutbox) == 1
+        assert mailoutbox[0].subject == 'Contact Form'
+        assert mailoutbox[0].body == 'I like your site'
+
+
 .. fixture:: mailoutbox
 
 ``mailoutbox``

pytest_django/fixtures.py

@@ -1,5 +1,5 @@
 """All pytest-django fixtures"""
-from typing import Any, Generator, Iterable, List, Optional, Tuple, Union
+from typing import Any, Callable, Generator, Iterable, List, Optional, Tuple, Union
 import os
 from contextlib import contextmanager
 from functools import partial
@@ -8,7 +8,7 @@
 
 from . import live_server_helper
 from .django_compat import is_django_unittest
-from .lazy_django import skip_if_no_django
+from .lazy_django import skip_if_no_django, get_django_version
 
 TYPE_CHECKING = False
 if TYPE_CHECKING:
@@ -38,6 +38,7 @@
     "_live_server_helper",
     "django_assert_num_queries",
     "django_assert_max_num_queries",
+    "django_capture_on_commit_callbacks",
 ]
 
 
@@ -542,3 +543,38 @@ def django_assert_num_queries(pytestconfig):
 @pytest.fixture(scope="function")
 def django_assert_max_num_queries(pytestconfig):
     return partial(_assert_num_queries, pytestconfig, exact=False)
+
+
+@contextmanager
+def _capture_on_commit_callbacks(
+    *,
+    using: Optional[str] = None,
+    execute: bool = False
+):
+    from django.db import DEFAULT_DB_ALIAS, connections
+    from django.test import TestCase
+
+    if using is None:
+        using = DEFAULT_DB_ALIAS
+
+    # Polyfill of Django code as of Django 3.2.
+    if get_django_version() < (3, 2):
+        callbacks = []  # type: List[Callable[[], Any]]
+        start_count = len(connections[using].run_on_commit)
+        try:
+            yield callbacks
+        finally:
+            run_on_commit = connections[using].run_on_commit[start_count:]
+            callbacks[:] = [func for sids, func in run_on_commit]
+            if execute:
+                for callback in callbacks:
+                    callback()
+
+    else:
+        with TestCase.captureOnCommitCallbacks(using=using, execute=execute) as callbacks:
+            yield callbacks
+
+
+@pytest.fixture(scope="function")
+def django_capture_on_commit_callbacks():
+    return _capture_on_commit_callbacks

pytest_django/plugin.py

@@ -25,6 +25,7 @@
 from .fixtures import django_db_modify_db_settings_parallel_suffix  # noqa
 from .fixtures import django_db_modify_db_settings_tox_suffix  # noqa
 from .fixtures import django_db_modify_db_settings_xdist_suffix  # noqa
+from .fixtures import django_capture_on_commit_callbacks  # noqa
 from .fixtures import _live_server_helper  # noqa
 from .fixtures import admin_client  # noqa
 from .fixtures import admin_user  # noqa

tests/test_fixtures.py

@@ -230,6 +230,70 @@ def test_queries(django_assert_num_queries):
     assert result.ret == 1
 
 
+@pytest.mark.django_db
+def test_django_capture_on_commit_callbacks(django_capture_on_commit_callbacks) -> None:
+    if not connection.features.supports_transactions:
+        pytest.skip("transactions required for this test")
+
+    scratch = []
+    with django_capture_on_commit_callbacks() as callbacks:
+        transaction.on_commit(lambda: scratch.append("one"))
+    assert len(callbacks) == 1
+    assert scratch == []
+    callbacks[0]()
+    assert scratch == ["one"]
+
+    scratch = []
+    with django_capture_on_commit_callbacks(execute=True) as callbacks:
+        transaction.on_commit(lambda: scratch.append("two"))
+        transaction.on_commit(lambda: scratch.append("three"))
+    assert len(callbacks) == 2
+    assert scratch == ["two", "three"]
+    callbacks[0]()
+    assert scratch == ["two", "three", "two"]
+
+
+@pytest.mark.django_db(databases=["default", "second"])
+def test_django_capture_on_commit_callbacks_multidb(django_capture_on_commit_callbacks) -> None:
+    if not connection.features.supports_transactions:
+        pytest.skip("transactions required for this test")
+
+    scratch = []
+    with django_capture_on_commit_callbacks(using="default", execute=True) as callbacks:
+        transaction.on_commit(lambda: scratch.append("one"))
+    assert len(callbacks) == 1
+    assert scratch == ["one"]
+
+    scratch = []
+    with django_capture_on_commit_callbacks(using="second", execute=True) as callbacks:
+        transaction.on_commit(lambda: scratch.append("two"))  # pragma: no cover
+    assert len(callbacks) == 0
+    assert scratch == []
+
+    scratch = []
+    with django_capture_on_commit_callbacks(using="default", execute=True) as callbacks:
+        transaction.on_commit(lambda: scratch.append("ten"))
+        transaction.on_commit(lambda: scratch.append("twenty"), using="second")  # pragma: no cover
+        transaction.on_commit(lambda: scratch.append("thirty"))
+    assert len(callbacks) == 2
+    assert scratch == ["ten", "thirty"]
+
+
+@pytest.mark.django_db(transaction=True)
+def test_django_capture_on_commit_callbacks_transactional(
+    django_capture_on_commit_callbacks,
+) -> None:
+    if not connection.features.supports_transactions:
+        pytest.skip("transactions required for this test")
+
+    # Bad usage: no transaction (executes immediately).
+    scratch = []
+    with django_capture_on_commit_callbacks() as callbacks:
+        transaction.on_commit(lambda: scratch.append("one"))
+    assert len(callbacks) == 0
+    assert scratch == ["one"]
+
+
 class TestSettings:
     """Tests for the settings fixture, order matters"""