From cd9a5d05709b767a8f2b460e363247267633fcd0 Mon Sep 17 00:00:00 2001 From: "Erlend E. Aasland" Date: Tue, 16 Nov 2021 13:01:38 +0100 Subject: [PATCH 1/2] bpo-45512: Extend sqlite3 test suite regarding isolation levels --- Lib/test/test_sqlite3/test_dbapi.py | 27 ++++++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py index 802a6919371f9d..4104bce19ea2d7 100644 --- a/Lib/test/test_sqlite3/test_dbapi.py +++ b/Lib/test/test_sqlite3/test_dbapi.py @@ -514,14 +514,35 @@ def test_connection_init_bad_isolation_level(self): "isolation_level string must be '', 'DEFERRED', 'IMMEDIATE', or " "'EXCLUSIVE'" ) - with self.assertRaisesRegex(ValueError, msg): - memory_database(isolation_level="BOGUS") + levels = ( + "BOGUS", + " ", + "DEFERRE", + "IMMEDIAT", + "EXCLUSIV", + "DEFERREDS", + "IMMEDIATES", + "EXCLUSIVES", + ) + for level in levels: + with self.subTest(level=level): + with self.assertRaisesRegex(ValueError, msg): + memory_database(isolation_level=level) + with memory_database() as cx: + with self.assertRaisesRegex(ValueError, msg): + cx.isolation_level = level + # Check that the default level is not changed + self.assertEqual(cx.isolation_level, "") def test_connection_init_good_isolation_levels(self): for level in ("", "DEFERRED", "IMMEDIATE", "EXCLUSIVE", None): with self.subTest(level=level): with memory_database(isolation_level=level) as cx: - cx.execute("select 'ok'") + self.assertEqual(cx.isolation_level, level) + with memory_database() as cx: + self.assertEqual(cx.isolation_level, "") + cx.isolation_level = level + self.assertEqual(cx.isolation_level, level) class UninitialisedConnectionTests(unittest.TestCase): From 54ad37722653b0d00090e00d8055882732c01302 Mon Sep 17 00:00:00 2001 From: "Erlend E. Aasland" Date: Tue, 16 Nov 2021 14:20:42 +0100 Subject: [PATCH 2/2] Also check that isolation level is correctly mapped to begin statements --- Lib/test/test_sqlite3/test_transactions.py | 100 +++++++++++++++++++++ 1 file changed, 100 insertions(+) diff --git a/Lib/test/test_sqlite3/test_transactions.py b/Lib/test/test_sqlite3/test_transactions.py index 8ea352b9ef0f9b..3efa2c1e604ff7 100644 --- a/Lib/test/test_sqlite3/test_transactions.py +++ b/Lib/test/test_sqlite3/test_transactions.py @@ -23,6 +23,8 @@ import os, unittest import sqlite3 as sqlite +from .test_dbapi import memory_database + def get_db_path(): return "sqlite_testdb" @@ -143,6 +145,7 @@ def test_rollback_cursor_consistency(self): with self.assertRaises(sqlite.InterfaceError): cur.fetchall() + class SpecialCommandTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") @@ -162,6 +165,7 @@ def tearDown(self): self.cur.close() self.con.close() + class TransactionalDDL(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") @@ -196,5 +200,101 @@ def tearDown(self): self.con.close() +class IsolationLevelFromInit(unittest.TestCase): + CREATE = "create table t(t)" + INSERT = "insert into t values(1)" + + def setUp(self): + self.traced = [] + + def _run_test(self, cx): + cx.execute(self.CREATE) + cx.set_trace_callback(lambda stmt: self.traced.append(stmt)) + with cx: + cx.execute(self.INSERT) + + def test_isolation_level_default(self): + with memory_database() as cx: + self._run_test(cx) + self.assertEqual(self.traced, ["BEGIN ", self.INSERT, "COMMIT"]) + + def test_isolation_level_begin(self): + with memory_database(isolation_level="") as cx: + self._run_test(cx) + self.assertEqual(self.traced, ["BEGIN ", self.INSERT, "COMMIT"]) + + def test_isolation_level_deferred(self): + with memory_database(isolation_level="DEFERRED") as cx: + self._run_test(cx) + self.assertEqual(self.traced, ["BEGIN DEFERRED", self.INSERT, "COMMIT"]) + + def test_isolation_level_immediate(self): + with memory_database(isolation_level="IMMEDIATE") as cx: + self._run_test(cx) + self.assertEqual(self.traced, + ["BEGIN IMMEDIATE", self.INSERT, "COMMIT"]) + + def test_isolation_level_exclusive(self): + with memory_database(isolation_level="EXCLUSIVE") as cx: + self._run_test(cx) + self.assertEqual(self.traced, + ["BEGIN EXCLUSIVE", self.INSERT, "COMMIT"]) + + def test_isolation_level_none(self): + with memory_database(isolation_level=None) as cx: + self._run_test(cx) + self.assertEqual(self.traced, [self.INSERT]) + + +class IsolationLevelPostInit(unittest.TestCase): + QUERY = "insert into t values(1)" + + def setUp(self): + self.cx = sqlite.connect(":memory:") + self.cx.execute("create table t(t)") + self.traced = [] + self.cx.set_trace_callback(lambda stmt: self.traced.append(stmt)) + + def tearDown(self): + self.cx.close() + + def test_isolation_level_default(self): + with self.cx: + self.cx.execute(self.QUERY) + self.assertEqual(self.traced, ["BEGIN ", self.QUERY, "COMMIT"]) + + def test_isolation_level_begin(self): + self.cx.isolation_level = "" + with self.cx: + self.cx.execute(self.QUERY) + self.assertEqual(self.traced, ["BEGIN ", self.QUERY, "COMMIT"]) + + def test_isolation_level_deferrred(self): + self.cx.isolation_level = "DEFERRED" + with self.cx: + self.cx.execute(self.QUERY) + self.assertEqual(self.traced, ["BEGIN DEFERRED", self.QUERY, "COMMIT"]) + + def test_isolation_level_immediate(self): + self.cx.isolation_level = "IMMEDIATE" + with self.cx: + self.cx.execute(self.QUERY) + self.assertEqual(self.traced, + ["BEGIN IMMEDIATE", self.QUERY, "COMMIT"]) + + def test_isolation_level_exclusive(self): + self.cx.isolation_level = "EXCLUSIVE" + with self.cx: + self.cx.execute(self.QUERY) + self.assertEqual(self.traced, + ["BEGIN EXCLUSIVE", self.QUERY, "COMMIT"]) + + def test_isolation_level_none(self): + self.cx.isolation_level = None + with self.cx: + self.cx.execute(self.QUERY) + self.assertEqual(self.traced, [self.QUERY]) + + if __name__ == "__main__": unittest.main()