commit_on_success_with_read_committed should only work with atmost

1 level of transactions.

TNL-367

common/djangoapps/util/db.py

@@ -13,6 +13,8 @@ def commit_on_success_with_read_committed(func):  # pylint: disable=invalid-name
     If the function returns a response the transaction is committed and if the function raises an exception the
     transaction is rolled back.
 
+    Raises TransactionManagementError if there are already more than 1 level of transactions open.
+
     Note: This only works on MySQL.
     """
     @wraps(func)
@@ -21,7 +23,12 @@ def commit_on_success_with_read_committed(func):  # pylint: disable=invalid-name
         if connection.vendor == 'mysql':
             # The isolation level cannot be changed while a transaction is in progress. So we close any existing one.
             if connection.transaction_state:
-                connection.commit()
+                if len(connection.transaction_state) == 1:
+                    connection.commit()
+                # We can commit all open transactions. But it does not seem like a good idea.
+                elif len(connection.transaction_state) > 1:
+                    raise transaction.TransactionManagementError('Cannot change isolation level. '
+                                                                 'More than 1 level of nested transactions.')
 
             # This will set the transaction isolation level to READ COMMITTED for the next transaction.
             cursor = connection.cursor()

common/djangoapps/util/tests/test_db.py

@@ -7,7 +7,7 @@ import unittest
 
 from django.contrib.auth.models import User
 from django.db import connection, IntegrityError
-from django.db.transaction import commit_on_success
+from django.db.transaction import commit_on_success, TransactionManagementError
 from django.test import TransactionTestCase
 
 from util.db import commit_on_success_with_read_committed
@@ -79,3 +79,23 @@ class TransactionIsolationLevelsTestCase(TransactionTestCase):
 
         self.assertIsNone(thread2.status.get('exception'))
         self.assertEqual(thread2.status.get('created'), created_in_2)
+
+    def test_transaction_nesting(self):
+        """Test that the decorator raises an error if there are already more than 1 levels of nested transactions."""
+
+        if connection.vendor != 'mysql':
+            raise unittest.SkipTest('Only works on MySQL.')
+
+        def do_nothing():
+            """Just return."""
+            return
+
+        commit_on_success_with_read_committed(do_nothing)()
+
+        with commit_on_success():
+            commit_on_success_with_read_committed(do_nothing)()
+
+        with self.assertRaises(TransactionManagementError):
+            with commit_on_success():
+                with commit_on_success():
+                    commit_on_success_with_read_committed(do_nothing)()