Fix is_disconnect false positive for mssql+pyodbc

Fixed an issue where the ``is_disconnect`` function in the SQL Server
pyodbc dialect was incorrectly reporting the disconnect state when the
exception messsage had a substring that matched a SQL Server ODBC error
code.

Fixes: #5359
Change-Id: I450c6818405a20f4daee20d58fce2d5ecb33e17f

1 files changed, 43 insertions, 2 deletions

diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py
index 0dea2688a..734224ed1 100644
--- a/test/dialect/mssql/test_engine.py
+++ b/test/dialect/mssql/test_engine.py
@@ -1,4 +1,5 @@
 # -*- encoding: utf-8
+
 from sqlalchemy import Column
 from sqlalchemy import engine_from_config
 from sqlalchemy import event
@@ -11,6 +12,8 @@ from sqlalchemy.dialects.mssql import base
 from sqlalchemy.dialects.mssql import pymssql
 from sqlalchemy.dialects.mssql import pyodbc
 from sqlalchemy.engine import url
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import assert_warnings
 from sqlalchemy.testing import engines
@@ -294,7 +297,7 @@ class ParseConnectTest(fixtures.TestBase):
         )
 
         for error in [
-            MockDBAPIError("[%s] some pyodbc message" % code)
+            MockDBAPIError(code, "[%s] some pyodbc message" % code)
             for code in [
                 "08S01",
                 "01002",
@@ -316,7 +319,9 @@ class ParseConnectTest(fixtures.TestBase):
 
         eq_(
             dialect.is_disconnect(
-                MockProgrammingError("not an error"), None, None
+                MockProgrammingError("Query with abc08007def failed"),
+                None,
+                None,
             ),
             False,
         )
@@ -511,3 +516,39 @@ class IsolationLevelDetectTest(fixtures.TestBase):
                 dialect.get_isolation_level,
                 connection,
             )
+
+
+class InvalidTransactionFalsePositiveTest(fixtures.TablesTest):
+    __only_on__ = "mssql"
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "error_t",
+            metadata,
+            Column("error_code", String(50), primary_key=True),
+        )
+
+    @classmethod
+    def insert_data(cls, connection):
+        connection.execute(
+            cls.tables.error_t.insert(), [{"error_code": "01002"}],
+        )
+
+    def test_invalid_transaction_detection(self, connection):
+        # issue #5359
+        t = self.tables.error_t
+
+        # force duplicate PK error
+        assert_raises(
+            IntegrityError,
+            connection.execute,
+            t.insert(),
+            {"error_code": "01002"},
+        )
+
+        # this should not fail with
+        # "Can't reconnect until invalid transaction is rolled back."
+        result = connection.execute(t.select()).fetchall()
+        eq_(len(result), 1)