feat(ldapobject): Add wrapper for ldap_connect

mistotebe · mistotebe · commit 6e471e3b828d · 2026-05-07T11:14:46.000+01:00
diff --git a/Lib/ldap/ldapobject.py b/Lib/ldap/ldapobject.py
@@ -171,6 +171,13 @@ def fileno(self):
     """
     return self.get_option(ldap.OPT_DESC)
 
+  def connect(self):
+    """
+    connect() -> None
+        Establishes LDAP connection if needed.
+    """
+    return self._ldap_call(self._l.connect)
+
   def abandon_ext(self,msgid,serverctrls=None,clientctrls=None):
     """
     abandon_ext(msgid[,serverctrls=None[,clientctrls=None]]) -> None
diff --git a/Modules/LDAPObject.c b/Modules/LDAPObject.c
@@ -1475,6 +1475,38 @@ l_ldap_extended_operation(LDAPObject *self, PyObject *args)
     return PyLong_FromLong(msgid);
 }
 
+/* ldap_connect */
+
+static PyObject *
+l_ldap_connect(LDAPObject *self, PyObject Py_UNUSED(args))
+{
+#if LDAP_VENDOR_VERSION >= 20500
+    int ldaperror;
+
+    if (ldap_version_info.ldapai_vendor_version < 20500)
+#endif
+    {
+        PyErr_SetString(PyExc_NotImplementedError,
+            "loaded libldap doesn't support this feature");
+        return NULL;
+    }
+
+#if LDAP_VENDOR_VERSION >= 20500
+    if (not_valid(self))
+        return NULL;
+
+    LDAP_BEGIN_ALLOW_THREADS(self);
+    ldaperror = ldap_connect(self->ldap);
+    LDAP_END_ALLOW_THREADS(self);
+
+    if ( ldaperror != LDAP_SUCCESS )
+        return LDAPerror(self->ldap);
+
+    Py_INCREF(Py_None);
+    return Py_None;
+#endif
+}
+
 /* methods */
 
 static PyMethodDef methods[] = {
@@ -1504,6 +1536,7 @@ static PyMethodDef methods[] = {
     {"cancel", (PyCFunction)l_ldap_cancel, METH_VARARGS},
 #endif
     {"extop", (PyCFunction)l_ldap_extended_operation, METH_VARARGS},
+    {"connect", (PyCFunction)l_ldap_connect, METH_NOARGS},
     {NULL, NULL}
 };
 
diff --git a/Tests/t_cext.py b/Tests/t_cext.py
@@ -280,6 +280,29 @@ def test_simple_anonymous_bind(self):
         self.assertEqual(pmsg, [])
         self.assertEqual(ctrls, [])
 
+    @unittest.skipUnless(
+        _ldap.VENDOR_VERSION >= 20500 and \
+            _ldap._VENDOR_VERSION_RUNTIME >= 20500,
+        reason="Test requires libldap 2.5+"
+    )
+    def test_connect(self):
+        l = self._open_conn(bind=False)
+        invalid_fileno = l.get_option(_ldap.OPT_DESC)
+        l.connect()
+        fileno = l.get_option(_ldap.OPT_DESC)
+        self.assertNotEqual(invalid_fileno, fileno)
+
+        self._bind_conn(l)
+
+    @unittest.skipUnless(
+        _ldap._VENDOR_VERSION_RUNTIME < 20500,
+        reason="Test requires linking to libldap < 2.5"
+    )
+    def test_connect_notimpl(self):
+        l = self._open_conn(bind=False)
+        with self.assertRaises(NotImplementedError):
+            l.connect()
+
     def test_anon_rootdse_search(self):
         l = self._open_conn(bind=False)
         # see if we can get the rootdse with anon search (without prior bind)
