Allow calling "transaction required" code without an explicit transaction.
This is useful for directly testing code marked with transaction_required
without going through other code which is responsible for managing a transaction.
This works by entering a new "atomic" block, so that the inner-most "atomic"
isn't the one created by the test-suite.
Note that this does not handle after-commit callback simulation. If you need that,
use transaction instead.
In production code and "transaction testcases" this will raise an error
to ensure we don't misleadingly run after-commit callbacks.
Source code in django_subatomic/test.py
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88 | @contextlib.contextmanager
def part_of_a_transaction(using: str | None = None) -> Generator[None]:
"""
Allow calling "transaction required" code without an explicit transaction.
This is useful for directly testing code marked with [`transaction_required`][django_subatomic.db.transaction_required]
without going through other code which is responsible for managing a transaction.
This works by entering a new "atomic" block, so that the inner-most "atomic"
isn't the one created by the test-suite.
Note that this does not handle after-commit callback simulation. If you need that,
use [`transaction`][django_subatomic.db.transaction] instead.
In production code and "transaction testcases" this will raise an error
to ensure we don't misleadingly run after-commit callbacks.
"""
connection = transaction.get_connection(using)
raise_unhandled_callbacks = getattr(
settings, "SUBATOMIC_CATCH_UNHANDLED_AFTER_COMMIT_CALLBACKS_IN_TESTS", True
)
# We must be called from inside an atomic block created by the test suite
# to avoid running after-commit callbacks on exit.
# We don't check that the atomic block is from the test suite though,
# because if it's created elsewhere we'll see an error from `durable=True` below.
if len(connection.atomic_blocks) == 0:
raise _OnlyForUseInDjangoTestTransaction
if raise_unhandled_callbacks:
callbacks = connection.run_on_commit
if callbacks:
raise _UnhandledCallbacks(tuple(callback for _, callback, _ in callbacks))
with transaction.atomic(using=using, durable=True):
atomic_block = connection.atomic_blocks[-1]
atomic_block._from_subatomic = True # noqa: SLF001
yield
# Throw away any callbacks that were registered during the partial transaction,
# so that they don't pollute later code.
# We don't need to do this in `try: ... finally:` because Django's roll
# back logic already clears the callbacks when an exception is raised.
connection.run_on_commit = []
|