bpo-45512: Extend sqlite3 test suite regarding isolation levels (GH-29576)

Lib/test/test_sqlite3/test_dbapi.py

@@ -514,14 +514,35 @@ class ConnectionTests(unittest.TestCase):
             "isolation_level string must be '', 'DEFERRED', 'IMMEDIATE', or "
             "'EXCLUSIVE'"
         )
-        with self.assertRaisesRegex(ValueError, msg):
+        levels = (
-            memory_database(isolation_level="BOGUS")
+            "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)
 
     def test_connection_reinit(self):
         db = ":memory:"

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 @@ class TransactionTests(unittest.TestCase):
         with self.assertRaises(sqlite.InterfaceError):
             cur.fetchall()
 
+
 class SpecialCommandTests(unittest.TestCase):
     def setUp(self):
         self.con = sqlite.connect(":memory:")
@@ -162,6 +165,7 @@ class SpecialCommandTests(unittest.TestCase):
         self.cur.close()
         self.con.close()
 
+
 class TransactionalDDL(unittest.TestCase):
     def setUp(self):
         self.con = sqlite.connect(":memory:")
@@ -196,5 +200,101 @@ class TransactionalDDL(unittest.TestCase):
         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()