Fix sqlite3 Connection initialization check

ever0de · ever0de · commit 38e9fc54e766 · 2025-10-21T01:02:50.000+09:00
Add proper __init__ validation for sqlite3.Connection to ensure base class
__init__ is called before using connection methods. This fixes the
test_connection_constructor_call_check test case.

Changes:
- Modified Connection.py_new to detect subclassing
- For base Connection class, initialization happens immediately in py_new
- For subclassed Connection, db is initialized as None
- Added __init__ method that performs actual database initialization
- Updated _db_lock error message to match CPython: 'Base Connection.__init__ not called.'

This ensures CPython compatibility where attempting to use a Connection
subclass instance without calling the base __init__ raises ProgrammingError.
diff --git a/Lib/test/test_sqlite3/test_regression.py b/Lib/test/test_sqlite3/test_regression.py
@@ -221,8 +221,6 @@ def test_str_subclass(self):
         class MyStr(str): pass
         self.con.execute("select ?", (MyStr("abc"),))
 
-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
     def test_connection_constructor_call_check(self):
         """
         Verifies that connection methods check whether base class __init__ was
diff --git a/stdlib/src/sqlite.rs b/stdlib/src/sqlite.rs
@@ -851,7 +851,31 @@ mod _sqlite {
         type Args = ConnectArgs;
 
         fn py_new(cls: PyTypeRef, args: Self::Args, vm: &VirtualMachine) -> PyResult {
-            Ok(Self::new(args, vm)?.into_ref_with_type(vm, cls)?.into())
+            let text_factory = PyStr::class(&vm.ctx).to_owned().into_object();
+
+            // For non-subclassed Connection, initialize in __new__
+            // For subclassed Connection, leave db as None and require __init__ to be called
+            let is_base_class = cls.is(Connection::class(&vm.ctx).as_object());
+
+            let db = if is_base_class {
+                // Initialize immediately for base class
+                Some(Connection::initialize_db(&args, vm)?)
+            } else {
+                // For subclasses, require __init__ to be called
+                None
+            };
+
+            let conn = Self {
+                db: PyMutex::new(db),
+                detect_types: args.detect_types,
+                isolation_level: PyAtomicRef::from(args.isolation_level),
+                check_same_thread: args.check_same_thread,
+                thread_ident: std::thread::current().id(),
+                row_factory: PyAtomicRef::from(None),
+                text_factory: PyAtomicRef::from(text_factory),
+            };
+
+            Ok(conn.into_ref_with_type(vm, cls)?.into())
         }
     }
 
@@ -873,25 +897,28 @@ mod _sqlite {
 
     #[pyclass(with(Constructor, Callable), flags(BASETYPE))]
     impl Connection {
-        fn new(args: ConnectArgs, vm: &VirtualMachine) -> PyResult<Self> {
+        fn initialize_db(args: &ConnectArgs, vm: &VirtualMachine) -> PyResult<Sqlite> {
             let path = args.database.to_cstring(vm)?;
             let db = Sqlite::from(SqliteRaw::open(path.as_ptr(), args.uri, vm)?);
             let timeout = (args.timeout * 1000.0) as c_int;
             db.busy_timeout(timeout);
             if let Some(isolation_level) = &args.isolation_level {
                 begin_statement_ptr_from_isolation_level(isolation_level, vm)?;
             }
-            let text_factory = PyStr::class(&vm.ctx).to_owned().into_object();
+            Ok(db)
+        }
 
-            Ok(Self {
-                db: PyMutex::new(Some(db)),
-                detect_types: args.detect_types,
-                isolation_level: PyAtomicRef::from(args.isolation_level),
-                check_same_thread: args.check_same_thread,
-                thread_ident: std::thread::current().id(),
-                row_factory: PyAtomicRef::from(None),
-                text_factory: PyAtomicRef::from(text_factory),
-            })
+        #[pymethod]
+        fn __init__(&self, args: ConnectArgs, vm: &VirtualMachine) -> PyResult<()> {
+            let mut guard = self.db.lock();
+            if guard.is_some() {
+                // Already initialized
+                return Ok(());
+            }
+
+            let db = Self::initialize_db(&args, vm)?;
+            *guard = Some(db);
+            Ok(())
         }
 
         fn db_lock(&self, vm: &VirtualMachine) -> PyResult<PyMappedMutexGuard<'_, Sqlite>> {
@@ -908,7 +935,7 @@ mod _sqlite {
             } else {
                 Err(new_programming_error(
                     vm,
-                    "Cannot operate on a closed database.".to_owned(),
+                    "Base Connection.__init__ not called.".to_owned(),
                 ))
             }
         }
