Fix 'No transaction found' error on Synapse.

Fixed issue where the SQL Server dialect's query for the current isolation
level would fail on Azure Synapse Analytics, due to the way in which this
database handles transaction rollbacks after an error has occurred. The
initial query has been modified to no longer rely upon catching an error
when attempting to detect the appropriate system view. Additionally, to
better support this database's very specific "rollback" behavior,
implemented new parameter ``ignore_no_transaction_on_rollback`` indicating
that a rollback should ignore Azure Synapse error 'No corresponding
transaction found. (111214)', which is raised if no transaction is present
in conflict with the Python DBAPI.

Fixes: #8231
Closes: #8233
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/8233
Pull-request-sha: c48bd44a9f53d00e5e94f1b8bf996711b6419562

Change-Id: I6407a03148f45cc9eba8fe1d31d4f59ebf9c7ef7

1 files changed, 63 insertions, 11 deletions

diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py
index 967ca85fe..028920203 100644
--- a/test/dialect/mssql/test_engine.py
+++ b/test/dialect/mssql/test_engine.py
@@ -1,6 +1,7 @@
 # -*- encoding: utf-8
 
 from decimal import Decimal
+import re
 from unittest.mock import Mock
 
 from sqlalchemy import Column
@@ -24,6 +25,7 @@ from sqlalchemy.testing import assert_warnings
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import expect_raises
+from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import mock
@@ -645,7 +647,7 @@ class RealIsolationLevelTest(fixtures.TestBase):
 
 
 class IsolationLevelDetectTest(fixtures.TestBase):
-    def _fixture(self, view):
+    def _fixture(self, view_result):
         class Error(Exception):
             pass
 
@@ -658,15 +660,25 @@ class IsolationLevelDetectTest(fixtures.TestBase):
         def fail_on_exec(
             stmt,
         ):
-            if view is not None and view in stmt:
+            result.clear()
+            if "SELECT name FROM sys.system_views" in stmt:
+                if view_result:
+                    result.append((view_result,))
+            elif re.match(
+                ".*SELECT CASE transaction_isolation_level.*FROM sys.%s"
+                % (view_result,),
+                stmt,
+                re.S,
+            ):
                 result.append(("SERIALIZABLE",))
             else:
-                raise Error("that didn't work")
+                assert False
 
         connection = Mock(
             cursor=Mock(
                 return_value=Mock(
-                    execute=fail_on_exec, fetchone=lambda: result[0]
+                    execute=fail_on_exec,
+                    fetchone=lambda: result[0] if result else None,
                 )
             )
         )
@@ -686,13 +698,12 @@ class IsolationLevelDetectTest(fixtures.TestBase):
     def test_not_supported(self):
         dialect, connection = self._fixture(None)
 
-        with expect_warnings("Could not fetch transaction isolation level"):
-            assert_raises_message(
-                NotImplementedError,
-                "Can't fetch isolation",
-                dialect.get_isolation_level,
-                connection,
-            )
+        assert_raises_message(
+            NotImplementedError,
+            "Can't fetch isolation level on this particular ",
+            dialect.get_isolation_level,
+            connection,
+        )
 
 
 class InvalidTransactionFalsePositiveTest(fixtures.TablesTest):
@@ -730,3 +741,44 @@ class InvalidTransactionFalsePositiveTest(fixtures.TablesTest):
         # "Can't reconnect until invalid transaction is rolled back."
         result = connection.execute(t.select()).fetchall()
         eq_(len(result), 1)
+
+
+class IgnoreNotransOnRollbackTest(fixtures.TestBase):
+    def test_ignore_no_transaction_on_rollback(self):
+        """test #8231"""
+
+        class ProgrammingError(Exception):
+            pass
+
+        dialect = base.dialect(ignore_no_transaction_on_rollback=True)
+        dialect.dbapi = mock.Mock(ProgrammingError=ProgrammingError)
+
+        connection = mock.Mock(
+            rollback=mock.Mock(
+                side_effect=ProgrammingError("Error 111214 happened")
+            )
+        )
+        with expect_warnings(
+            "ProgrammingError 111214 'No corresponding transaction found.' "
+            "has been suppressed via ignore_no_transaction_on_rollback=True"
+        ):
+            dialect.do_rollback(connection)
+
+    def test_other_programming_error_on_rollback(self):
+        """test #8231"""
+
+        class ProgrammingError(Exception):
+            pass
+
+        dialect = base.dialect(ignore_no_transaction_on_rollback=True)
+        dialect.dbapi = mock.Mock(ProgrammingError=ProgrammingError)
+
+        connection = mock.Mock(
+            rollback=mock.Mock(
+                side_effect=ProgrammingError("Some other error happened")
+            )
+        )
+        with expect_raises_message(
+            ProgrammingError, "Some other error happened"
+        ):
+            dialect.do_rollback(connection)