Added ConnectParams.parse_dsn_with_credentials() and associated tests to

ensure that the ability to use a dsn containing credentials works as
expected (and corrected bug that was preventing this from happening).

Author: anthony-tuininga

doc/src/api_manual/connect_param.rst

@@ -32,6 +32,15 @@ ConnectParams Methods
   pairs, or a simple alias which is looked up in ``tnsnames.ora``. Parameters
   that are found in the connect string override any currently stored values.
 
+.. method:: ConnectParams.parse_dsn_with_credentials(dsn)
+
+  Parses a dsn in the form <user>/<password>@<connect_string> or in the form
+  <user>/<password> and returns a 3-tuple containing the parsed user, password
+  and connect string. Empty strings are returned as the value ``None``. This is
+  done automatically when a value is passed to the ``dsn`` parameter but no
+  value is passed to the ``user`` password when creating a standalone
+  connection or connection pool.
+
 .. method:: ConnectParams.set(user=None, proxy_user=None, password=None, \
     newpassword=None, wallet_password=None, access_token=None, host=None, \
     port=None, protocol=None, https_proxy=None, https_proxy_port=None, service_name=None, \

doc/src/release_notes.rst

@@ -61,11 +61,15 @@ Thick Mode Changes
     for a type containing an attribute or element with an unsupported data type
     until the first attempt to reference the attribute or element with the
     unsupported data type.
+#)  Fixed bug when attempting to create bequeath connections using the DSN
+    parameter with credentials.
 
 Common Changes
 ++++++++++++++
 
 #)  Improved type annotations.
+#)  Added method for parsing a DSN with credentials by calling
+    :meth:`ConnectParams.parse_dsn_with_credentials()`.
 #)  Error ``DPY-2038: element at index {index} does not exist`` is now raised
     whenever an element in a database collection is missing. Previously, thick
     mode raised ``DPI-1024: element at index {index} does not exist`` and thin

src/oracledb/connect_params.py

@@ -1,5 +1,5 @@
 #------------------------------------------------------------------------------
-# Copyright (c) 2021, 2022, Oracle and/or its affiliates.
+# Copyright (c) 2021, 2023, Oracle and/or its affiliates.
 #
 # This software is dual-licensed to you under the Universal Permissive License
 # (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
@@ -599,6 +599,17 @@ def parse_connect_string(self, connect_string: str) -> None:
         """
         self._impl.parse_connect_string(connect_string)
 
+    def parse_dsn_with_credentials(self, dsn: str) -> tuple:
+        """
+        Parses a dsn in the form <user>/<password>@<connect_string> or in the
+        form <user>/<password> and returns a 3-tuple containing the parsed
+        user, password and connect string. Empty strings are returned as the
+        value None. This is done automatically when a value is passed to
+        the dsn parameter but no value is passed to the user password when
+        creating a standalone connection or connection pool.
+        """
+        return self._impl.parse_dsn_with_credentials(dsn)
+
     @utils.params_setter
     def set(self, *,
             user: str=None,

src/oracledb/connection.py

@@ -1,5 +1,5 @@
 #------------------------------------------------------------------------------
-# Copyright (c) 2020, 2022, Oracle and/or its affiliates.
+# Copyright (c) 2020, 2023, Oracle and/or its affiliates.
 #
 # This software is dual-licensed to you under the Universal Permissive License
 # (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
@@ -109,12 +109,7 @@ def __init__(self,
                 errors._raise_err(errors.ERR_INVALID_CONNECT_PARAMS)
             else:
                 params_impl = params._impl.copy()
-            if kwargs:
-                params_impl.set(kwargs)
-            if dsn is not None:
-                dsn = params_impl.parse_dsn(dsn, thin)
-            if dsn is None:
-                dsn = params_impl.get_connect_string()
+            dsn = params_impl.process_args(dsn, kwargs, thin)
 
             # see if connection is being acquired from a pool
             if pool is None:

src/oracledb/impl/base/connect_params.pyx

@@ -527,10 +527,7 @@ cdef class ConnectParamsImpl:
 
     def get_connect_string(self):
         """
-        Internal method for getting the connect string. This will either be the
-        connect string supplied to parse_connect_string() or parse_dsn(), or it
-        will be a connect string built up from the components supplied when the
-        object was built.
+        Returns a connect string generated from the parameters.
         """
         return self.description_list.build_connect_string()
 
@@ -557,28 +554,28 @@ cdef class ConnectParamsImpl:
             errors._raise_err(errors.ERR_CANNOT_PARSE_CONNECT_STRING, cause=e,
                               data=connect_string)
 
-    def parse_dsn(self, str dsn, bint parse_connect_string):
+    def parse_dsn_with_credentials(self, str dsn):
         """
         Parse a dsn (data source name) string supplied by the user. This can be
-        in the form user/password@connect_string or it can be a simple connect
-        string. The connect string is returned and the user, proxy_user and
-        password values are retained.
+        in the form user/password@connect_string or it can be in the form
+        user/password. The user, password and connect string are returned in a
+        3-tuple.
         """
-        connect_string = dsn
         pos = dsn.rfind("@")
         if pos >= 0:
             credentials = dsn[:pos]
-            connect_string = dsn[pos + 1:]
-            pos = credentials.find("/")
-            if pos >= 0:
-                user = credentials[:pos]
-                self._set_password(credentials[pos + 1:])
-            else:
-                user = credentials
-            self.parse_user(user)
-        if parse_connect_string:
-            self.parse_connect_string(connect_string)
-        return connect_string
+            connect_string = dsn[pos + 1:] or None
+        else:
+            credentials = dsn
+            connect_string = None
+        pos = credentials.find("/")
+        if pos >= 0:
+            user = credentials[:pos] or None
+            password = credentials[pos + 1:] or None
+        else:
+            user = credentials or None
+            password = None
+        return (user, password, connect_string)
 
     def parse_user(self, str user):
         """
@@ -593,6 +590,30 @@ cdef class ConnectParamsImpl:
         else:
             self.user = user
 
+    def process_args(self, str dsn, dict kwargs, bint thin):
+        """
+        Processes the arguments to connect() and create_pool().
+
+            - the keyword arguments are set
+            - if no user was specified in the keyword arguments and a dsn is
+              specified, it is parsed to determine the user, password and
+              connect string and the user and password are stored
+            - in thin mode, the connect string is then parsed into its
+              components and stored
+            - if no dsn was specified, one is built from the components
+            - the connect string is returned
+        """
+        if kwargs:
+            self.set(kwargs)
+        if self.user is None and dsn is not None:
+            user, password, dsn = self.parse_dsn_with_credentials(dsn)
+            self.set(dict(user=user, password=password))
+        if dsn is not None and thin:
+            self.parse_connect_string(dsn)
+        if dsn is None:
+            dsn = self.get_connect_string()
+        return dsn
+
 
 cdef class Address:
     """

src/oracledb/pool.py

@@ -1,5 +1,5 @@
 #------------------------------------------------------------------------------
-# Copyright (c) 2020, 2022, Oracle and/or its affiliates.
+# Copyright (c) 2020, 2023, Oracle and/or its affiliates.
 #
 # This software is dual-licensed to you under the Universal Permissive License
 # (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
@@ -84,16 +84,11 @@ def __init__(self, dsn: str=None, *,
             errors._raise_err(errors.ERR_INVALID_POOL_PARAMS)
         else:
             params_impl = params._impl.copy()
-        if kwargs:
-            params_impl.set(kwargs)
-        self._connection_type = \
-                params_impl.connectiontype or connection_module.Connection
         with driver_mode.get_manager() as mode_mgr:
             thin = mode_mgr.thin
-            if dsn is not None:
-                dsn = params_impl.parse_dsn(dsn, thin)
-            if dsn is None:
-                dsn = params_impl.get_connect_string()
+            dsn = params_impl.process_args(dsn, kwargs, thin)
+            self._connection_type = \
+                    params_impl.connectiontype or connection_module.Connection
             if thin:
                 impl = thin_impl.ThinPoolImpl(dsn, params_impl)
             else:

tests/test_1100_connection.py

@@ -54,12 +54,6 @@ def __verify_fetched_data(self, connection):
                 fetched_data = [s for s, in cursor.execute(sql)]
                 self.assertEqual(fetched_data, expected_data)
 
-    def __verify_args(self, connection):
-        self.assertEqual(connection.username, test_env.get_main_user(),
-                         "user name differs")
-        self.assertEqual(connection.dsn, test_env.get_connect_string(),
-                         "dsn differs")
-
     def __verify_attributes(self, connection, attrName, value, sql):
         setattr(connection, attrName, value)
         cursor = connection.cursor()
@@ -70,7 +64,10 @@ def __verify_attributes(self, connection, attrName, value, sql):
     def test_1100_simple_connection(self):
         "1100 - simple connection to database"
         connection = test_env.get_connection()
-        self.__verify_args(connection)
+        self.assertEqual(connection.username, test_env.get_main_user(),
+                         "user name differs")
+        self.assertEqual(connection.dsn, test_env.get_connect_string(),
+                         "dsn differs")
 
     @unittest.skipIf(test_env.get_is_thin(),
                      "thin mode doesn't support application context yet")
@@ -226,14 +223,6 @@ def test_1113_connect_with_handle(self):
                                connection2.close)
         connection.close()
 
-    def test_1115_single_arg(self):
-        "1115 - connection to database with user, password, DSN together"
-        arg = "%s/%s@%s" % (test_env.get_main_user(),
-                            test_env.get_main_password(),
-                            test_env.get_connect_string())
-        connection = test_env.get_connection(arg)
-        self.__verify_args(connection)
-
     def test_1116_version(self):
         "1116 - connection version is a string"
         connection = test_env.get_connection()

tests/test_4500_connect_params.py

@@ -591,5 +591,33 @@ def test_4537_no_connect_string(self):
         params.set(mode=oracledb.SYSDBA)
         self.assertEqual(params.get_connect_string(), None)
 
+    def test_4538_dsn_with_credentials_and_connect_string(self):
+        "4538 - test parsing a DSN with credentials and a connect string"
+        params = oracledb.ConnectParams()
+        dsn = "my_user4538/my_password4538@localhost:1525/my_service_name"
+        user, password, dsn = params.parse_dsn_with_credentials(dsn)
+        self.assertEqual(user, "my_user4538")
+        self.assertEqual(password, "my_password4538")
+        self.assertEqual(dsn, "localhost:1525/my_service_name")
+
+    def test_4539_dsn_with_only_credentials(self):
+        "4539 - test parsing a DSN with only credentials"
+        params = oracledb.ConnectParams()
+        dsn = "my_user4539/my_password4539"
+        user, password, dsn = params.parse_dsn_with_credentials(dsn)
+        self.assertEqual(user, "my_user4539")
+        self.assertEqual(password, "my_password4539")
+        self.assertEqual(dsn, None)
+
+    def test_4560_dsn_with_empty_credentials(self):
+        "4560 - test parsing a DSN with empty credentials"
+        dsns = ["", "/"]
+        for dsn in ("", "/"):
+            params = oracledb.ConnectParams()
+            user, password, dsn = params.parse_dsn_with_credentials(dsn)
+            self.assertEqual(user, None)
+            self.assertEqual(password, None)
+            self.assertEqual(dsn, None)
+
 if __name__ == "__main__":
     test_env.run_test_cases()

utils/templates/connect_params.py

@@ -1,5 +1,5 @@
 #------------------------------------------------------------------------------
-# Copyright (c) 2021, 2022, Oracle and/or its affiliates.
+# Copyright (c) 2021, 2023, Oracle and/or its affiliates.
 #
 # This software is dual-licensed to you under the Universal Permissive License
 # (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
@@ -112,6 +112,17 @@ def parse_connect_string(self, connect_string: str) -> None:
         """
         self._impl.parse_connect_string(connect_string)
 
+    def parse_dsn_with_credentials(self, dsn: str) -> tuple:
+        """
+        Parses a dsn in the form <user>/<password>@<connect_string> or in the
+        form <user>/<password> and returns a 3-tuple containing the parsed
+        user, password and connect string. Empty strings are returned as the
+        value None. This is done automatically when a value is passed to
+        the dsn parameter but no value is passed to the user password when
+        creating a standalone connection or connection pool.
+        """
+        return self._impl.parse_dsn_with_credentials(dsn)
+
     @utils.params_setter
     def set(self, *,
             #{{ args_without_defaults }}

utils/templates/connection.py

@@ -1,5 +1,5 @@
 #------------------------------------------------------------------------------
-# Copyright (c) 2020, 2022, Oracle and/or its affiliates.
+# Copyright (c) 2020, 2023, Oracle and/or its affiliates.
 #
 # This software is dual-licensed to you under the Universal Permissive License
 # (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
@@ -107,12 +107,7 @@ def __init__(self,
                 errors._raise_err(errors.ERR_INVALID_CONNECT_PARAMS)
             else:
                 params_impl = params._impl.copy()
-            if kwargs:
-                params_impl.set(kwargs)
-            if dsn is not None:
-                dsn = params_impl.parse_dsn(dsn, thin)
-            if dsn is None:
-                dsn = params_impl.get_connect_string()
+            dsn = params_impl.process_args(dsn, kwargs, thin)
 
             # see if connection is being acquired from a pool
             if pool is None: