aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/Lib/test/test_sqlite3/test_transactions.py
diff options
context:
space:
mode:
authorErlend Egeberg Aasland <erlend.aasland@innova.no>2022-01-03 20:02:39 +0100
committerGitHub <noreply@github.com>2022-01-03 19:02:39 +0000
commit9d6a239a34a66e16188d76c23a3a770515ca44ca (patch)
tree254afa367df14a95548df952452285de93cb37a3 /Lib/test/test_sqlite3/test_transactions.py
parent9d35dedc5e7e940b639230fba93c763bd9f19c09 (diff)
downloadcpython-9d6a239a34a66e16188d76c23a3a770515ca44ca.tar.gz
cpython-9d6a239a34a66e16188d76c23a3a770515ca44ca.zip
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.
Diffstat (limited to 'Lib/test/test_sqlite3/test_transactions.py')
-rw-r--r--Lib/test/test_sqlite3/test_transactions.py45
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):