summaryrefslogtreecommitdiff
path: root/tests/transactions
diff options
context:
space:
mode:
authordjango-bot <ops@djangoproject.com>2022-02-03 20:24:19 +0100
committerMariusz Felisiak <felisiak.mariusz@gmail.com>2022-02-07 20:37:05 +0100
commit9c19aff7c7561e3a82978a272ecdaad40dda5c00 (patch)
treef0506b668a013d0063e5fba3dbf4863b466713ba /tests/transactions
parentf68fa8b45dfac545cfc4111d4e52804c86db68d3 (diff)
downloaddjango-9c19aff7c7561e3a82978a272ecdaad40dda5c00.tar.gz
Refs #33476 -- Reformatted code with Black.
Diffstat (limited to 'tests/transactions')
-rw-r--r--tests/transactions/models.py2
-rw-r--r--tests/transactions/tests.py94
2 files changed, 59 insertions, 37 deletions
diff --git a/tests/transactions/models.py b/tests/transactions/models.py
index 2f2bac463a..9506bace35 100644
--- a/tests/transactions/models.py
+++ b/tests/transactions/models.py
@@ -15,7 +15,7 @@ class Reporter(models.Model):
email = models.EmailField()
class Meta:
- ordering = ('first_name', 'last_name')
+ ordering = ("first_name", "last_name")
def __str__(self):
return ("%s %s" % (self.first_name, self.last_name)).strip()
diff --git a/tests/transactions/tests.py b/tests/transactions/tests.py
index dda4b3680a..3215ecb175 100644
--- a/tests/transactions/tests.py
+++ b/tests/transactions/tests.py
@@ -4,17 +4,24 @@ import time
from unittest import skipIf, skipUnless
from django.db import (
- DatabaseError, Error, IntegrityError, OperationalError, connection,
+ DatabaseError,
+ Error,
+ IntegrityError,
+ OperationalError,
+ connection,
transaction,
)
from django.test import (
- TestCase, TransactionTestCase, skipIfDBFeature, skipUnlessDBFeature,
+ TestCase,
+ TransactionTestCase,
+ skipIfDBFeature,
+ skipUnlessDBFeature,
)
from .models import Reporter
-@skipUnlessDBFeature('uses_savepoints')
+@skipUnlessDBFeature("uses_savepoints")
class AtomicTests(TransactionTestCase):
"""
Tests for the atomic decorator and context manager.
@@ -27,12 +34,13 @@ class AtomicTests(TransactionTestCase):
syntax and the bulk of the tests use the context manager syntax.
"""
- available_apps = ['transactions']
+ available_apps = ["transactions"]
def test_decorator_syntax_commit(self):
@transaction.atomic
def make_reporter():
return Reporter.objects.create(first_name="Tintin")
+
reporter = make_reporter()
self.assertSequenceEqual(Reporter.objects.all(), [reporter])
@@ -41,6 +49,7 @@ class AtomicTests(TransactionTestCase):
def make_reporter():
Reporter.objects.create(first_name="Haddock")
raise Exception("Oops, that's his last name")
+
with self.assertRaisesMessage(Exception, "Oops"):
make_reporter()
self.assertSequenceEqual(Reporter.objects.all(), [])
@@ -49,6 +58,7 @@ class AtomicTests(TransactionTestCase):
@transaction.atomic()
def make_reporter():
return Reporter.objects.create(first_name="Tintin")
+
reporter = make_reporter()
self.assertSequenceEqual(Reporter.objects.all(), [reporter])
@@ -57,6 +67,7 @@ class AtomicTests(TransactionTestCase):
def make_reporter():
Reporter.objects.create(first_name="Haddock")
raise Exception("Oops, that's his last name")
+
with self.assertRaisesMessage(Exception, "Oops"):
make_reporter()
self.assertSequenceEqual(Reporter.objects.all(), [])
@@ -77,7 +88,9 @@ class AtomicTests(TransactionTestCase):
with transaction.atomic():
reporter1 = Reporter.objects.create(first_name="Tintin")
with transaction.atomic():
- reporter2 = Reporter.objects.create(first_name="Archibald", last_name="Haddock")
+ reporter2 = Reporter.objects.create(
+ first_name="Archibald", last_name="Haddock"
+ )
self.assertSequenceEqual(Reporter.objects.all(), [reporter2, reporter1])
def test_nested_commit_rollback(self):
@@ -113,7 +126,9 @@ class AtomicTests(TransactionTestCase):
with transaction.atomic():
reporter1 = Reporter.objects.create(first_name="Tintin")
with transaction.atomic(savepoint=False):
- reporter2 = Reporter.objects.create(first_name="Archibald", last_name="Haddock")
+ reporter2 = Reporter.objects.create(
+ first_name="Archibald", last_name="Haddock"
+ )
self.assertSequenceEqual(Reporter.objects.all(), [reporter2, reporter1])
def test_merged_commit_rollback(self):
@@ -151,7 +166,9 @@ class AtomicTests(TransactionTestCase):
with atomic:
reporter1 = Reporter.objects.create(first_name="Tintin")
with atomic:
- reporter2 = Reporter.objects.create(first_name="Archibald", last_name="Haddock")
+ reporter2 = Reporter.objects.create(
+ first_name="Archibald", last_name="Haddock"
+ )
self.assertSequenceEqual(Reporter.objects.all(), [reporter2, reporter1])
def test_reuse_commit_rollback(self):
@@ -202,8 +219,7 @@ class AtomicTests(TransactionTestCase):
with self.assertRaises(DatabaseError):
with transaction.atomic(savepoint=False):
with connection.cursor() as cursor:
- cursor.execute(
- "SELECT no_such_col FROM transactions_reporter")
+ cursor.execute("SELECT no_such_col FROM transactions_reporter")
# prevent atomic from rolling back since we're recovering manually
self.assertTrue(transaction.get_rollback())
transaction.set_rollback(False)
@@ -235,11 +251,11 @@ class AtomicWithoutAutocommitTests(AtomicTests):
transaction.set_autocommit(True)
-@skipUnlessDBFeature('uses_savepoints')
+@skipUnlessDBFeature("uses_savepoints")
class AtomicMergeTests(TransactionTestCase):
"""Test merging transactions with savepoint=False."""
- available_apps = ['transactions']
+ available_apps = ["transactions"]
def test_merged_outer_rollback(self):
with transaction.atomic():
@@ -285,25 +301,31 @@ class AtomicMergeTests(TransactionTestCase):
self.assertSequenceEqual(Reporter.objects.all(), [reporter])
-@skipUnlessDBFeature('uses_savepoints')
+@skipUnlessDBFeature("uses_savepoints")
class AtomicErrorsTests(TransactionTestCase):
- available_apps = ['transactions']
+ available_apps = ["transactions"]
forbidden_atomic_msg = "This is forbidden when an 'atomic' block is active."
def test_atomic_prevents_setting_autocommit(self):
autocommit = transaction.get_autocommit()
with transaction.atomic():
- with self.assertRaisesMessage(transaction.TransactionManagementError, self.forbidden_atomic_msg):
+ with self.assertRaisesMessage(
+ transaction.TransactionManagementError, self.forbidden_atomic_msg
+ ):
transaction.set_autocommit(not autocommit)
# Make sure autocommit wasn't changed.
self.assertEqual(connection.autocommit, autocommit)
def test_atomic_prevents_calling_transaction_methods(self):
with transaction.atomic():
- with self.assertRaisesMessage(transaction.TransactionManagementError, self.forbidden_atomic_msg):
+ with self.assertRaisesMessage(
+ transaction.TransactionManagementError, self.forbidden_atomic_msg
+ ):
transaction.commit()
- with self.assertRaisesMessage(transaction.TransactionManagementError, self.forbidden_atomic_msg):
+ with self.assertRaisesMessage(
+ transaction.TransactionManagementError, self.forbidden_atomic_msg
+ ):
transaction.rollback()
def test_atomic_prevents_queries_in_broken_transaction(self):
@@ -321,7 +343,7 @@ class AtomicErrorsTests(TransactionTestCase):
r2.save(force_update=True)
self.assertEqual(Reporter.objects.get(pk=r1.pk).last_name, "Haddock")
- @skipIfDBFeature('atomic_transactions')
+ @skipIfDBFeature("atomic_transactions")
def test_atomic_allows_queries_after_fixing_transaction(self):
r1 = Reporter.objects.create(first_name="Archibald", last_name="Haddock")
with transaction.atomic():
@@ -333,7 +355,7 @@ class AtomicErrorsTests(TransactionTestCase):
r2.save(force_update=True)
self.assertEqual(Reporter.objects.get(pk=r1.pk).last_name, "Calculus")
- @skipUnlessDBFeature('test_db_allows_multiple_connections')
+ @skipUnlessDBFeature("test_db_allows_multiple_connections")
def test_atomic_prevents_queries_in_broken_transaction_after_client_close(self):
with transaction.atomic():
Reporter.objects.create(first_name="Archibald", last_name="Haddock")
@@ -348,10 +370,10 @@ class AtomicErrorsTests(TransactionTestCase):
self.assertEqual(Reporter.objects.count(), 0)
-@skipUnless(connection.vendor == 'mysql', "MySQL-specific behaviors")
+@skipUnless(connection.vendor == "mysql", "MySQL-specific behaviors")
class AtomicMySQLTests(TransactionTestCase):
- available_apps = ['transactions']
+ available_apps = ["transactions"]
@skipIf(threading is None, "Test requires threading")
def test_implicit_savepoint_rollback(self):
@@ -375,7 +397,7 @@ class AtomicMySQLTests(TransactionTestCase):
other_thread = threading.Thread(target=other_thread)
other_thread.start()
- with self.assertRaisesMessage(OperationalError, 'Deadlock found'):
+ with self.assertRaisesMessage(OperationalError, "Deadlock found"):
# Double atomic to enter a transaction and create a savepoint.
with transaction.atomic():
with transaction.atomic():
@@ -392,7 +414,7 @@ class AtomicMySQLTests(TransactionTestCase):
class AtomicMiscTests(TransactionTestCase):
- available_apps = ['transactions']
+ available_apps = ["transactions"]
def test_wrap_callable_instance(self):
"""#20028 -- Atomic must support wrapping callable instances."""
@@ -404,7 +426,7 @@ class AtomicMiscTests(TransactionTestCase):
# Must not raise an exception
transaction.atomic(Callable())
- @skipUnlessDBFeature('can_release_savepoints')
+ @skipUnlessDBFeature("can_release_savepoints")
def test_atomic_does_not_leak_savepoints_on_failure(self):
"""#23074 -- Savepoints must be released after rollback."""
@@ -489,8 +511,8 @@ class NonAutocommitTests(TransactionTestCase):
ORM queries are allowed after an error and a rollback in non-autocommit
mode (#27504).
"""
- r1 = Reporter.objects.create(first_name='Archibald', last_name='Haddock')
- r2 = Reporter(first_name='Cuthbert', last_name='Calculus', id=r1.id)
+ r1 = Reporter.objects.create(first_name="Archibald", last_name="Haddock")
+ r2 = Reporter(first_name="Cuthbert", last_name="Calculus", id=r1.id)
with self.assertRaises(IntegrityError):
r2.save(force_insert=True)
transaction.rollback()
@@ -502,32 +524,32 @@ class NonAutocommitTests(TransactionTestCase):
class DurableTestsBase:
- available_apps = ['transactions']
+ available_apps = ["transactions"]
def test_commit(self):
with transaction.atomic(durable=True):
- reporter = Reporter.objects.create(first_name='Tintin')
+ reporter = Reporter.objects.create(first_name="Tintin")
self.assertEqual(Reporter.objects.get(), reporter)
def test_nested_outer_durable(self):
with transaction.atomic(durable=True):
- reporter1 = Reporter.objects.create(first_name='Tintin')
+ reporter1 = Reporter.objects.create(first_name="Tintin")
with transaction.atomic():
reporter2 = Reporter.objects.create(
- first_name='Archibald',
- last_name='Haddock',
+ first_name="Archibald",
+ last_name="Haddock",
)
self.assertSequenceEqual(Reporter.objects.all(), [reporter2, reporter1])
def test_nested_both_durable(self):
- msg = 'A durable atomic block cannot be nested within another atomic block.'
+ msg = "A durable atomic block cannot be nested within another atomic block."
with transaction.atomic(durable=True):
with self.assertRaisesMessage(RuntimeError, msg):
with transaction.atomic(durable=True):
pass
def test_nested_inner_durable(self):
- msg = 'A durable atomic block cannot be nested within another atomic block.'
+ msg = "A durable atomic block cannot be nested within another atomic block."
with transaction.atomic():
with self.assertRaisesMessage(RuntimeError, msg):
with transaction.atomic(durable=True):
@@ -535,11 +557,11 @@ class DurableTestsBase:
def test_sequence_of_durables(self):
with transaction.atomic(durable=True):
- reporter = Reporter.objects.create(first_name='Tintin 1')
- self.assertEqual(Reporter.objects.get(first_name='Tintin 1'), reporter)
+ reporter = Reporter.objects.create(first_name="Tintin 1")
+ self.assertEqual(Reporter.objects.get(first_name="Tintin 1"), reporter)
with transaction.atomic(durable=True):
- reporter = Reporter.objects.create(first_name='Tintin 2')
- self.assertEqual(Reporter.objects.get(first_name='Tintin 2'), reporter)
+ reporter = Reporter.objects.create(first_name="Tintin 2")
+ self.assertEqual(Reporter.objects.get(first_name="Tintin 2"), reporter)
class DurableTransactionTests(DurableTestsBase, TransactionTestCase):