WIP: Add support for sharing DB data across across a scope

pytest_django/fixtures.py

@@ -329,6 +329,13 @@ def django_db_reset_sequences(
     # is requested.
 
 
+@pytest.fixture(scope='session')
+def django_test_data() -> None:
+    """A fixture that may be requested by other fixtures to allow
+    them to modify the database, with all database modifications
+    rolled back at the end of the scope."""
+
+
 @pytest.fixture()
 def client() -> "django.test.client.Client":
     """A Django test client instance."""

pytest_django/plugin.py

@@ -37,6 +37,7 @@
 from .fixtures import django_db_use_migrations  # noqa
 from .fixtures import django_user_model  # noqa
 from .fixtures import django_username_field  # noqa
+from .fixtures import django_test_data  # noqa
 from .fixtures import live_server  # noqa
 from .fixtures import rf  # noqa
 from .fixtures import settings  # noqa
@@ -740,3 +741,38 @@ def apifun(urls: List[str]) -> List[str]:
         return urls
 
     return apifun(*marker.args, **marker.kwargs)
+
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_fixture_setup(fixturedef, request):
+    # Implement the magic of `django_test_data`.
+    # When a user fixture *directly* requests `django_test_data`,
+    # its body is wrapped in an atomic block, which is rolled back
+    # at the end of the fixture's scope (*not* the fixture itself!).
+
+    if 'django_test_data' not in fixturedef.argnames:
+        yield
+        return
+
+    from django.db import transaction
+
+    atomics = {}
+    # TODO
+    databases = ('default',)
+    django_db_blocker = request.getfixturevalue('django_db_blocker')
+
+    with django_db_blocker.unblock():
+        for db_name in databases:
+            atomic = transaction.atomic(using=db_name)
+            atomic._from_testcase = True
+            atomic.__enter__()
+            atomics[db_name] = atomic
+
+        yield
+
+    def finalize_django_test_data():
+        with django_db_blocker.unblock():
+            for db_name in reversed(databases):
+                transaction.set_rollback(True, using=db_name)
+                atomics[db_name].__exit__(None, None, None)
+    request.node.addfinalizer(finalize_django_test_data)