1 files changed, 58 insertions, 2 deletions
diff --git a/tests/regressiontests/transactions_regress/tests.py b/tests/regressiontests/transactions_regress/tests.py
index abd7a4ceaa..fb26138eed 100644
--- a/tests/regressiontests/transactions_regress/tests.py
+++ b/tests/regressiontests/transactions_regress/tests.py
@@ -1,11 +1,11 @@
 from __future__ import absolute_import
 
 from django.core.exceptions import ImproperlyConfigured
-from django.db import connection, transaction
+from django.db import connection, connections, transaction, DEFAULT_DB_ALIAS
 from django.db.transaction import commit_on_success, commit_manually, TransactionManagementError
 from django.test import TransactionTestCase, skipUnlessDBFeature
 from django.test.utils import override_settings
-from django.utils.unittest import skipIf
+from django.utils.unittest import skipIf, skipUnless
 
 from .models import Mod, M2mA, M2mB
 
@@ -175,6 +175,62 @@ class TestTransactionClosing(TransactionTestCase):
         self.test_failing_query_transaction_closed()
 
 
+class TestPostgresAutocommit(TransactionTestCase):
+    """
+    Tests to make sure psycopg2's autocommit mode is restored after entering
+    and leaving transaction management. Refs #16047.
+    """
+    def setUp(self):
+        from psycopg2.extensions import (ISOLATION_LEVEL_AUTOCOMMIT,
+                                         ISOLATION_LEVEL_READ_COMMITTED)
+        self._autocommit = ISOLATION_LEVEL_AUTOCOMMIT
+        self._read_committed = ISOLATION_LEVEL_READ_COMMITTED
+
+        # 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['autocommit'] = True
+        settings['OPTIONS'] = opts
+        new_backend = self._old_backend.__class__(settings, DEFAULT_DB_ALIAS)
+        connections[DEFAULT_DB_ALIAS] = new_backend
+
+    def test_initial_autocommit_state(self):
+        self.assertTrue(connection.features.uses_autocommit)
+        self.assertEqual(connection.isolation_level, self._autocommit)
+
+    def test_transaction_management(self):
+        transaction.enter_transaction_management()
+        transaction.managed(True)
+        self.assertEqual(connection.isolation_level, self._read_committed)
+
+        transaction.leave_transaction_management()
+        self.assertEqual(connection.isolation_level, self._autocommit)
+
+    def test_transaction_stacking(self):
+        transaction.enter_transaction_management()
+        transaction.managed(True)
+        self.assertEqual(connection.isolation_level, self._read_committed)
+
+        transaction.enter_transaction_management()
+        self.assertEqual(connection.isolation_level, self._read_committed)
+
+        transaction.leave_transaction_management()
+        self.assertEqual(connection.isolation_level, self._read_committed)
+
+        transaction.leave_transaction_management()
+        self.assertEqual(connection.isolation_level, self._autocommit)
+
+    def tearDown(self):
+        connections[DEFAULT_DB_ALIAS] = self._old_backend
+
+TestPostgresAutocommit = skipUnless(connection.vendor == 'postgresql',
+    "This test only valid for PostgreSQL")(TestPostgresAutocommit)
+TestPostgresAutoCommit = skipUnlessDBFeature('supports_transactions')(
+    TestPostgresAutocommit)
+
+
 class TestManyToManyAddTransaction(TransactionTestCase):
     def test_manyrelated_add_commit(self):
         "Test for https://code.djangoproject.com/ticket/16818"