0
# Transaction Callbacks
1
2
Fixtures for testing Django's on_commit callbacks and transaction behavior. These fixtures allow testing of code that uses Django's transaction.on_commit() functionality.
3
4
## Capabilities
5
6
### Capture On-Commit Callbacks
7
8
Capture and inspect Django transaction on_commit callbacks during testing.
9
10
```python { .api }
11
def django_capture_on_commit_callbacks() -> DjangoCaptureOnCommitCallbacks:
12
"""
13
Fixture for capturing Django on_commit callbacks.
14
15
Returns a context manager that captures callbacks registered with
16
transaction.on_commit() during test execution. Useful for testing
17
code that defers operations until transaction commit.
18
19
Returns:
20
DjangoCaptureOnCommitCallbacks: Callback capture context manager
21
"""
22
23
class DjangoCaptureOnCommitCallbacks:
24
"""Context manager for capturing on_commit callbacks."""
25
26
def __call__(
27
self,
28
*,
29
using: str = "default",
30
execute: bool = False,
31
) -> ContextManager[List[Callable[[], Any]]]:
32
"""
33
Create context manager for capturing callbacks.
34
35
Args:
36
using: Database alias to capture callbacks for
37
execute: Whether to execute captured callbacks automatically
38
39
Returns:
40
ContextManager that yields list of captured callback functions
41
"""
42
```
43
44
Usage examples:
45
46
```python
47
from django.db import transaction
48
49
def test_on_commit_callback(django_capture_on_commit_callbacks):
50
"""Test that on_commit callbacks are registered correctly."""
51
52
def my_callback():
53
print("Transaction committed!")
54
55
# Capture callbacks registered during context
56
with django_capture_on_commit_callbacks() as callbacks:
57
# Register callback that should run on commit
58
transaction.on_commit(my_callback)
59
60
# Perform database operations
61
from myapp.models import MyModel
62
MyModel.objects.create(name="test")
63
64
# Verify callback was captured
65
assert len(callbacks) == 1
66
assert callbacks[0] == my_callback
67
68
def test_multiple_callbacks(django_capture_on_commit_callbacks):
69
"""Test multiple on_commit callbacks."""
70
71
callback_results = []
72
73
def callback_1():
74
callback_results.append("callback_1")
75
76
def callback_2():
77
callback_results.append("callback_2")
78
79
with django_capture_on_commit_callbacks() as callbacks:
80
transaction.on_commit(callback_1)
81
transaction.on_commit(callback_2)
82
83
# Some database operation
84
from myapp.models import MyModel
85
MyModel.objects.create(name="test")
86
87
# Both callbacks captured
88
assert len(callbacks) == 2
89
assert callback_1 in callbacks
90
assert callback_2 in callbacks
91
92
# Manually execute callbacks to test behavior
93
for callback in callbacks:
94
callback()
95
96
assert callback_results == ["callback_1", "callback_2"]
97
98
def test_conditional_callback(django_capture_on_commit_callbacks):
99
"""Test conditional on_commit callback registration."""
100
101
def process_model(instance, send_email=False):
102
instance.save()
103
104
if send_email:
105
def send_notification():
106
# Email sending logic here
107
pass
108
transaction.on_commit(send_notification)
109
110
from myapp.models import MyModel
111
112
# Test without email
113
with django_capture_on_commit_callbacks() as callbacks:
114
obj = MyModel(name="test")
115
process_model(obj, send_email=False)
116
117
assert len(callbacks) == 0
118
119
# Test with email
120
with django_capture_on_commit_callbacks() as callbacks:
121
obj = MyModel(name="test2")
122
process_model(obj, send_email=True)
123
124
assert len(callbacks) == 1
125
126
@pytest.mark.django_db(transaction=True)
127
def test_callback_with_rollback(django_capture_on_commit_callbacks):
128
"""Test that callbacks are not called on rollback."""
129
130
callback_called = []
131
132
def my_callback():
133
callback_called.append(True)
134
135
from myapp.models import MyModel
136
137
try:
138
with transaction.atomic():
139
with django_capture_on_commit_callbacks() as callbacks:
140
transaction.on_commit(my_callback)
141
MyModel.objects.create(name="test")
142
143
# Force rollback
144
raise Exception("Force rollback")
145
except Exception:
146
pass
147
148
# Callback was registered but not executed due to rollback
149
assert len(callbacks) == 1
150
assert len(callback_called) == 0
151
152
def test_nested_transactions(django_capture_on_commit_callbacks):
153
"""Test callbacks with nested transactions."""
154
155
outer_callback_called = []
156
inner_callback_called = []
157
158
def outer_callback():
159
outer_callback_called.append(True)
160
161
def inner_callback():
162
inner_callback_called.append(True)
163
164
from myapp.models import MyModel
165
166
with django_capture_on_commit_callbacks() as callbacks:
167
# Outer transaction callback
168
transaction.on_commit(outer_callback)
169
170
with transaction.atomic():
171
# Inner transaction callback
172
transaction.on_commit(inner_callback)
173
MyModel.objects.create(name="test")
174
175
# Both callbacks should be captured
176
assert len(callbacks) == 2
177
assert outer_callback in callbacks
178
assert inner_callback in callbacks
179
```
180
181
## Callback Testing Types
182
183
```python { .api }
184
from typing import ContextManager, List, Callable, Any
185
from django.db import transaction
186
187
# Callback function type
188
CallbackFunction = Callable[[], Any]
189
CallbackList = List[CallbackFunction]
190
191
# Callback capture context manager
192
class DjangoCaptureOnCommitCallbacks:
193
"""Context manager for capturing Django on_commit callbacks."""
194
195
def __call__(self) -> ContextManager[CallbackList]:
196
"""
197
Create context manager that captures callbacks.
198
199
Returns:
200
ContextManager that yields list of callback functions
201
registered with transaction.on_commit() during context
202
"""
203
204
def __enter__(self) -> CallbackList:
205
"""Enter context and start capturing callbacks."""
206
207
def __exit__(self, exc_type, exc_value, traceback) -> None:
208
"""Exit context and stop capturing callbacks."""
209
210
# Transaction state types
211
class TransactionState:
212
"""Represents transaction state during callback capture."""
213
in_atomic_block: bool
214
needs_rollback: bool
215
savepoint_ids: List[str]
216
217
# Internal callback management
218
class CallbackManager:
219
"""Manages callback registration and execution."""
220
callbacks: List[CallbackFunction]
221
222
def register(self, callback: CallbackFunction) -> None: ...
223
def clear(self) -> None: ...
224
def execute_all(self) -> None: ...
225
```