mirror of
https://github.com/python/cpython.git
synced 2025-07-24 11:44:31 +00:00
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.
This commit is contained in:
parent
9d35dedc5e
commit
9d6a239a34
5 changed files with 47 additions and 52 deletions
|
@ -231,28 +231,6 @@ class RegressionTests(unittest.TestCase):
|
|||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
cur = con.cursor()
|
||||
|
||||
def test_cursor_registration(self):
|
||||
"""
|
||||
Verifies that subclassed cursor classes are correctly registered with
|
||||
the connection object, too. (fetch-across-rollback problem)
|
||||
"""
|
||||
class Connection(sqlite.Connection):
|
||||
def cursor(self):
|
||||
return Cursor(self)
|
||||
|
||||
class Cursor(sqlite.Cursor):
|
||||
def __init__(self, con):
|
||||
sqlite.Cursor.__init__(self, con)
|
||||
|
||||
con = Connection(":memory:")
|
||||
cur = con.cursor()
|
||||
cur.execute("create table foo(x)")
|
||||
cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)])
|
||||
cur.execute("select x from foo")
|
||||
con.rollback()
|
||||
with self.assertRaises(sqlite.InterfaceError):
|
||||
cur.fetchall()
|
||||
|
||||
def test_auto_commit(self):
|
||||
"""
|
||||
Verifies that creating a connection in autocommit mode works.
|
||||
|
|
|
@ -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):
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue