tests for the atomic decorator and context manager the tests make assertions on internal attributes because there isn t a robust way to ask the database for its current transaction state since the decorator syntax is converted into a context manager see the implementation there are only a few basic tests with the decorator syntax and the bulk of the tests use the context manager syntax writes in the outer block are rolled back too atomic block shouldn t rollback but force it trigger a database error inside an inner atomic without savepoint prevent atomic from rolling back since we re recovering manually incorrect savepoint id to provoke a database error all basic tests for atomic should also pass within an existing transaction def setupself self atomic transaction atomic self atomic enter def teardownself self atomic exitsys excinfo class atomicwithoutautocommittestsatomictests the tests access the database after exercising atomic initiating a transaction a rollback is required before restoring autocommit test merging transactions with savepointfalse availableapps transactions def testmergedouterrollbackself with transaction atomic reporter objects createfirstnametintin with transaction atomicsavepointfalse reporter objects createfirstnamearchibald lastnamehaddock with self assertraisesmessageexception oops with transaction atomicsavepointfalse reporter objects createfirstnamecalculus raise exceptionoops that s his last name the third insert couldn t be roll back temporarily mark the connection as not needing rollback to check it self asserttruetransaction getrollback transaction setrollbackfalse self assertequalreporter objects count 3 transaction setrollbacktrue the second insert couldn t be roll back temporarily mark the connection as not needing rollback to check it self asserttruetransaction getrollback transaction setrollbackfalse self assertequalreporter objects count 3 transaction setrollbacktrue the first block has a savepoint and must roll back self assertsequenceequalreporter objects all def testmergedinnersavepointrollbackself with transaction atomic reporter reporter objects createfirstnametintin with transaction atomic reporter objects createfirstnamearchibald lastnamehaddock with self assertraisesmessageexception oops with transaction atomicsavepointfalse reporter objects createfirstnamecalculus raise exceptionoops that s his last name the third insert couldn t be roll back temporarily mark the connection as not needing rollback to check it self asserttruetransaction getrollback transaction setrollbackfalse self assertequalreporter objects count 3 transaction setrollbacktrue the second block has a savepoint and must roll back self assertequalreporter objects count 1 self assertsequenceequalreporter objects all reporter skipunlessdbfeatureusessavepoints class atomicerrorsteststransactiontestcase availableapps transactions forbiddenatomicmsg this is forbidden when an atomic block is active def testatomicpreventssettingautocommitself autocommit transaction getautocommit with transaction atomic with self assertraisesmessage transaction transactionmanagementerror self forbiddenatomicmsg transaction setautocommitnot autocommit make sure autocommit wasn t changed self assertequalconnection autocommit autocommit def testatomicpreventscallingtransactionmethodsself with transaction atomic with self assertraisesmessage transaction transactionmanagementerror self forbiddenatomicmsg transaction commit with self assertraisesmessage transaction transactionmanagementerror self forbiddenatomicmsg transaction rollback def testatomicpreventsqueriesinbrokentransactionself r1 reporter objects createfirstnamearchibald lastnamehaddock with transaction atomic r2 reporterfirstnamecuthbert lastnamecalculus idr1 id with self assertraisesintegrityerror r2 saveforceinserttrue the transaction is marked as needing rollback msg an error occurred in the current transaction you can t execute queries until the end of the atomic block with self assertraisesmessage transaction transactionmanagementerror msg as cm r2 saveforceupdatetrue self assertisinstancecm exception cause integrityerror self assertequalreporter objects getpkr1 pk lastname haddock skipifdbfeatureatomictransactions def testatomicallowsqueriesafterfixingtransactionself r1 reporter objects createfirstnamearchibald lastnamehaddock with transaction atomic r2 reporterfirstnamecuthbert lastnamecalculus idr1 id with self assertraisesintegrityerror r2 saveforceinserttrue mark the transaction as no longer needing rollback transaction setrollbackfalse r2 saveforceupdatetrue self assertequalreporter objects getpkr1 pk lastname calculus skipunlessdbfeaturetestdballowsmultipleconnections def testatomicpreventsqueriesinbrokentransactionafterclientcloseself with transaction atomic reporter objects createfirstnamearchibald lastnamehaddock connection close the connection is closed and the transaction is marked as needing rollback this will raise an interfaceerror on databases that refuse to create cursors on closed connections postgresql and a transactionmanagementerror on other databases with self assertraiseserror reporter objects createfirstnamecuthbert lastnamecalculus the connection is usable again self assertequalreporter objects count 0 skipunlessdbfeatureusessavepoints skipunlessconnection vendor mysql mysqlspecific behaviors class atomicmysqlteststransactiontestcase availableapps transactions skipifthreading is none test requires threading def testimplicitsavepointrollbackself 1 this line locks see below for 2 this is the threadlocal connection not the main connection double atomic to enter a transaction and create a savepoint the two threads can t be synchronized with an event here because the other thread locks sleep for a little while 2 and this line deadlocks see above for 1 20028 atomic must support wrapping callable instances class callable def callself pass must not raise an exception transaction atomiccallable skipunlessdbfeaturecanreleasesavepoints def testatomicdoesnotleaksavepointsonfailureself expect an error when rolling back a savepoint that doesn t exist done outside of the transaction block to ensure proper recovery start a plain transaction swallow the intentional error raised in the subtransaction start a subtransaction with a savepoint this is expected to fail because the savepoint no longer exists swallow the intentional error raised wrap in markforrollbackonerror to check if the transaction is marked broken ensure that we are still in a good state markforrollbackonerror marked the transaction as broken and further queries fail transaction errors are reset at the end of an transaction so this should just work swallow the intentional error raised wrap in markforrollbackonerror to check if the transaction is marked broken ensure that we are still in a good state ensure that markforrollbackonerror did not mark the transaction as broken since we are in autocommit mode and further queries work nicely orm queries are allowed after an error and a rollback in nonautocommit mode 27504 24921 orm queries must be possible after setautocommitfalse reporter objects createfirstnametintin class durabletestsbase availableapps transactions def testcommitself with transaction atomicdurabletrue reporter reporter objects createfirstnametintin self assertequalreporter objects get reporter def testnestedouterdurableself with transaction atomicdurabletrue reporter1 reporter objects createfirstnametintin with transaction atomic reporter2 reporter objects create firstnamearchibald lastnamehaddock self assertsequenceequalreporter objects all reporter2 reporter1 def testnestedbothdurableself msg a durable atomic block cannot be nested within another atomic block with transaction atomicdurabletrue with self assertraisesmessageruntimeerror msg with transaction atomicdurabletrue pass def testnestedinnerdurableself msg a durable atomic block cannot be nested within another atomic block with transaction atomic with self assertraisesmessageruntimeerror msg with transaction atomicdurabletrue pass def testsequenceofdurablesself with transaction atomicdurabletrue reporter reporter objects createfirstnametintin 1 self assertequalreporter objects getfirstnametintin 1 reporter with transaction atomicdurabletrue reporter reporter objects createfirstnametintin 2 self assertequalreporter objects getfirstnametintin 2 reporter class durabletransactiontestsdurabletestsbase transactiontestcase pass class durabletestsdurabletestsbase testcase pass tests for the atomic decorator and context manager the tests make assertions on internal attributes because there isn t a robust way to ask the database for its current transaction state since the decorator syntax is converted into a context manager see the implementation there are only a few basic tests with the decorator syntax and the bulk of the tests use the context manager syntax writes in the outer block are rolled back too atomic block shouldn t rollback but force it trigger a database error inside an inner atomic without savepoint prevent atomic from rolling back since we re recovering manually incorrect savepoint id to provoke a database error all basic tests for atomic should also pass within an existing transaction all basic tests for atomic should also pass when autocommit is turned off the tests access the database after exercising atomic initiating a transaction a rollback is required before restoring autocommit test merging transactions with savepoint false the third insert couldn t be roll back temporarily mark the connection as not needing rollback to check it the second insert couldn t be roll back temporarily mark the connection as not needing rollback to check it the first block has a savepoint and must roll back the third insert couldn t be roll back temporarily mark the connection as not needing rollback to check it the second block has a savepoint and must roll back make sure autocommit wasn t changed the transaction is marked as needing rollback mark the transaction as no longer needing rollback the connection is closed and the transaction is marked as needing rollback this will raise an interfaceerror on databases that refuse to create cursors on closed connections postgresql and a transactionmanagementerror on other databases the connection is usable again mysql implicitly rolls back savepoints when it deadlocks 22291 1 this line locks see below for 2 this is the thread local connection not the main connection double atomic to enter a transaction and create a savepoint the two threads can t be synchronized with an event here because the other thread locks sleep for a little while 2 and this line deadlocks see above for 1 20028 atomic must support wrapping callable instances must not raise an exception 23074 savepoints must be released after rollback expect an error when rolling back a savepoint that doesn t exist done outside of the transaction block to ensure proper recovery start a plain transaction swallow the intentional error raised in the sub transaction start a sub transaction with a savepoint this is expected to fail because the savepoint no longer exists swallow the intentional error raised wrap in mark_for_rollback_on_error to check if the transaction is marked broken ensure that we are still in a good state mark_for_rollback_on_error marked the transaction as broken and further queries fail transaction errors are reset at the end of an transaction so this should just work swallow the intentional error raised wrap in mark_for_rollback_on_error to check if the transaction is marked broken ensure that we are still in a good state ensure that mark_for_rollback_on_error did not mark the transaction as broken since we are in autocommit mode and further queries work nicely orm queries are allowed after an error and a rollback in non autocommit mode 27504 24921 orm queries must be possible after set_autocommit false
import sys
import threading
import time
from unittest import skipIf, skipUnless

