# -*- coding: utf-8 -*- # Copyright (c) 2018, 2024, Oracle and/or its affiliates. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License, version 2.0, as # published by the Free Software Foundation. # # This program is designed to work with certain software (including # but not limited to OpenSSL) that is licensed under separate terms, # as designated in a particular file or component or in included license # documentation. The authors of MySQL hereby grant you an # additional permission to link the program and your derivative works # with the separately licensed software that they have either included with # the program or referenced in the documentation. # # Without limiting anything contained in the foregoing, this file, # which is part of MySQL Connector/Python, is also subject to the # Universal FOSS Exception, version 1.0, a copy of which can be found at # http://oss.oracle.com/licenses/universal-foss-exception. # # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. # See the GNU General Public License, version 2.0, for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA """Unittests for mysqlx.client """ import os import platform import re import socket import sys import unittest from threading import Thread from time import sleep import tests import mysqlx from mysqlx.errors import InterfaceError from mysqlx.utils import linux_distribution from mysqlx.version import LICENSE, VERSION from . import check_tls_versions_support from .test_mysqlx_connection import build_uri CREATE_USER = "CREATE USER '{user}'@'{host}' IDENTIFIED BY '{password}'" GRANT_USER = "GRANT ALL ON {database}.* TO '{user}'@'{host}'" DROP_USER = "DROP USER IF EXISTS '{user}'@'{host}'" def get_current_connections(session): """Retrieves open connections using the the given session""" # Use Show process list to count the open sesions. res = session.sql("SHOW PROCESSLIST").execute() rows = res.fetch_all() connections = {} for row in rows: if row.get_string("User") not in connections: connections[row.get_string("User")] = [row.get_string("Host")] else: connections[row.get_string("User")].append(row.get_string("Host")) return connections def wait_for_connections(session, user, exp_connections, tries=5): """Waits for current connections to be closed or be the expected""" open_connections = -1 while tries > 0: current_conns = get_current_connections(session) open_connections = len(current_conns.get(user, [])) if open_connections <= exp_connections: return open_connections sleep(1) tries -= 1 return open_connections @unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 14), "XPlugin not compatible") class MySQLxClientTests(tests.MySQLxTests): def setUp(self): self.connect_kwargs = tests.get_mysqlx_config() self.schema_name = self.connect_kwargs["schema"] self.host = self.connect_kwargs["host"] try: self.session = mysqlx.get_session(self.connect_kwargs) except mysqlx.Error as err: self.fail("{0}".format(err)) self.users = [("client_user", "passclient"), ("max_size", "max_pass")] for user, password in self.users: self.session.sql(DROP_USER.format(user=user, host=self.host)).execute() self.session.sql( CREATE_USER.format(user=user, host=self.host, password=password) ).execute() # Grant all to new user on database self.session.sql( GRANT_USER.format(database=self.schema_name, user=user, host=self.host) ).execute() def tearDown(self): for user, _ in self.users: self.session.sql(DROP_USER.format(user=user, host=self.host)).execute() def test_get_client(self): """Test valid and invalid parameters of get_client().""" # Test invalid settings invalid_params = [(), (""), ({}), ("1", "2", "3"), ({}, {}, {})] for params in invalid_params: with self.assertRaises(TypeError, msg="with params {}".format(params)): mysqlx.get_client(*params) invalid_params = [ ("", ""), ({}, {}), ] for params in invalid_params: with self.assertRaises(InterfaceError, msg="with params {}".format(params)): mysqlx.get_client(*params) settings = self.connect_kwargs.copy() # Raise error for invalid values for pooling option invalid_values = [ False, True, "False", "true", "", -1, 0, 1, "1", None, ] for value in invalid_values: cnx_options = {"pooling": value} with self.assertRaises(InterfaceError, msg="with value {}".format(value)): mysqlx.get_client(settings, cnx_options) # Raise error for unrecognized settings for pooling option invalid_values = [{"max_pool_size": 100}, {"min_pool_size": "10"}] for value in invalid_values: pooling_dict = value cnx_options = {"pooling": pooling_dict} with self.assertRaises(InterfaceError): mysqlx.get_client(settings, cnx_options) # Raise error for invalid values for max_size option invalid_values = [ False, True, "False", "true", "", -1, 0, "1", None, {}, (), ] for value in invalid_values: pooling_dict = {"max_size": value} cnx_options = {"pooling": pooling_dict} with self.assertRaises(AttributeError, msg="with value {}".format(value)): mysqlx.get_client(settings, cnx_options) # Test valid values for max_size option valid_values = [1, 100] for value in valid_values: pooling_dict = {"max_size": value} cnx_options = {"pooling": pooling_dict} client = mysqlx.get_client(settings, cnx_options) self.assertEqual(client.max_size, value) client.close() # Raise error for invalid values for max_idle_time option invalid_values = [ False, True, "False", "true", "", -1, "1", None, {}, (), ] for value in invalid_values: pooling_dict = {"max_idle_time": value} cnx_options = {"pooling": pooling_dict} with self.assertRaises(AttributeError, msg="with value {}".format(value)): mysqlx.get_client(settings, cnx_options) # Test valid values for max_idle_time option valid_values = [0, 1000, 10000] exp_values = [0, 1000, 10000] for indx_value in range(len(valid_values)): value = valid_values[indx_value] pooling_dict = {"max_idle_time": value} cnx_options = {"pooling": pooling_dict} client = mysqlx.get_client(settings, cnx_options) self.assertEqual(client.max_idle_time, exp_values[indx_value]) client.close() # Raise error for invalid values for queue_timeout option invalid_values = [ False, True, "False", "true", "", -1, "1", None, {}, (), ] for value in invalid_values: pooling_dict = {"queue_timeout": value} cnx_options = {"pooling": pooling_dict} with self.assertRaises(AttributeError, msg="with value {}".format(value)): mysqlx.get_client(settings, cnx_options) # Test valid values for queue_timeout option valid_values = [0, 1000, 10000] exp_values = [0, 1000, 10000] for indx_value in range(len(valid_values)): value = valid_values[indx_value] pooling_dict = {"queue_timeout": value} cnx_options = {"pooling": pooling_dict} client = mysqlx.get_client(settings, cnx_options) self.assertEqual(client.queue_timeout, exp_values[indx_value]) client.close() # Raise error for invalid values for enabled option invalid_values = ["False", "true", "", -1, 1.5, "0", "1", 0, 1, None] for value in invalid_values: pooling_dict = {"enabled": value} cnx_options = {"pooling": pooling_dict} with self.assertRaises(AttributeError, msg="with value {}".format(value)): mysqlx.get_client(settings, cnx_options) # Test valid values for enabled option valid_values = [True, False] for value in valid_values: pooling_dict = {"enabled": value} cnx_options = {"pooling": pooling_dict} client = mysqlx.get_client(settings, cnx_options) self.assertEqual(client.pooling_enabled, value) client.close() # Test default values for pooling option for value in valid_values: pooling_dict = {} cnx_options = {"pooling": pooling_dict} client = mysqlx.get_client(settings, cnx_options) self.assertEqual(client.max_idle_time, 0) self.assertEqual(client.queue_timeout, 0) self.assertEqual(client.pooling_enabled, True) self.assertEqual(client.max_size, 25) client.close() @unittest.skipIf( tests.MYSQL_EXTERNAL_SERVER, "Test not available for external MySQL servers", ) def test_get_session(self): """Test get_session() opens new connections.""" # Auxiliary session to query server old_session = mysqlx.get_session(self.connect_kwargs.copy()) # Setup a client to get sessions from settings = self.connect_kwargs.copy() pooling_dict = {"enabled": True, "max_idle_time": 3000} cnx_options = {"pooling": pooling_dict} settings["user"] = self.users[0][0] settings["password"] = self.users[0][1] settings["host"] = self.host settings.pop("socket") client = mysqlx.get_client(settings, cnx_options) # Get 10 sessions, client will start 10 connections total_connections = 10 sessions = [] for _ in range(0, total_connections): session = client.get_session() self.assertTrue(isinstance(session, mysqlx.connection.Session)) sessions.append(session) # Verify the number of connections open in the server connections = get_current_connections(old_session) self.assertEqual(len(connections[self.users[0][0]]), total_connections) # Verify that clossing the session returns the connection # to the pool instead of being closed sessions[5].close() sessions[9].close() connections = get_current_connections(old_session) self.assertTrue(len(connections[self.users[0][0]]) >= (total_connections - 2)) if tests.MYSQL_VERSION < (8, 0, 16): # Send reset message requires the user to re-authentificate # the connection user stays in unauthenticated user open_connections = wait_for_connections( old_session, "unauthenticated user", 2 ) self.assertGreaterEqual(open_connections, 2) else: open_connections = wait_for_connections( old_session, "unauthenticated user", 0 ) self.assertEqual(open_connections, 0) # Connections must be closed when client.close() is invoked # check len(pool) == total_connections client.close() # Verify the connections on the pool are closed open_connections = wait_for_connections(old_session, self.users[0][0], 0) self.assertEqual(open_connections, 0) @unittest.skipIf( tests.MYSQL_EXTERNAL_SERVER, "Test not available for external MySQL servers", ) def test_max_pool_size(self): """Test exausted pool behavior""" # Initial pool limit size pool_limit = 5 # Setup a client to get sessions from. settings = self.connect_kwargs.copy() pooling_dict = { "enabled": True, "max_idle_time": 10000, "queue_timeout": 3000, "max_size": pool_limit, # initial pool limit } cnx_options = {"pooling": pooling_dict} settings["user"] = self.users[1][0] settings["password"] = self.users[1][1] settings["host"] = self.host settings.pop("socket") client = mysqlx.get_client(settings, cnx_options) # Get sessions as in pool_limit sessions = [] last_session = None for _ in range(0, pool_limit): session = client.get_session() self.assertTrue(isinstance(session, mysqlx.connection.Session)) sessions.append(session) last_session = session self.assertEqual(len(sessions), pool_limit) # Verify the server has open sessions as in pool_limit connections = get_current_connections(self.session) open_connections = connections.get(self.users[1][0], -1) self.assertEqual(len(open_connections), pool_limit) # verify exception is raised if the pool is exausted with self.assertRaises(mysqlx.errors.PoolError): client.get_session() # Verify that closing the last open session is returned to the pool # so then can be retrieved again, which is only possibble after the # pool has become available one more time (timeout has been reached). last_session.close() sleep(3) _ = client.get_session() client.close() # verify all sessions are closed for session in sessions: with self.assertRaises((mysqlx.errors.OperationalError, InterfaceError)): session.sql("SELECT 1").execute() session.get_schema(settings["schema"]) def test_pooling(self): """Test pooled Session works as a normal Session.""" settings = self.connect_kwargs.copy() pooling_dict = {"enabled": True} cnx_options = {"pooling": pooling_dict} client = mysqlx.get_client(settings, cnx_options) session = client.get_session() session.get_schema(settings["schema"]) session.close() # Verify that clossing the session again does not raise eceptions session.close() # Verify that trying to use a closed session raises error with self.assertRaises((mysqlx.errors.OperationalError, InterfaceError)): session.sql("SELECT 1").execute() client.close() # Verify that clossing the client again does not raise eceptions client.close() @unittest.skipIf(tests.MYSQL_VERSION < (8, 0, 16), "not reset compatible") def test_reset_keeps_same_id(self): """Test pooled Session keeps the same session id.""" settings = self.connect_kwargs.copy() pooling_dict = {"enabled": True, "max_size": 1} cnx_options = {"pooling": pooling_dict} client = mysqlx.get_client(settings, cnx_options) session1 = client.get_session() conn_id1 = session1.sql("select connection_id()").execute().fetch_all()[0][0] session1.close() # Verify that new session is has the same id from previous one session2 = client.get_session() conn_id2 = session2.sql("select connection_id()").execute().fetch_all()[0][0] self.assertEqual(conn_id1, conn_id2, "The connection id was not the same") session2.close() client.close() @unittest.skipIf(tests.MYSQL_VERSION < (8, 0, 16), "not reset compatible") def test_reset_get_new_connection(self): """Test connection is closed by max idle time.""" # Verify max_idle_time closses an idle session and so for the # connection id must increase due to the pool which has created a # new connection to the server to return a new session. settings = self.connect_kwargs.copy() pooling_dict = { "enabled": True, "max_idle_time": 2000, "max_size": 3, "queue_timeout": 1000, } cnx_options = {"pooling": pooling_dict} client = mysqlx.get_client(settings, cnx_options) # Getting session 0 session0 = client.get_session() conn_id0 = session0.sql("select connection_id()").execute().fetch_all()[0][0] # Closing session 0 session0.close() # Getting session 1 session1 = client.get_session() conn_id1 = session1.sql("select connection_id()").execute().fetch_all()[0][0] # Closing session 1 self.assertEqual(conn_id1, conn_id0, "The connection id was not greater") session1.close() # Verify that new session does not has the same id from previous one # goint to sleep 2 sec just above the max idle time sleep(4) # Getting session 2 session2 = client.get_session() conn_id2 = session2.sql("select connection_id()").execute().fetch_all()[0][0] self.assertNotEqual( conn_id0, conn_id2, "The connection id was the same from the old" ) self.assertNotEqual( conn_id1, conn_id2, "The connection id was the same from the old" ) # Verify pool integrity # Open the 4th connection in client life time, when max size is 3 # Getting the max allowed connections # Getting session 3 session3 = client.get_session() conn_id3 = session3.sql("select connection_id()").execute().fetch_all()[0][0] self.assertGreater(conn_id3, conn_id2, "The connection id was not greater") # Getting session 4 session4 = client.get_session() # Verify exception is raised if the pool is exausted with self.assertRaises(mysqlx.errors.PoolError): client.get_session() # closing all connections session2.close() session3.close() session4.close() # No errors should raise by closing it again session4.close() # No errors should raise from closing client. client.close() @unittest.skipIf( sys.platform == "darwin" and platform.mac_ver()[0].startswith("12"), "This test fails due to a bug on macOS 12", ) @unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 40), "TLSv1.1 incompatible") @unittest.skipIf( tests.MYSQL_VERSION > (8, 0, 27), "TLSv1 and TLSv1.1 support removed as of MySQL 8.0.28", ) @unittest.skipIf( tests.MYSQL_VERSION > (5, 7, 36) and tests.MYSQL_VERSION < (5, 8, 0), "TLSv1 and TLSv1.1 support removed as of MySQL 5.7.37", ) def test_get_client_with_tls_version(self): # Test None value is returned if no schema name is specified settings = self.connect_kwargs.copy() settings.pop("schema") settings.pop("socket") pooling_dict = { "enabled": True, "max_idle_time": 2000, "max_size": 3, "queue_timeout": 1000, } cnx_options = {"pooling": pooling_dict} # Dictionary connection settings tests # Empty tls_version list settings["tls-ciphersuites"] = ["DHE-RSA-AES256-SHA"] settings["tls-versions"] = [] with self.assertRaises(InterfaceError) as context: _ = mysqlx.get_client(settings, cnx_options) self.assertTrue( ("At least one" in context.exception.msg), "Unexpected " "exception message found: {}" "".format(context.exception.msg), ) # Empty tls_ciphersuites list settings["tls-ciphersuites"] = [] settings["tls-versions"] = ["TLSv1.2"] with self.assertRaises(InterfaceError) as context: _ = mysqlx.get_client(settings, cnx_options) self.assertTrue( ("No valid cipher" in context.exception.msg), "Unexpected exception message found: {}".format(context.exception.msg), ) # URI string connection settings tests # Empty tls_version list settings["tls-ciphersuites"] = ["DHE-RSA-AES256-SHA"] settings["tls-versions"] = [] uri_settings = build_uri(**settings) with self.assertRaises(InterfaceError) as context: _ = mysqlx.get_client(uri_settings, cnx_options) self.assertTrue( ("At least one" in context.exception.msg), "Unexpected " "exception message found: {}" "".format(context.exception.msg), ) # Empty tls_ciphersuites list settings["tls-ciphersuites"] = [] settings["tls-versions"] = ["TLSv1.2"] uri_settings = build_uri(**settings) with self.assertRaises(InterfaceError) as context: _ = mysqlx.get_client(uri_settings, cnx_options) self.assertTrue( ("No valid cipher" in context.exception.msg), "Unexpected exception message found: {}".format(context.exception.msg), ) # Verify InterfaceError exception is raised With invalid TLS version settings["tls-ciphersuites"] = ["DHE-RSA-AES256-SHA"] settings["tls-versions"] = ["TLSv8"] uri_settings = build_uri(**settings) with self.assertRaises(InterfaceError) as context: _ = mysqlx.get_client(uri_settings, cnx_options) self.assertTrue( ("not recognized" in context.exception.msg), "Unexpected exception message found: {}".format(context.exception.msg), ) # Verify unkown cipher suite case settings["tls-ciphersuites"] = ["NOT-KNOWN"] settings["tls-versions"] = ["TLSv1.2"] uri_settings = build_uri(**settings) with self.assertRaises(InterfaceError) as context: _ = mysqlx.get_client(uri_settings, cnx_options) # Verify unsupported TLSv1.3 version is ignored (connection success) # when is not supported, TLSv1.2 is used settings["tls-ciphersuites"] = ["DHE-RSA-AES256-GCM-SHA384"] settings["tls-versions"] = ["TLSv1.3", "TLSv1.2"] uri_settings = build_uri(**settings) client = mysqlx.get_client(uri_settings, cnx_options) session = client.get_session() session.close() client.close() supported_tls = check_tls_versions_support(["TLSv1.2", "TLSv1.1", "TLSv1"]) if not supported_tls: self.fail("No TLS version to test: {}".format(supported_tls)) if len(supported_tls) > 1: # Verify given TLS version is used settings["tls-ciphersuites"] = ["DHE-RSA-AES256-SHA"] for tes_ver in supported_tls: settings["tls-versions"] = [tes_ver] uri_settings = build_uri(**settings) client = mysqlx.get_client(uri_settings, cnx_options) session = client.get_session() res = session.sql("SHOW STATUS").execute().fetch_all() for row in res: if row.get_string("Variable_name") == "Mysqlx_ssl_version": self.assertEqual( row.get_string("Value"), tes_ver, "Unexpected TLS version found: {} for: {}" "".format(row.get_string("Value"), tes_ver), ) session.close() client.close() # Following tests requires TLSv1.2 if tests.MYSQL_VERSION < (8, 0, 17): return if "TLSv1.1" in supported_tls: # Verify the newest TLS version is used from the given list exp_res = ["TLSv1.2", "TLSv1.1", "TLSv1.2"] test_vers = [ ["TLSv1", "TLSv1.2", "TLSv1.1"], ["TLSv1", "TLSv1.1"], ["TLSv1.2", "TLSv1"], ] for tes_ver, exp_ver in zip(test_vers, exp_res): settings["tls-versions"] = tes_ver uri_settings = build_uri(**settings) client = mysqlx.get_client(uri_settings, cnx_options) session = client.get_session() res = session.sql("SHOW STATUS").execute().fetch_all() for row in res: if row.get_string("Variable_name") == "Mysqlx_ssl_version": self.assertEqual( row.get_string("Value"), exp_ver, "Unexpected TLS version found: {}" "".format(row.get_string("Value")), ) session.close() client.close() # Verify given TLS cipher suite is used exp_res = [ "ECDHE-RSA-AES256-GCM-SHA384", "DHE-RSA-AES256-GCM-SHA384", "DHE-RSA-AES128-GCM-SHA256", ] test_ciphers = [ ["TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"], ["TLS_DHE_RSA_WITH_AES_256_GCM_SHA384"], ["TLS_DHE_RSA_WITH_AES_128_GCM_SHA256"], ] settings["tls-versions"] = "TLSv1.2" for test_cipher, exp_ver in zip(test_ciphers, exp_res): settings["tls-ciphersuites"] = test_cipher uri_settings = build_uri(**settings) client = mysqlx.get_client(uri_settings, cnx_options) session = client.get_session() res = session.sql("SHOW STATUS").execute().fetch_all() for row in res: if row.get_string("Variable_name") == "Mysqlx_ssl_cipher": self.assertEqual( row.get_string("Value"), exp_ver, "Unexpected TLS version found: {} for: {}" "".format(row.get_string("Value"), test_cipher), ) session.close() client.close() # Verify one of TLS cipher suite is used from the given list exp_res = [ "DHE-RSA-AES256-SHA256", "DHE-RSA-AES256-SHA256", "DHE-RSA-AES128-GCM-SHA256", ] test_ciphers = [ "TLS_DHE_RSA_WITH_AES_256_CBC_SHA256", "DHE-RSA-AES256-SHA256", "TLS_DHE_RSA_WITH_AES_128_GCM_SHA256", ] settings["tls_ciphersuites"] = test_ciphers settings["tls_versions"] = "TLSv1.2" uri_settings = build_uri(**settings) client = mysqlx.get_client(uri_settings, cnx_options) session = client.get_session() res = session.sql("SHOW STATUS").execute().fetch_all() for row in res: if row.get_string("Variable_name") == "Mysqlx_ssl_cipher": self.assertIn( row.get_string("Value"), exp_res, "Unexpected TLS version found: {} not in {}" "".format(row.get_string("Value"), exp_res), ) session.close() client.close() def test_context_manager(self): """Test mysqlx.get_client() context manager.""" settings = self.connect_kwargs.copy() pooling_dict = {"enabled": True, "max_size": 5} cnx_options = {"pooling": pooling_dict} with mysqlx.get_client(settings, cnx_options) as client: with client.get_session() as session: self.assertIsInstance(session, mysqlx.Session) self.assertTrue(session.is_open()) self.assertFalse(session.is_open()) # Create one more session _ = client.get_session() for session in client.sessions: self.assertFalse(session.is_open()) @unittest.skipIf( tests.MYSQL_EXTERNAL_SERVER, "Test not available for external MySQL servers", ) @unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 12), "XPlugin not compatible") class MySQLxClientPoolingTests(tests.MySQLxTests): def setUp(self): settings = tests.get_mysqlx_config() self.schema_name = settings["schema"] self.host = settings["host"] self.user = "router_user" self.password = "passpool" self.hosts = "127.0.0.1", "localhost" try: self.session = mysqlx.get_session(settings) except mysqlx.Error as err: self.fail("{0}".format(err)) for host in self.hosts: self.session.sql(DROP_USER.format(user=self.user, host=host)).execute() for host in self.hosts: self.session.sql( CREATE_USER.format(user=self.user, host=host, password=self.password) ).execute() # Grant all to new user on database self.session.sql( GRANT_USER.format( database=self.schema_name, user=self.user, host=host, password=self.password, ) ).execute() def tearDown(self): session = mysqlx.get_session(tests.get_mysqlx_config()) for host in self.hosts: session.sql(DROP_USER.format(user=self.user, host=host)).execute() def test_routing(self): settings = tests.get_mysqlx_config() pooling_dict = {"enabled": True} cnx_options = {"pooling": pooling_dict} uri = ( "mysqlx://{user}:{pwd}@[(address=1.0.0.2:{port}, priority=30)," " (address=1.0.0.1:{port}, priority=40)," " (address=127.0.0.1:{port}, priority=20)," " (address=localhost:{port}, priority=50)]" "".format( user=settings["user"], pwd=settings["password"], port=settings["port"], ) ) client = mysqlx.get_client(uri, cnx_options) # Getting a session must success, the higher priority will cause to a # valid address to be used client.get_session() client.close() def test_pools_are_not_shared(self): settings = tests.get_mysqlx_config() pooling_dict = { "max_size": 1, "max_idle_time": 10000, } cnx_options = {"pooling": pooling_dict} uri = ( "mysqlx://{user}:{pwd}@[" " (address={host}:{port}, priority=50)]" "".format( user=self.user, pwd=self.password, host=settings["host"], port=settings["port"], ) ) client1 = mysqlx.get_client(uri, cnx_options) # Getting a session from client1 session1 = client1.get_session() session1.sql("SELECT 1").execute() session1.get_schema(settings["schema"]) # Getting a session from client2 client2 = mysqlx.get_client(uri, cnx_options) # Getting a session must success, the higher priority will cause to a # valid address to be used session2 = client2.get_session() session2.sql("SELECT 2").execute() # Verify the server connections connections = get_current_connections(self.session) open_connections = connections.get(self.user, -1) self.assertEqual(len(open_connections), 2) # Closing pools in client1 must not close connections in client2 client1.close() with self.assertRaises((mysqlx.errors.OperationalError, InterfaceError)): session1.sql("SELECT 1").execute() session2.sql("SELECT 2").execute() session2.get_schema(settings["schema"]) client2.close() def test_routing_random(self): settings = tests.get_mysqlx_config() pooling_dict = {"enabled": True, "max_size": 5, "queue_timeout": 1000} cnx_options = {"pooling": pooling_dict} uri = ( "mysqlx://{user}:{pwd}@[" "(address=1.0.0.1:{port}, priority=20)," " (address=1.0.0.2:{port}, priority=20)," " (address=127.0.0.1:{port}, priority=30)," " (address=localhost:{port}, priority=30)," " (address=1.0.0.3:{port}, priority=60)," " (address=1.0.0.4:{port}, priority=60)]" "".format( user=settings["user"], pwd=settings["password"], port=settings["port"], ) ) client = mysqlx.get_client(uri, cnx_options) # Getting a session must success, the higher priority will cause to a # valid address to be used sessions = [] for _ in range(10): session = client.get_session() sessions.append(session) client.close() # Test routers without priority settings = tests.get_mysqlx_config() pooling_dict = {"enabled": True, "max_size": 5, "queue_timeout": 1000} cnx_options = {"pooling": pooling_dict} uri = ( "mysqlx://{user}:{pwd}@[" "(address=1.0.0.2:{port})," " (address=1.0.0.1:{port})," " (address=127.0.0.1:{port})," " (address=localhost:{port})," " (address=127.0.0.1:{port})," " (address=localhost:{port})]" "".format( user=settings["user"], pwd=settings["password"], port=settings["port"], ) ) client = mysqlx.get_client(uri, cnx_options) # Getting the total of 10 sessions must success sessions = [] for _ in range(10): session = client.get_session() sessions.append(session) # verify error is thrown when the total sum of the pool sizes is reached with self.assertRaises(mysqlx.errors.PoolError) as context: _ = client.get_session() self.assertTrue( ("pool max size has been reached" in context.exception.msg), "Unexpected exception message found: {}".format(context.exception.msg), ) client.close() # Verify "Unable to connect to any of the target hosts" error message settings = tests.get_mysqlx_config() pooling_dict = {"enabled": True, "max_size": 5, "queue_timeout": 50} cnx_options = {"pooling": pooling_dict} uri = ( "mysqlx://{user}:{pwd}@[" "(address=1.0.0.1:{port})," " (address=1.0.0.2:{port})," " (address=1.0.0.3:{port})," " (address=1.0.0.4:{port})," " (address=1.0.0.5:{port})," " (address=1.0.0.6:{port})," " (address=1.0.0.7:{port})," " (address=1.0.0.8:{port})," " (address=1.0.0.9:{port})," " (address=1.0.0.10:{port})]?connect-timeout=500" "".format( user=settings["user"], pwd=settings["password"] + "$%^", port=settings["port"], ) ) client = mysqlx.get_client(uri, cnx_options) with self.assertRaises(mysqlx.errors.PoolError) as context: _ = client.get_session() self.assertTrue( ("Unable to connect to any of the target hosts" in context.exception.msg), "Unexpected exception message found: {}".format(context.exception.msg), ) try: client.close() except: # due to the small connect-timeout closing the client may fail. pass # Verify random order of the pools ip_orders = {} attempts = 10 for _ in range(attempts): client = mysqlx.get_client(uri, cnx_options) with self.assertRaises(mysqlx.errors.PoolError) as context: _ = client.get_session() re_ip = re.compile(" pool: 1.0.0.(\d+).*") order_list = [] for line in context.exception.msg.splitlines(): if "pool:" not in line: continue match = re_ip.match(line) if match: order_list.append(match.group(1)) # Verify the 10 pools were verified self.assertEqual( len(order_list), 10, "10 exception messages were expected but found: {}" "".format(context.exception.msg), ) key = ",".join(order_list) if not key in ip_orders: ip_orders[key] = 1 else: ip_orders[key] = ip_orders[key] + 1 try: client.close() except: # due to the small connect-timeout closing the client may fail. pass max_repeated = -1 for ip_order in ip_orders: cur = ip_orders[ip_order] max_repeated = cur if cur > max_repeated else max_repeated # The possiblility of getting 2 times the same order is : ((1/(10!))^2)*9 # Getting the same number of different orders than attempts ensures no repetitions. self.assertEqual( len(ip_orders), attempts, "Expected less repetions found: {}" "".format( [ "ip_order: {} reps: {}".format(ip_order, ip_orders[ip_order]) for ip_order in ip_orders ] ), ) @unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 14), "XPlugin not compatible") class MySQLxConnectionPoolingTests(tests.MySQLxTests): def setUp(self): settings = tests.get_mysqlx_config() self.schema_name = settings["schema"] self.host = settings["host"] self.user = "pool_user" self.password = "passpool" self.hosts = "127.0.0.1", "localhost" try: self.session = mysqlx.get_session(settings) except mysqlx.Error as err: self.fail("{0}".format(err)) for host in self.hosts: self.session.sql(DROP_USER.format(user=self.user, host=host)).execute() for host in self.hosts: self.session.sql( CREATE_USER.format(user=self.user, host=host, password=self.password) ).execute() # Grant all to new user on database self.session.sql( GRANT_USER.format( database=self.schema_name, user=self.user, host=host, password=self.password, ) ).execute() def tearDown(self): for host in self.hosts: self.session.sql(DROP_USER.format(user=self.user, host=host)).execute() @unittest.skipIf( tests.MYSQL_EXTERNAL_SERVER, "Test not available for external MySQL servers", ) def test_pools_recycle(self): settings = tests.get_mysqlx_config() pooling_dict = { "max_size": 1, "max_idle_time": 3000, "queue_timeout": 1000, } cnx_options = {"pooling": pooling_dict} uri = ( "mysqlx://{user}:{pwd}@[" " (address={host}:{port}, priority=50)]?connect_timeout=20000" "".format( user=self.user, pwd=self.password, host=settings["host"], port=settings["port"], ) ) client = mysqlx.get_client(uri, cnx_options) def thread1(client): # Getting a session from client session1 = client.get_session() session1.sql("SELECT 1").execute() sleep(2) session1.close() def thread2(client): # Getting a session from client session2 = client.get_session() session2.sql("SELECT 2").execute() sleep(1) session2.close() worker1 = Thread(target=thread1, args=[client]) worker1.start() worker1.join() sleep(0.5) worker2 = Thread(target=thread2, args=[client]) worker2.start() worker2.join() # Verify the server connections connections = get_current_connections(self.session) open_connections = connections.get("unauthenticated user", []) if tests.MYSQL_VERSION < (8, 0, 16): # Send reset message requires the user to re-authentificate # the connection user stays in unauthenticated user self.assertTrue(len(open_connections) >= 1) else: self.assertEqual(len(open_connections), 0) client.close() class MySQLxClientConnectionAttributesTests(tests.MySQLConnectorTests): def setUp(self): self.connect_kwargs = tests.get_mysqlx_config() self.schema_name = self.connect_kwargs["schema"] try: self.client = mysqlx.get_client(self.connect_kwargs, "{}") self.session = self.client.get_session() except mysqlx.Error as err: self.fail("{0}".format(err)) if os.name == "nt": if "64" in platform.architecture()[0]: self.platform_arch = "x86_64" elif "32" in platform.architecture()[0]: self.platform_arch = "i386" else: self.platform_arch = platform.architecture() self.os_ver = "Windows-{}".format(platform.win32_ver()[1]) else: self.platform_arch = platform.machine() if platform.system() == "Darwin": self.os_ver = "{}-{}".format("macOS", platform.mac_ver()[0]) else: self.os_ver = "-".join(linux_distribution()[0:2]) license_chunks = LICENSE.split(" ") if license_chunks[0] == "GPLv2": self.client_license = "GPL-2.0" else: self.client_license = "Commercial" @unittest.skipIf(tests.MYSQL_VERSION < (8, 0, 16), "XPlugin not compatible") def test_connection_attributes(self): # Validate an error is raised if URL user defined connection attributes # through a connection URL name is duplicate connection_attributes = { "foo": "bar", "repeated": "attribute", "baz": "zoom", } uri = build_uri( user=self.connect_kwargs["user"], password=self.connect_kwargs["password"], host=self.connect_kwargs["host"], port=self.connect_kwargs["port"], schema=self.connect_kwargs["schema"], connection_attributes=connection_attributes, ) uri = "{},repeated=duplicate_attribute]".format(uri[0:-1]) with self.assertRaises(InterfaceError) as context: my_client = mysqlx.get_client(uri, "{}") _ = my_client.get_session() self.assertTrue( "Duplicate key " in context.exception.msg, "error found: {}".format(context.exception.msg), ) # Test error is raised for attribute name starting with '_' connection_attributes = [ {"foo": "bar", "_baz": "zoom"}, {"_baz": "zoom"}, {"foo": "bar", "_baz": "zoom", "puuuuum": "kaplot"}, ] for conn_attr in connection_attributes: connect_kwargs = self.connect_kwargs.copy() connect_kwargs["connection_attributes"] = conn_attr with self.assertRaises(InterfaceError) as context: my_client = mysqlx.get_client(connect_kwargs, "{}") _ = my_client.get_session() self.assertTrue("connection-attributes" in context.exception.msg) self.assertTrue("cannot start with '_'" in context.exception.msg) # Test error is raised for attribute name size exceeds 32 characters connection_attributes = [ {"foo": "bar", "p{}w".format("o" * 31): "kaplot"}, {"p{}w".format("o" * 31): "kaplot"}, {"baz": "zoom", "p{}w".format("o" * 31): "kaplot", "a": "b"}, ] for conn_attr in connection_attributes: connect_kwargs = self.connect_kwargs.copy() connect_kwargs["connection_attributes"] = conn_attr with self.assertRaises(InterfaceError) as context: my_client = mysqlx.get_client(connect_kwargs, "{}") _ = my_client.get_session() self.assertTrue("exceeds 32 characters limit size" in context.exception.msg) # Test error is raised for attribute value size exceeds 1024 characters connection_attributes = [ {"foo": "bar", "pum": "kr{}nk".format("u" * 1024)}, {"pum": "kr{}nk".format("u" * 1024)}, {"baz": "zoom", "pum": "kr{}nk".format("u" * 1024), "a": "b"}, ] for conn_attr in connection_attributes: connect_kwargs = self.connect_kwargs.copy() connect_kwargs["connection-attributes"] = conn_attr with self.assertRaises(InterfaceError) as context: my_client = mysqlx.get_client(connect_kwargs, "{}") _ = my_client.get_session() self.assertTrue( "exceeds 1024 characters limit size" in context.exception.msg ) # Validate the user defined attributes are created in the server # Test user defined connection attributes through a connection URL connection_attributes = { "foo": "bar", "baz": "zoom", "quash": "", "puuuuum": "kaplot", } uri = build_uri( user=self.connect_kwargs["user"], password=self.connect_kwargs["password"], host=self.connect_kwargs["host"], port=self.connect_kwargs["port"], schema=self.connect_kwargs["schema"], connection_attributes=connection_attributes, ) # Verify user defined session-connection-attributes are in the server my_session = mysqlx.get_session(uri) row = ( my_session.sql('SHOW VARIABLES LIKE "pseudo_thread_id"') .execute() .fetch_all()[0] ) get_attrs = ( "SELECT ATTR_NAME, ATTR_VALUE FROM " "performance_schema.session_account_connect_attrs " 'where PROCESSLIST_ID = "{}"' ) rows = ( my_session.sql(get_attrs.format(row.get_string("Value"))) .execute() .fetch_all() ) expected_attrs = connection_attributes.copy() expected_attrs.update( { "_pid": str(os.getpid()), "_platform": self.platform_arch, "_source_host": socket.gethostname(), "_client_name": "mysql-connector-python", "_client_license": self.client_license, "_client_version": ".".join([str(x) for x in VERSION[0:3]]), "_os": self.os_ver, } ) # Note that for an empty string "" value the server stores a Null value expected_attrs["quash"] = "None" for row in rows: attr_name, attr_value = ( row["ATTR_NAME"].decode(), row["ATTR_VALUE"].decode() if row["ATTR_VALUE"] else "None", ) self.assertEqual( expected_attrs[attr_name], attr_value, "Attribute {} with value {} differs of {}".format( attr_name, attr_value, expected_attrs[attr_name] ), ) @unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 14), "XPlugin not compatible") class MySQLxPoolingSessionTests(tests.MySQLxTests): def setUp(self): self.connect_kwargs = tests.get_mysqlx_config() self.session = mysqlx.get_session(self.connect_kwargs) self.session.sql("DROP DATABASE IF EXISTS my_test_schema").execute() def test_get_default_schema(self): pooling_dict = { "max_size": 1, "max_idle_time": 3000, "queue_timeout": 1000, } # Test None value is returned if no schema name is specified settings = self.connect_kwargs.copy() settings.pop("schema") client = mysqlx.get_client(settings, pooling_dict) session = client.get_session() schema = session.get_default_schema() self.assertIsNone(schema, "None value was expected but got '{}'".format(schema)) session.close() # Test SQL statements not fully qualified, which must not raise error: # mysqlx.errors.OperationalError: No database selected self.session.sql("CREATE DATABASE my_test_schema").execute() self.session.sql("CREATE TABLE my_test_schema.pets(name VARCHAR(20))").execute() settings = self.connect_kwargs.copy() settings["schema"] = "my_test_schema" client = mysqlx.get_client(settings, pooling_dict) session = client.get_session() schema = session.get_default_schema() self.assertTrue(schema, mysqlx.Schema) self.assertEqual(schema.get_name(), "my_test_schema") result = session.sql("SHOW TABLES").execute().fetch_all() self.assertEqual("pets", result[0][0]) self.session.sql("DROP DATABASE my_test_schema").execute() self.assertFalse(schema.exists_in_database()) self.assertRaises(mysqlx.ProgrammingError, session.get_default_schema) session.close() client.close() # Test without default schema configured at connect time (passing None) settings = self.connect_kwargs.copy() settings["schema"] = None client = mysqlx.get_client(settings, pooling_dict) session = client.get_session() schema = session.get_default_schema() self.assertIsNone(schema, "None value was expected but got '{}'".format(schema)) session.close() client.close() # Test not existing default schema at get_session raise error settings = self.connect_kwargs.copy() settings["schema"] = "nonexistent" client = mysqlx.get_client(settings, pooling_dict) self.assertRaises(InterfaceError, client.get_session)