[Python-checkins] bpo-45512: Extend `sqlite3` test suite regarding isolation levels (GH-29576)
corona10
webhook-mailer at python.org
Wed Nov 17 05:02:00 EST 2021
https://github.com/python/cpython/commit/5f9247e36a0213b0dcfd43533db5cf6570895cfd
commit: 5f9247e36a0213b0dcfd43533db5cf6570895cfd
branch: main
author: Erlend Egeberg Aasland <erlend.aasland at innova.no>
committer: corona10 <donghee.na92 at gmail.com>
date: 2021-11-17T19:01:54+09:00
summary:
bpo-45512: Extend `sqlite3` test suite regarding isolation levels (GH-29576)
files:
M Lib/test/test_sqlite3/test_dbapi.py
M Lib/test/test_sqlite3/test_transactions.py
diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py
index 18359e1a5e2ab..4eb4e180bf117 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)
def test_connection_reinit(self):
db = ":memory:"
diff --git a/Lib/test/test_sqlite3/test_transactions.py b/Lib/test/test_sqlite3/test_transactions.py
index 8ea352b9ef0f9..3efa2c1e604ff 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()
More information about the Python-checkins
mailing list