django/tests/transactions_regress/tests.py

from __future__ import absolute_import

from django.db import (connection, connections, transaction, DEFAULT_DB_ALIAS, DatabaseError,
                       IntegrityError)
from django.db.transaction import commit_on_success, commit_manually, TransactionManagementError
from django.test import TransactionTestCase, skipUnlessDBFeature
from django.test.utils import override_settings, IgnorePendingDeprecationWarningsMixin
from django.utils.unittest import skipIf, skipUnless

from .models import Mod, M2mA, M2mB, SubMod

class ModelInheritanceTests(TransactionTestCase):
    def test_save(self):
        # First, create a SubMod, then try to save another with conflicting
        # cnt field. The problem was that transactions were committed after
        # every parent save when not in managed transaction. As the cnt
        # conflict is in the second model, we can check if the first save
        # was committed or not.
        SubMod(fld=1, cnt=1).save()
        # We should have committed the transaction for the above - assert this.
        connection.rollback()
        self.assertEqual(SubMod.objects.count(), 1)
        try:
            SubMod(fld=2, cnt=1).save()
        except IntegrityError:
            connection.rollback()
        self.assertEqual(SubMod.objects.count(), 1)
        self.assertEqual(Mod.objects.count(), 1)

class TestTransactionClosing(IgnorePendingDeprecationWarningsMixin, TransactionTestCase):
    """
    Tests to make sure that transactions are properly closed
    when they should be, and aren't left pending after operations
    have been performed in them. Refs #9964.
    """
    def test_raw_committed_on_success(self):
        """
        Make sure a transaction consisting of raw SQL execution gets
        committed by the commit_on_success decorator.
        """
        @commit_on_success
        def raw_sql():
            "Write a record using raw sql under a commit_on_success decorator"
            cursor = connection.cursor()
            cursor.execute("INSERT into transactions_regress_mod (fld) values (18)")

        raw_sql()
        # Rollback so that if the decorator didn't commit, the record is unwritten
        transaction.rollback()
        self.assertEqual(Mod.objects.count(), 1)
        # Check that the record is in the DB
        obj = Mod.objects.all()[0]
        self.assertEqual(obj.fld, 18)

    def test_commit_manually_enforced(self):
        """
        Make sure that under commit_manually, even "read-only" transaction require closure
        (commit or rollback), and a transaction left pending is treated as an error.
        """
        @commit_manually
        def non_comitter():
            "Execute a managed transaction with read-only operations and fail to commit"
            Mod.objects.count()

        self.assertRaises(TransactionManagementError, non_comitter)

    def test_commit_manually_commit_ok(self):
        """
        Test that under commit_manually, a committed transaction is accepted by the transaction
        management mechanisms
        """
        @commit_manually
        def committer():
            """
            Perform a database query, then commit the transaction
            """
            Mod.objects.count()
            transaction.commit()

        try:
            committer()
        except TransactionManagementError:
            self.fail("Commit did not clear the transaction state")

    def test_commit_manually_rollback_ok(self):
        """
        Test that under commit_manually, a rolled-back transaction is accepted by the transaction
        management mechanisms
        """
        @commit_manually
        def roller_back():
            """
            Perform a database query, then rollback the transaction
            """
            Mod.objects.count()
            transaction.rollback()

        try:
            roller_back()
        except TransactionManagementError:
            self.fail("Rollback did not clear the transaction state")

    def test_commit_manually_enforced_after_commit(self):
        """
        Test that under commit_manually, if a transaction is committed and an operation is
        performed later, we still require the new transaction to be closed
        """
        @commit_manually
        def fake_committer():
            "Query, commit, then query again, leaving with a pending transaction"
            Mod.objects.count()
            transaction.commit()
            Mod.objects.count()

        self.assertRaises(TransactionManagementError, fake_committer)

    @skipUnlessDBFeature('supports_transactions')
    def test_reuse_cursor_reference(self):
        """
        Make sure transaction closure is enforced even when the queries are performed
        through a single cursor reference retrieved in the beginning
        (this is to show why it is wrong to set the transaction dirty only when a cursor
        is fetched from the connection).
        """
        @commit_on_success
        def reuse_cursor_ref():
            """
            Fetch a cursor, perform an query, rollback to close the transaction,
            then write a record (in a new transaction) using the same cursor object
            (reference). All this under commit_on_success, so the second insert should
            be committed.
            """
            cursor = connection.cursor()
            cursor.execute("INSERT into transactions_regress_mod (fld) values (2)")
            transaction.rollback()
            cursor.execute("INSERT into transactions_regress_mod (fld) values (2)")

        reuse_cursor_ref()
        # Rollback so that if the decorator didn't commit, the record is unwritten
        transaction.rollback()
        self.assertEqual(Mod.objects.count(), 1)
        obj = Mod.objects.all()[0]
        self.assertEqual(obj.fld, 2)

    def test_failing_query_transaction_closed(self):
        """
        Make sure that under commit_on_success, a transaction is rolled back even if
        the first database-modifying operation fails.
        This is prompted by http://code.djangoproject.com/ticket/6669 (and based on sample
        code posted there to exemplify the problem): Before Django 1.3,
        transactions were only marked "dirty" by the save() function after it successfully
        wrote the object to the database.
        """
        from django.contrib.auth.models import User

        @transaction.commit_on_success
        def create_system_user():
            "Create a user in a transaction"
            user = User.objects.create_user(username='system', password='iamr00t',
                                            email='root@SITENAME.com')
            # Redundant, just makes sure the user id was read back from DB
            Mod.objects.create(fld=user.pk)

        # Create a user
        create_system_user()

        with self.assertRaises(DatabaseError):
            # The second call to create_system_user should fail for violating
            # a unique constraint (it's trying to re-create the same user)
            create_system_user()

        # Try to read the database. If the last transaction was indeed closed,
        # this should cause no problems
        User.objects.all()[0]

    @override_settings(DEBUG=True)
    def test_failing_query_transaction_closed_debug(self):
        """
        Regression for #6669. Same test as above, with DEBUG=True.
        """
        self.test_failing_query_transaction_closed()

