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

[corona10]

https://github.com/python/cpython/commit/822c3dcce3996e411c1ff5c432c6ac7d2845cfd6
commit: 822c3dcce3996e411c1ff5c432c6ac7d2845cfd6
branch: main
author: Erlend Egeberg Aasland <erlend.aasland at innova.no>
committer: corona10 <donghee.na92 at gmail.com>
date: 2021-11-15T21:55:38+09:00
summary:

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 at gmail.com>

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

files:
M Lib/test/test_sqlite3/test_dbapi.py
M Modules/_sqlite/connection.c

diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py
index a5ec66fe22a2a7..802a6919371f9d 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 @@ def test_connection_bad_limit_category(self):
         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 e4f0013ae604e4..b902dc845c6188 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;