bpo-44092: Don't reset statements/cursors before rollback (GH-26026)

In SQLite versions pre 3.7.11, pending statements would block a rollback.  This is no longer the case, so remove the workaround.

1 files changed, 39 insertions, 6 deletions

diff --git a/Lib/test/test_sqlite3/test_transactions.py b/Lib/test/test_sqlite3/test_transactions.py
index 3efa2c1e604..55cf8f1fdfc 100644
--- a/Lib/test/test_sqlite3/test_transactions.py
+++ b/Lib/test/test_sqlite3/test_transactions.py
@@ -131,10 +131,7 @@ class TransactionTests(unittest.TestCase):
         self.con1.commit()
 
     def test_rollback_cursor_consistency(self):
-        """
-        Checks if cursors on the connection are set into a "reset" state
-        when a rollback is done on the connection.
-        """
+        """Check that cursors behave correctly after rollback."""
         con = sqlite.connect(":memory:")
         cur = con.cursor()
         cur.execute("create table test(x)")
@@ -142,8 +139,44 @@ class TransactionTests(unittest.TestCase):
         cur.execute("select 1 union select 2 union select 3")
 
         con.rollback()
-        with self.assertRaises(sqlite.InterfaceError):
-            cur.fetchall()
+        self.assertEqual(cur.fetchall(), [(1,), (2,), (3,)])
+
+
+class RollbackTests(unittest.TestCase):
+    """bpo-44092: sqlite3 now leaves it to SQLite to resolve rollback issues"""
+
+    def setUp(self):
+        self.con = sqlite.connect(":memory:")
+        self.cur1 = self.con.cursor()
+        self.cur2 = self.con.cursor()
+        with self.con:
+            self.con.execute("create table t(c)");
+            self.con.executemany("insert into t values(?)", [(0,), (1,), (2,)])
+        self.cur1.execute("begin transaction")
+        select = "select c from t"
+        self.cur1.execute(select)
+        self.con.rollback()
+        self.res = self.cur2.execute(select)  # Reusing stmt from cache
+
+    def tearDown(self):
+        self.con.close()
+
+    def _check_rows(self):
+        for i, row in enumerate(self.res):
+            self.assertEqual(row[0], i)
+
+    def test_no_duplicate_rows_after_rollback_del_cursor(self):
+        del self.cur1
+        self._check_rows()
+
+    def test_no_duplicate_rows_after_rollback_close_cursor(self):
+        self.cur1.close()
+        self._check_rows()
+
+    def test_no_duplicate_rows_after_rollback_new_query(self):
+        self.cur1.execute("select c from t where c = 1")
+        self._check_rows()
+
 
 
 class SpecialCommandTests(unittest.TestCase):