from django.db import (
    DatabaseError,
    Error,
    IntegrityError,
    OperationalError,
    connection,
    transaction,
)
from django.test import (
    TestCase,
    TransactionTestCase,
    skipIfDBFeature,
    skipUnlessDBFeature,
)

from .models import Reporter


@skipUnlessDBFeature("uses_savepoints")
class AtomicTests(TransactionTestCase):
    

    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])

    def test_decorator_syntax_rollback(self):
        @transaction.atomic
        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(), [])

    def test_alternate_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])

    def test_alternate_decorator_syntax_rollback(self):
        @transaction.atomic()
        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(), [])

    def test_commit(self):
        with transaction.atomic():
            reporter = Reporter.objects.create(first_name="Tintin")
        self.assertSequenceEqual(Reporter.objects.all(), [reporter])

    def test_rollback(self):
        with self.assertRaisesMessage(Exception, "Oops"):
            with transaction.atomic():
                Reporter.objects.create(first_name="Haddock")
                raise Exception("Oops, that's his last name")
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_nested_commit_commit(self):
        with transaction.atomic():
            reporter1 = Reporter.objects.create(first_name="Tintin")
            with transaction.atomic():
                reporter2 = Reporter.objects.create(
                    first_name="Archibald", last_name="Haddock"
                )
        self.assertSequenceEqual(Reporter.objects.all(), [reporter2, reporter1])

    def test_nested_commit_rollback(self):
        with transaction.atomic():
            reporter = Reporter.objects.create(first_name="Tintin")
            with self.assertRaisesMessage(Exception, "Oops"):
                with transaction.atomic():
                    Reporter.objects.create(first_name="Haddock")
                    raise Exception("Oops, that's his last name")
        self.assertSequenceEqual(Reporter.objects.all(), [reporter])

    def test_nested_rollback_commit(self):
        with self.assertRaisesMessage(Exception, "Oops"):
            with transaction.atomic():
                Reporter.objects.create(last_name="Tintin")
                with transaction.atomic():
                    Reporter.objects.create(last_name="Haddock")
                raise Exception("Oops, that's his first name")
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_nested_rollback_rollback(self):
        with self.assertRaisesMessage(Exception, "Oops"):
            with transaction.atomic():
                Reporter.objects.create(last_name="Tintin")
                with self.assertRaisesMessage(Exception, "Oops"):
                    with transaction.atomic():
                        Reporter.objects.create(first_name="Haddock")
                    raise Exception("Oops, that's his last name")
                raise Exception("Oops, that's his first name")
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_merged_commit_commit(self):
        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"
                )
        self.assertSequenceEqual(Reporter.objects.all(), [reporter2, reporter1])

    def test_merged_commit_rollback(self):
        with transaction.atomic():
            Reporter.objects.create(first_name="Tintin")
            with self.assertRaisesMessage(Exception, "Oops"):
                with transaction.atomic(savepoint=False):
                    Reporter.objects.create(first_name="Haddock")
                    raise Exception("Oops, that's his last name")
        
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_merged_rollback_commit(self):
        with self.assertRaisesMessage(Exception, "Oops"):
            with transaction.atomic():
                Reporter.objects.create(last_name="Tintin")
                with transaction.atomic(savepoint=False):
                    Reporter.objects.create(last_name="Haddock")
                raise Exception("Oops, that's his first name")
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_merged_rollback_rollback(self):
        with self.assertRaisesMessage(Exception, "Oops"):
            with transaction.atomic():
                Reporter.objects.create(last_name="Tintin")
                with self.assertRaisesMessage(Exception, "Oops"):
                    with transaction.atomic(savepoint=False):
                        Reporter.objects.create(first_name="Haddock")
                    raise Exception("Oops, that's his last name")
                raise Exception("Oops, that's his first name")
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_reuse_commit_commit(self):
        atomic = transaction.atomic()
        with atomic:
            reporter1 = Reporter.objects.create(first_name="Tintin")
            with atomic:
                reporter2 = Reporter.objects.create(
                    first_name="Archibald", last_name="Haddock"
                )
        self.assertSequenceEqual(Reporter.objects.all(), [reporter2, reporter1])

    def test_reuse_commit_rollback(self):
        atomic = transaction.atomic()
        with atomic:
            reporter = Reporter.objects.create(first_name="Tintin")
            with self.assertRaisesMessage(Exception, "Oops"):
                with atomic:
                    Reporter.objects.create(first_name="Haddock")
                    raise Exception("Oops, that's his last name")
        self.assertSequenceEqual(Reporter.objects.all(), [reporter])

    def test_reuse_rollback_commit(self):
        atomic = transaction.atomic()
        with self.assertRaisesMessage(Exception, "Oops"):
            with atomic:
                Reporter.objects.create(last_name="Tintin")
                with atomic:
                    Reporter.objects.create(last_name="Haddock")
                raise Exception("Oops, that's his first name")
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_reuse_rollback_rollback(self):
        atomic = transaction.atomic()
        with self.assertRaisesMessage(Exception, "Oops"):
            with atomic:
                Reporter.objects.create(last_name="Tintin")
                with self.assertRaisesMessage(Exception, "Oops"):
                    with atomic:
                        Reporter.objects.create(first_name="Haddock")
                    raise Exception("Oops, that's his last name")
                raise Exception("Oops, that's his first name")
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_force_rollback(self):
        with transaction.atomic():
            Reporter.objects.create(first_name="Tintin")
            
            self.assertFalse(transaction.get_rollback())
            transaction.set_rollback(True)
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_prevent_rollback(self):
        with transaction.atomic():
            reporter = Reporter.objects.create(first_name="Tintin")
            sid = transaction.savepoint()
            
            with self.assertRaises(DatabaseError):
                with transaction.atomic(savepoint=False):
                    with connection.cursor() as cursor:
                        cursor.execute("SELECT no_such_col FROM transactions_reporter")
            
            self.assertTrue(transaction.get_rollback())
            transaction.set_rollback(False)
            transaction.savepoint_rollback(sid)
        self.assertSequenceEqual(Reporter.objects.all(), [reporter])

    @skipUnlessDBFeature("can_release_savepoints")
    def test_failure_on_exit_transaction(self):
        with transaction.atomic():
            with self.assertRaises(DatabaseError):
                with transaction.atomic():
                    Reporter.objects.create(last_name="Tintin")
                    self.assertEqual(len(Reporter.objects.all()), 1)
                    
                    connection.savepoint_ids.append("12")
            with self.assertRaises(transaction.TransactionManagementError):
                len(Reporter.objects.all())
            self.assertIs(connection.needs_rollback, True)
            if connection.savepoint_ids:
                connection.savepoint_ids.pop()
        self.assertSequenceEqual(Reporter.objects.all(), [])


