BUG#33923516: Allow tuple of dictionaries as "failover" argument

nmariz · nmariz · commit 8d1c9c5f7f51 · 2022-04-21T11:39:10.000+01:00
In Connector/Python documentation states that the connect() method
accepts a failover argument that provides information to use for the
server failover in the event of connection failures, and this argument
value should be a tuple or a list of dictionaries.

Using a tuple fails with an AttributeError exception.

This patch fixes this issue by applying the sorted() function instead
of using the sort() method in the failover argument, for sorting the
servers by priority.

Change-Id: I8d6911d70d425317374539a830a839ca1cff751f
diff --git a/CHANGES.txt b/CHANGES.txt
@@ -16,6 +16,7 @@ v8.0.30
 - WL#15035: Enforce PEP 7 and PEP 8 coding style
 - WL#14822: Refactor the authentication plugin mechanism
 - WL#14815: Support OpenSSL 3.0
+- BUG#33923516: Allow tuple of dictionaries as "failover" argument
 - BUG#28821983: Fix rounding errors for decimal values
 
 v8.0.29
diff --git a/lib/mysql/connector/pooling.py b/lib/mysql/connector/pooling.py
@@ -164,11 +164,9 @@ def _get_failover_connection(**kwargs):
             "every router"
         )
 
-    failover.sort(key=lambda x: x["priority"], reverse=True)
-
     server_directory = {}
     server_priority_list = []
-    for server in failover:
+    for server in sorted(failover, key=lambda x: x["priority"], reverse=True):
         if server["priority"] not in server_directory:
             server_directory[server["priority"]] = [server]
             server_priority_list.append(server["priority"])
diff --git a/tests/test_connection.py b/tests/test_connection.py
@@ -3486,7 +3486,7 @@ def test_failover(self):
             "passwd": passw,
         }
 
-        # connection must fail with error: "Unable to connect to any of the target hosts"
+        # Connection must fail with error: "Unable to connect to any of the target hosts"
         with self.assertRaises(InterfaceError) as context:
             _ = connect(**settings)
 
@@ -3499,8 +3499,13 @@ def test_failover(self):
         settings["failover"].append(
             {"host": host, "port": port, "user": user, "priority": 100}
         )
-        # connection must be successful
-        _ = connect(**cnx_config)
+        # Connection must be successful
+        _ = connect(**settings)
+
+        # Test 'failover' using a tuple instead
+        settings["failover"] = tuple(settings["failover"])
+        # Connection must be successful
+        _ = connect(**settings)
 
 
 @unittest.skipIf(
