bpo-43553: Improve `sqlite3` test coverage (GH-26886)

1 files changed, 24 insertions, 2 deletions

diff --git a/Lib/sqlite3/test/dbapi.py b/Lib/sqlite3/test/dbapi.py
index 7e44cac76f5..1a4b44188bd 100644
--- a/Lib/sqlite3/test/dbapi.py
+++ b/Lib/sqlite3/test/dbapi.py
@@ -26,9 +26,8 @@ import sys
 import threading
 import unittest
 
-from test.support import check_disallow_instantiation
+from test.support import check_disallow_instantiation, threading_helper
 from test.support.os_helper import TESTFN, unlink
-from test.support import threading_helper
 
 
 # Helper for tests using TESTFN
@@ -110,6 +109,10 @@ class ModuleTests(unittest.TestCase):
         cx = sqlite.connect(":memory:")
         check_disallow_instantiation(self, type(cx("select 1")))
 
+    def test_complete_statement(self):
+        self.assertFalse(sqlite.complete_statement("select t"))
+        self.assertTrue(sqlite.complete_statement("create table t(t);"))
+
 
 class ConnectionTests(unittest.TestCase):
 
@@ -225,6 +228,20 @@ class ConnectionTests(unittest.TestCase):
                 self.assertTrue(hasattr(self.cx, exc))
                 self.assertIs(getattr(sqlite, exc), getattr(self.cx, exc))
 
+    def test_interrupt_on_closed_db(self):
+        cx = sqlite.connect(":memory:")
+        cx.close()
+        with self.assertRaises(sqlite.ProgrammingError):
+            cx.interrupt()
+
+    def test_interrupt(self):
+        self.assertIsNone(self.cx.interrupt())
+
+    def test_drop_unused_refs(self):
+        for n in range(500):
+            cu = self.cx.execute(f"select {n}")
+            self.assertEqual(cu.fetchone()[0], n)
+
 
 class OpenTests(unittest.TestCase):
     _sql = "create table test(id integer)"
@@ -594,6 +611,11 @@ class CursorTests(unittest.TestCase):
         new_count = len(res.description)
         self.assertEqual(new_count - old_count, 1)
 
+    def test_same_query_in_multiple_cursors(self):
+        cursors = [self.cx.execute("select 1") for _ in range(3)]
+        for cu in cursors:
+            self.assertEqual(cu.fetchall(), [(1,)])
+
 
 class ThreadTests(unittest.TestCase):
     def setUp(self):