@skipIf(connection.vendor == 'sqlite' and
        (connection.settings_dict['NAME'] == ':memory:' or
         not connection.settings_dict['NAME']),
        'Test uses multiple connections, but in-memory sqlite does not support this')
class TestNewConnection(IgnorePendingDeprecationWarningsMixin, TransactionTestCase):
    """
    Check that new connections don't have special behaviour.
    """
    def setUp(self):
        self._old_backend = connections[DEFAULT_DB_ALIAS]
        settings = self._old_backend.settings_dict.copy()
        new_backend = self._old_backend.__class__(settings, DEFAULT_DB_ALIAS)
        connections[DEFAULT_DB_ALIAS] = new_backend

    def tearDown(self):
        try:
            connections[DEFAULT_DB_ALIAS].abort()
            connections[DEFAULT_DB_ALIAS].close()
        finally:
            connections[DEFAULT_DB_ALIAS] = self._old_backend

    def test_commit(self):
        """
        Users are allowed to commit and rollback connections.
        """
        connection.set_autocommit(False)
        try:
            # The starting value is False, not None.
            self.assertIs(connection._dirty, False)
            list(Mod.objects.all())
            self.assertTrue(connection.is_dirty())
            connection.commit()
            self.assertFalse(connection.is_dirty())
            list(Mod.objects.all())
            self.assertTrue(connection.is_dirty())
            connection.rollback()
            self.assertFalse(connection.is_dirty())
        finally:
            connection.set_autocommit(True)

    def test_enter_exit_management(self):
        orig_dirty = connection._dirty
        connection.enter_transaction_management()
        connection.leave_transaction_management()
        self.assertEqual(orig_dirty, connection._dirty)


@skipUnless(connection.vendor == 'postgresql',
            "This test only valid for PostgreSQL")