class AtomicInsideTransactionTests(AtomicTests):
    

    def setUp(self):
        self.atomic = transaction.atomic()
        self.atomic.__enter__()

    def tearDown(self):
        self.atomic.__exit__(*sys.exc_info())


class AtomicWithoutAutocommitTests(AtomicTests):
    

    def setUp(self):
        transaction.set_autocommit(False)

    def tearDown(self):
        
        
        transaction.rollback()
        transaction.set_autocommit(True)


@skipUnlessDBFeature("uses_savepoints")
class AtomicMergeTests(TransactionTestCase):
    

    available_apps = ["transactions"]

    def test_merged_outer_rollback(self):
        with transaction.atomic():
            Reporter.objects.create(first_name="Tintin")
            with transaction.atomic(savepoint=False):
                Reporter.objects.create(first_name="Archibald", last_name="Haddock")
                with self.assertRaisesMessage(Exception, "Oops"):
                    with transaction.atomic(savepoint=False):
                        Reporter.objects.create(first_name="Calculus")
                        raise Exception("Oops, that's his last name")
                
                
                self.assertTrue(transaction.get_rollback())
                transaction.set_rollback(False)
                self.assertEqual(Reporter.objects.count(), 3)
                transaction.set_rollback(True)
            
            
            self.assertTrue(transaction.get_rollback())
            transaction.set_rollback(False)
            self.assertEqual(Reporter.objects.count(), 3)
            transaction.set_rollback(True)
        
        self.assertSequenceEqual(Reporter.objects.all(), [])

    def test_merged_inner_savepoint_rollback(self):
        with transaction.atomic():
            reporter = Reporter.objects.create(first_name="Tintin")
            with transaction.atomic():
                Reporter.objects.create(first_name="Archibald", last_name="Haddock")
                with self.assertRaisesMessage(Exception, "Oops"):
                    with transaction.atomic(savepoint=False):
                        Reporter.objects.create(first_name="Calculus")
                        raise Exception("Oops, that's his last name")
                
                
                self.assertTrue(transaction.get_rollback())
                transaction.set_rollback(False)
                self.assertEqual(Reporter.objects.count(), 3)
                transaction.set_rollback(True)
            
            self.assertEqual(Reporter.objects.count(), 1)
        self.assertSequenceEqual(Reporter.objects.all(), [reporter])


