diff options
author | Erlend Egeberg Aasland <erlend.aasland@innova.no> | 2022-01-03 20:02:39 +0100 |
---|---|---|
committer | GitHub <noreply@github.com> | 2022-01-03 19:02:39 +0000 |
commit | 9d6a239a34a66e16188d76c23a3a770515ca44ca (patch) | |
tree | 254afa367df14a95548df952452285de93cb37a3 /Lib/test/test_sqlite3/test_transactions.py | |
parent | 9d35dedc5e7e940b639230fba93c763bd9f19c09 (diff) | |
download | cpython-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.py | 45 |
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): |