class TestPostgresAutocommitAndIsolation(IgnorePendingDeprecationWarningsMixin, TransactionTestCase):
    """
    Tests to make sure psycopg2's autocommit mode and isolation level
    is restored after entering and leaving transaction management.
    Refs #16047, #18130.
    """
    def setUp(self):
        from psycopg2.extensions import (ISOLATION_LEVEL_AUTOCOMMIT,
                                         ISOLATION_LEVEL_SERIALIZABLE,
                                         TRANSACTION_STATUS_IDLE)
        self._autocommit = ISOLATION_LEVEL_AUTOCOMMIT
        self._serializable = ISOLATION_LEVEL_SERIALIZABLE
        self._idle = TRANSACTION_STATUS_IDLE

        # We want a clean backend with autocommit = True, so
        # first we need to do a bit of work to have that.
        self._old_backend = connections[DEFAULT_DB_ALIAS]
        settings = self._old_backend.settings_dict.copy()
        opts = settings['OPTIONS'].copy()
        opts['isolation_level'] = ISOLATION_LEVEL_SERIALIZABLE
        settings['OPTIONS'] = opts
        new_backend = self._old_backend.__class__(settings, DEFAULT_DB_ALIAS)
        connections[DEFAULT_DB_ALIAS] = new_backend

    def tearDown(self):
        try:
            connections[DEFAULT_DB_ALIAS].abort()
        finally:
            connections[DEFAULT_DB_ALIAS].close()
            connections[DEFAULT_DB_ALIAS] = self._old_backend

    def test_initial_autocommit_state(self):
        # Autocommit is activated when the connection is created.
        connection.cursor().close()
        self.assertTrue(connection.autocommit)

    def test_transaction_management(self):
        transaction.enter_transaction_management()
        self.assertFalse(connection.autocommit)
        self.assertEqual(connection.isolation_level, self._serializable)

        transaction.leave_transaction_management()
        self.assertTrue(connection.autocommit)

    def test_transaction_stacking(self):
        transaction.enter_transaction_management()
        self.assertFalse(connection.autocommit)
        self.assertEqual(connection.isolation_level, self._serializable)

        transaction.enter_transaction_management()
        self.assertFalse(connection.autocommit)
        self.assertEqual(connection.isolation_level, self._serializable)

        transaction.leave_transaction_management()
        self.assertFalse(connection.autocommit)
        self.assertEqual(connection.isolation_level, self._serializable)

        transaction.leave_transaction_management()
        self.assertTrue(connection.autocommit)

    def test_enter_autocommit(self):
        transaction.enter_transaction_management()
        self.assertFalse(connection.autocommit)
        self.assertEqual(connection.isolation_level, self._serializable)
        list(Mod.objects.all())
        self.assertTrue(transaction.is_dirty())
        # Enter autocommit mode again.
        transaction.enter_transaction_management(False)
        self.assertFalse(transaction.is_dirty())
        self.assertEqual(
            connection.connection.get_transaction_status(),
            self._idle)
        list(Mod.objects.all())
        self.assertFalse(transaction.is_dirty())
        transaction.leave_transaction_management()
        self.assertFalse(connection.autocommit)
        self.assertEqual(connection.isolation_level, self._serializable)
        transaction.leave_transaction_management()
        self.assertTrue(connection.autocommit)


class TestManyToManyAddTransaction(IgnorePendingDeprecationWarningsMixin, TransactionTestCase):
    def test_manyrelated_add_commit(self):
        "Test for https://code.djangoproject.com/ticket/16818"
        a = M2mA.objects.create()
        b = M2mB.objects.create(fld=10)
        a.others.add(b)

        # We're in a TransactionTestCase and have not changed transaction
        # behavior from default of "autocommit", so this rollback should not
        # actually do anything. If it does in fact undo our add, that's a bug
        # that the bulk insert was not auto-committed.
        transaction.rollback()
        self.assertEqual(a.others.count(), 1)


class SavepointTest(IgnorePendingDeprecationWarningsMixin, TransactionTestCase):

    @skipIf(connection.vendor == 'sqlite',
            "SQLite doesn't support savepoints in managed mode")
    @skipUnlessDBFeature('uses_savepoints')
    def test_savepoint_commit(self):
        @commit_manually
        def work():
            mod = Mod.objects.create(fld=1)
            pk = mod.pk
            sid = transaction.savepoint()
            Mod.objects.filter(pk=pk).update(fld=10)
            transaction.savepoint_commit(sid)
            mod2 = Mod.objects.get(pk=pk)
            transaction.commit()
            self.assertEqual(mod2.fld, 10)

        work()

    @skipIf(connection.vendor == 'sqlite',
            "SQLite doesn't support savepoints in managed mode")
    @skipIf(connection.vendor == 'mysql' and
            connection.features._mysql_storage_engine == 'MyISAM',
            "MyISAM MySQL storage engine doesn't support savepoints")
    @skipUnlessDBFeature('uses_savepoints')
    def test_savepoint_rollback(self):
        @commit_manually
        def work():
            mod = Mod.objects.create(fld=1)
            pk = mod.pk
            sid = transaction.savepoint()
            Mod.objects.filter(pk=pk).update(fld=20)
            transaction.savepoint_rollback(sid)
            mod2 = Mod.objects.get(pk=pk)
            transaction.commit()
            self.assertEqual(mod2.fld, 1)

        work()