1 files changed, 100 insertions, 0 deletions
diff --git a/Lib/test/test_sqlite3/test_transactions.py b/Lib/test/test_sqlite3/test_transactions.py
index 8ea352b9ef0..3efa2c1e604 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 @@ 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()