diff options
Diffstat (limited to 'Lib/sqlite3/test/regression.py')
| -rw-r--r-- | Lib/sqlite3/test/regression.py | 101 |
1 files changed, 66 insertions, 35 deletions
diff --git a/Lib/sqlite3/test/regression.py b/Lib/sqlite3/test/regression.py index eaaaa2c..7dd0050 100644 --- a/Lib/sqlite3/test/regression.py +++ b/Lib/sqlite3/test/regression.py @@ -73,7 +73,7 @@ class RegressionTests(unittest.TestCase): def CheckStatementFinalizationOnCloseDb(self): # pysqlite versions <= 2.3.3 only finalized statements in the statement # cache when closing the database. statements that were still - # referenced in cursors weren't closed an could provoke " + # referenced in cursors weren't closed and could provoke " # "OperationalError: Unable to close due to unfinalised statements". con = sqlite.connect(":memory:") cursors = [] @@ -84,9 +84,8 @@ class RegressionTests(unittest.TestCase): cur.execute("select 1 x union select " + str(i)) con.close() + @unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 2), 'needs sqlite 3.2.2 or newer') def CheckOnConflictRollback(self): - if sqlite.sqlite_version_info < (3, 2, 2): - return con = sqlite.connect(":memory:") con.execute("create table foo(x, unique(x) on conflict rollback)") con.execute("insert into foo(x) values (1)") @@ -134,17 +133,11 @@ class RegressionTests(unittest.TestCase): def CheckErrorMsgDecodeError(self): # When porting the module to Python 3.0, the error message about # decoding errors disappeared. This verifies they're back again. - failure = None - try: + with self.assertRaises(sqlite.OperationalError) as cm: self.con.execute("select 'xxx' || ? || 'yyy' colname", (bytes(bytearray([250])),)).fetchone() - failure = "should have raised an OperationalError with detailed description" - except sqlite.OperationalError as e: - msg = e.args[0] - if not msg.startswith("Could not decode to UTF-8 column 'colname' with text 'xxx"): - failure = "OperationalError did not have expected description text" - if failure: - self.fail(failure) + msg = "Could not decode to UTF-8 column 'colname' with text 'xxx" + self.assertIn(msg, str(cm.exception)) def CheckRegisterAdapter(self): """ @@ -153,11 +146,34 @@ class RegressionTests(unittest.TestCase): self.assertRaises(TypeError, sqlite.register_adapter, {}, None) def CheckSetIsolationLevel(self): - """ - See issue 3312. - """ + # See issue 27881. + class CustomStr(str): + def upper(self): + return None + def __del__(self): + con.isolation_level = "" + con = sqlite.connect(":memory:") - setattr(con, "isolation_level", "\xe9") + con.isolation_level = None + for level in "", "DEFERRED", "IMMEDIATE", "EXCLUSIVE": + with self.subTest(level=level): + con.isolation_level = level + con.isolation_level = level.lower() + con.isolation_level = level.capitalize() + con.isolation_level = CustomStr(level) + + # setting isolation_level failure should not alter previous state + con.isolation_level = None + con.isolation_level = "DEFERRED" + pairs = [ + (1, TypeError), (b'', TypeError), ("abc", ValueError), + ("IMMEDIATE\0EXCLUSIVE", ValueError), ("\xe9", ValueError), + ] + for value, exc in pairs: + with self.subTest(level=value): + with self.assertRaises(exc): + con.isolation_level = value + self.assertEqual(con.isolation_level, "DEFERRED") def CheckCursorConstructorCallCheck(self): """ @@ -170,14 +186,8 @@ class RegressionTests(unittest.TestCase): con = sqlite.connect(":memory:") cur = Cursor(con) - try: + with self.assertRaises(sqlite.ProgrammingError): cur.execute("select 4+5").fetchall() - self.fail("should have raised ProgrammingError") - except sqlite.ProgrammingError: - pass - except: - self.fail("should have raised ProgrammingError") - def CheckStrSubclass(self): """ @@ -196,13 +206,8 @@ class RegressionTests(unittest.TestCase): pass con = Connection(":memory:") - try: + with self.assertRaises(sqlite.ProgrammingError): cur = con.cursor() - self.fail("should have raised ProgrammingError") - except sqlite.ProgrammingError: - pass - except: - self.fail("should have raised ProgrammingError") def CheckCursorRegistration(self): """ @@ -223,13 +228,8 @@ class RegressionTests(unittest.TestCase): cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)]) cur.execute("select x from foo") con.rollback() - try: + with self.assertRaises(sqlite.InterfaceError): cur.fetchall() - self.fail("should have raised InterfaceError") - except sqlite.InterfaceError: - pass - except: - self.fail("should have raised InterfaceError") def CheckAutoCommit(self): """ @@ -345,6 +345,37 @@ class RegressionTests(unittest.TestCase): self.assertRaises(ValueError, cur.execute, " \0select 2") self.assertRaises(ValueError, cur.execute, "select 2\0") + def CheckCommitCursorReset(self): + """ + Connection.commit() did reset cursors, which made sqlite3 + to return rows multiple times when fetched from cursors + after commit. See issues 10513 and 23129 for details. + """ + con = sqlite.connect(":memory:") + con.executescript(""" + create table t(c); + create table t2(c); + insert into t values(0); + insert into t values(1); + insert into t values(2); + """) + + self.assertEqual(con.isolation_level, "") + + counter = 0 + for i, row in enumerate(con.execute("select c from t")): + with self.subTest(i=i, row=row): + con.execute("insert into t2(c) values (?)", (i,)) + con.commit() + if counter == 0: + self.assertEqual(row[0], 0) + elif counter == 1: + self.assertEqual(row[0], 1) + elif counter == 2: + self.assertEqual(row[0], 2) + counter += 1 + self.assertEqual(counter, 3, "should have returned exactly three rows") + def suite(): regression_suite = unittest.makeSuite(RegressionTests, "Check") |