@skipUnlessDBFeature("uses_savepoints")
class AtomicErrorsTests(TransactionTestCase):
    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
            ):
                transaction.set_autocommit(not autocommit)
        
        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
            ):
                transaction.commit()
            with self.assertRaisesMessage(
                transaction.TransactionManagementError, self.forbidden_atomic_msg
            ):
                transaction.rollback()

    def test_atomic_prevents_queries_in_broken_transaction(self):
        r1 = Reporter.objects.create(first_name="Archibald", last_name="Haddock")
        with transaction.atomic():
            r2 = Reporter(first_name="Cuthbert", last_name="Calculus", id=r1.id)
            with self.assertRaises(IntegrityError):
                r2.save(force_insert=True)
            
            msg = (
                "An error occurred in the current transaction. You can't "
                "execute queries until the end of the 'atomic' block."
            )
            with self.assertRaisesMessage(
                transaction.TransactionManagementError, msg
            ) as cm:
                r2.save(force_update=True)
        self.assertIsInstance(cm.exception.__cause__, IntegrityError)
        self.assertEqual(Reporter.objects.get(pk=r1.pk).last_name, "Haddock")

    @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():
            r2 = Reporter(first_name="Cuthbert", last_name="Calculus", id=r1.id)
            with self.assertRaises(IntegrityError):
                r2.save(force_insert=True)
            
            transaction.set_rollback(False)
            r2.save(force_update=True)
        self.assertEqual(Reporter.objects.get(pk=r1.pk).last_name, "Calculus")

    @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")
            connection.close()
            
            
            
            
            with self.assertRaises(Error):
                Reporter.objects.create(first_name="Cuthbert", last_name="Calculus")
        
        self.assertEqual(Reporter.objects.count(), 0)


