bpo-45126: Harden `sqlite3` connection initialisation (GH-28227)

1 files changed, 38 insertions, 0 deletions

diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py
index 802a691..18359e1 100644
--- a/Lib/test/test_sqlite3/test_dbapi.py
+++ b/Lib/test/test_sqlite3/test_dbapi.py
@@ -523,6 +523,44 @@ class ConnectionTests(unittest.TestCase):
                 with memory_database(isolation_level=level) as cx:
                     cx.execute("select 'ok'")
 
+    def test_connection_reinit(self):
+        db = ":memory:"
+        cx = sqlite.connect(db)
+        cx.text_factory = bytes
+        cx.row_factory = sqlite.Row
+        cu = cx.cursor()
+        cu.execute("create table foo (bar)")
+        cu.executemany("insert into foo (bar) values (?)",
+                       ((str(v),) for v in range(4)))
+        cu.execute("select bar from foo")
+
+        rows = [r for r in cu.fetchmany(2)]
+        self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))
+        self.assertEqual([r[0] for r in rows], [b"0", b"1"])
+
+        cx.__init__(db)
+        cx.execute("create table foo (bar)")
+        cx.executemany("insert into foo (bar) values (?)",
+                       ((v,) for v in ("a", "b", "c", "d")))
+
+        # This uses the old database, old row factory, but new text factory
+        rows = [r for r in cu.fetchall()]
+        self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))
+        self.assertEqual([r[0] for r in rows], ["2", "3"])
+
+    def test_connection_bad_reinit(self):
+        cx = sqlite.connect(":memory:")
+        with cx:
+            cx.execute("create table t(t)")
+        with temp_dir() as db:
+            self.assertRaisesRegex(sqlite.OperationalError,
+                                   "unable to open database file",
+                                   cx.__init__, db)
+            self.assertRaisesRegex(sqlite.ProgrammingError,
+                                   "Base Connection.__init__ not called",
+                                   cx.executemany, "insert into t values(?)",
+                                   ((v,) for v in range(3)))
+
 
 class UninitialisedConnectionTests(unittest.TestCase):
     def setUp(self):