bpo-45512: Raise exception if sqlite3.Connection.__init__ is called with bad isolation level (#29561)

* bpo-45512: Raise sqlite3.Connection.__init__ is called with bad isolation level

* Also explicitly test allowed isolation levels

* Use subTest for better error messages if something goes wrong

* Update Lib/test/test_sqlite3/test_dbapi.py

Co-authored-by: Dong-hee Na <donghee.na92@gmail.com>

Co-authored-by: Dong-hee Na <donghee.na92@gmail.com>

2 files changed, 19 insertions, 5 deletions

diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py
index a5ec66f..802a691 100644
--- a/Lib/test/test_sqlite3/test_dbapi.py
+++ b/Lib/test/test_sqlite3/test_dbapi.py
@@ -48,8 +48,8 @@ def managed_connect(*args, in_mem=False, **kwargs):
 
 
 # Helper for temporary memory databases
-def memory_database():
-    cx = sqlite.connect(":memory:")
+def memory_database(*args, **kwargs):
+    cx = sqlite.connect(":memory:", *args, **kwargs)
     return contextlib.closing(cx)
 
 
@@ -509,6 +509,20 @@ class ConnectionTests(unittest.TestCase):
         self.assertRaisesRegex(sqlite.ProgrammingError, msg,
                                self.cx.setlimit, cat, 0)
 
+    def test_connection_init_bad_isolation_level(self):
+        msg = (
+            "isolation_level string must be '', 'DEFERRED', 'IMMEDIATE', or "
+            "'EXCLUSIVE'"
+        )
+        with self.assertRaisesRegex(ValueError, msg):
+            memory_database(isolation_level="BOGUS")
+
+    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'")
+
 
 class UninitialisedConnectionTests(unittest.TestCase):
     def setUp(self):
diff --git a/Modules/_sqlite/connection.c b/Modules/_sqlite/connection.c
index e4f0013..b902dc8 100644
--- a/Modules/_sqlite/connection.c
+++ b/Modules/_sqlite/connection.c
@@ -127,6 +127,9 @@ get_begin_statement(const char *level)
             return begin_statements[i];
         }
     }
+    PyErr_SetString(PyExc_ValueError,
+                    "isolation_level string must be '', 'DEFERRED', "
+                    "'IMMEDIATE', or 'EXCLUSIVE'");
     return NULL;
 }
 
@@ -1389,9 +1392,6 @@ pysqlite_connection_set_isolation_level(pysqlite_Connection* self, PyObject* iso
         }
         const char *stmt = get_begin_statement(cstr_level);
         if (stmt == NULL) {
-            PyErr_SetString(PyExc_ValueError,
-                            "isolation_level string must be '', 'DEFERRED', "
-                            "'IMMEDIATE', or 'EXCLUSIVE'");
             return -1;
         }
         self->begin_statement = stmt;