@skipUnlessDBFeature("uses_savepoints")
@skipUnless(connection.vendor == "mysql", "MySQL-specific behaviors")
class AtomicMySQLTests(TransactionTestCase):
    available_apps = ["transactions"]

    @skipIf(threading is None, "Test requires threading")
    def test_implicit_savepoint_rollback(self):
        
        Reporter.objects.create(id=1)
        Reporter.objects.create(id=2)

        main_thread_ready = threading.Event()

        def other_thread():
            try:
                with transaction.atomic():
                    Reporter.objects.select_for_update().get(id=1)
                    main_thread_ready.wait()
                    
                    Reporter.objects.exclude(id=1).update(id=2)
            finally:
                
                connection.close()

        other_thread = threading.Thread(target=other_thread)
        other_thread.start()

        with self.assertRaisesMessage(OperationalError, "Deadlock found"):
            
            with transaction.atomic():
                with transaction.atomic():
                    Reporter.objects.select_for_update().get(id=2)
                    main_thread_ready.set()
                    
                    
                    time.sleep(1)
                    
                    Reporter.objects.exclude(id=2).update(id=1)

        other_thread.join()


class AtomicMiscTests(TransactionTestCase):
    available_apps = ["transactions"]

    def test_wrap_callable_instance(self):
        

        class Callable:
            def __call__(self):
                pass

        
        transaction.atomic(Callable())

    @skipUnlessDBFeature("can_release_savepoints")
    def test_atomic_does_not_leak_savepoints_on_failure(self):
        

        
        
        with self.assertRaises(Error):
            
            with transaction.atomic():
                
                with self.assertRaisesMessage(Exception, "Oops"):
                    
                    with transaction.atomic():
                        sid = connection.savepoint_ids[-1]
                        raise Exception("Oops")

                
                connection.savepoint_rollback(sid)

    def test_mark_for_rollback_on_error_in_transaction(self):
        with transaction.atomic(savepoint=False):
            
            with self.assertRaisesMessage(Exception, "Oops"):
                
                
                with transaction.mark_for_rollback_on_error():
                    
                    self.assertFalse(transaction.get_rollback())

                    raise Exception("Oops")

                
                self.assertTrue(transaction.get_rollback())

            
            msg = "You can't execute queries until the end of the 'atomic' block."
            with self.assertRaisesMessage(transaction.TransactionManagementError, msg):
                Reporter.objects.create()

        
        
        Reporter.objects.create()

    def test_mark_for_rollback_on_error_in_autocommit(self):
        self.assertTrue(transaction.get_autocommit())

        
        with self.assertRaisesMessage(Exception, "Oops"):
            
            
            with transaction.mark_for_rollback_on_error():
                
                self.assertFalse(transaction.get_connection().needs_rollback)

                raise Exception("Oops")

            
            
            self.assertFalse(transaction.get_connection().needs_rollback)

        
        Reporter.objects.create()


class NonAutocommitTests(TransactionTestCase):
    available_apps = []

    def setUp(self):
        transaction.set_autocommit(False)

    def tearDown(self):
        transaction.rollback()
        transaction.set_autocommit(True)

    def test_orm_query_after_error_and_rollback(self):
        
        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()
        Reporter.objects.last()

    def test_orm_query_without_autocommit(self):
        
        Reporter.objects.create(first_name="Tintin")


class DurableTestsBase:
    available_apps = ["transactions"]

    def test_commit(self):
        with transaction.atomic(durable=True):
            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")
            with transaction.atomic():
                reporter2 = Reporter.objects.create(
                    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."
        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."
        with transaction.atomic():
            with self.assertRaisesMessage(RuntimeError, msg):
                with transaction.atomic(durable=True):
                    pass

    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)
        with transaction.atomic(durable=True):
            reporter = Reporter.objects.create(first_name="Tintin 2")
        self.assertEqual(Reporter.objects.get(first_name="Tintin 2"), reporter)


class DurableTransactionTests(DurableTestsBase, TransactionTestCase):
    pass


class DurableTests(DurableTestsBase, TestCase):
    pass