# Copyright (c) 2009, 2020, Oracle and/or its affiliates. All rights reserved.
#
# 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 also distributed 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 included with
# MySQL.
#
# 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 mysql.connector.connection
"""

import copy
import os
import logging
import platform
import timeit
import unittest
from decimal import Decimal
import io
import socket
import sys

import tests
from . import check_tls_versions_support
from . import PY2

try:
    from mysql.connector.connection_cext import HAVE_CMYSQL, CMySQLConnection
except ImportError:
    # Test without C Extension
    CMySQLConnection = None
    HAVE_CMYSQL = False

from mysql.connector.conversion import (MySQLConverterBase, MySQLConverter)
from mysql.connector import (connect, connection, network, errors,
                             constants, cursor, abstracts, catch23,
                             HAVE_DNSPYTHON)
from mysql.connector.errors import InterfaceError
from mysql.connector.optionfiles import read_option_files
from mysql.connector.network import TLS_V1_3_SUPPORTED
from mysql.connector.utils import linux_distribution
from mysql.connector.version import VERSION, LICENSE

LOGGER = logging.getLogger(tests.LOGGER_NAME)

OK_PACKET = bytearray(b'\x07\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00')
OK_PACKET_RESULT = {
    'insert_id': 0,
    'affected_rows': 0,
    'field_count': 0,
    'warning_count': 0,
    'status_flag': 0
}

ERR_PACKET = bytearray(
    b'\x47\x00\x00\x02\xff\x15\x04\x23\x32\x38\x30\x30\x30'
    b'\x41\x63\x63\x65\x73\x73\x20\x64\x65\x6e\x69\x65\x64'
    b'\x20\x66\x6f\x72\x20\x75\x73\x65\x72\x20\x27\x68\x61'
    b'\x6d\x27\x40\x27\x6c\x6f\x63\x61\x6c\x68\x6f\x73\x74'
    b'\x27\x20\x28\x75\x73\x69\x6e\x67\x20\x70\x61\x73\x73'
    b'\x77\x6f\x72\x64\x3a\x20\x59\x45\x53\x29'
)

EOF_PACKET = bytearray(b'\x05\x00\x00\x00\xfe\x00\x00\x00\x00')
EOF_PACKET_RESULT = {'status_flag': 0, 'warning_count': 0}

COLUMNS_SINGLE = bytearray(
    b'\x17\x00\x00\x02\x03\x64\x65\x66\x00\x00\x00\x01'
    b'\x31\x00\x0c\x3f\x00\x01\x00\x00\x00\x08\x81\x00'
    b'\x00\x00\x00'
)

COLUMNS_SINGLE_COUNT = bytearray(b'\x01\x00\x00\x01\x01')

class _DummyMySQLConnection(connection.MySQLConnection):
    def _open_connection(self, *args, **kwargs):
        pass

    def _post_connection(self, *args, **kwargs):
        pass


class ConnectionTests(object):
    def test_DEFAULT_CONFIGURATION(self):
        exp = {
            'database': None,
            'user': '',
            'password': '',
            'host': '127.0.0.1',
            'port': 3306,
            'unix_socket': None,
            'use_unicode': True,
            'charset': 'utf8',
            'collation': None,
            'converter_class': MySQLConverter,
            'autocommit': False,
            'time_zone': None,
            'sql_mode': None,
            'get_warnings': False,
            'raise_on_warnings': False,
            'connection_timeout': None,
            'client_flags': 0,
            'compress': False,
            'buffered': False,
            'raw': False,
            'ssl_ca': None,
            'ssl_cert': None,
            'ssl_key': None,
            'ssl_verify_cert': False,
            'ssl_verify_identity': False,
            'passwd': None,
            'db': None,
            'connect_timeout': None,
            'dsn': None,
            'force_ipv6': False,
            'auth_plugin': None,
            'allow_local_infile': True,
            'consume_results': False,
        }
        self.assertEqual(exp, connection.DEFAULT_CONFIGURATION)


class MySQLConnectionTests(tests.MySQLConnectorTests):
    def setUp(self):
        config = tests.get_mysql_config()
        self.cnx = connection.MySQLConnection(**config)

    def tearDown(self):
        try:
            self.cnx.close()
        except:
            pass

    def test_init(self):
        """MySQLConnection initialization"""
        cnx = connection.MySQLConnection()
        exp = {
            'converter': None,
            '_converter_class': MySQLConverter,
            '_client_flags': constants.ClientFlag.get_default(),
            '_charset_id': 45,
            '_user': '',
            '_password': '',
            '_database': '',
            '_host': '127.0.0.1',
            '_port': 3306,
            '_unix_socket': None,
            '_use_unicode': True,
            '_get_warnings': False,
            '_raise_on_warnings': False,
            '_connection_timeout': None,
            '_buffered': False,
            '_unread_result': False,
            '_have_next_result': False,
            '_raw': False,
            '_client_host': '',
            '_client_port': 0,
            '_ssl': {},
            '_in_transaction': False,
            '_force_ipv6': False,
            '_auth_plugin': None,
            '_pool_config_version': None,
            '_consume_results': False,
        }
        for key, value in exp.items():
            self.assertEqual(
                value, cnx.__dict__[key],
                msg="Default for '{0}' did not match.".format(key))

        # Make sure that when at least one argument is given,
        # connect() is called
        class FakeMySQLConnection(connection.MySQLConnection):
            def connect(self, **kwargs):
                self._database = kwargs['database']

        exp = 'test'
        cnx = FakeMySQLConnection(database=exp)
        self.assertEqual(exp, cnx._database)

    def test__get_self(self):
        """Return self"""
        self.assertEqual(self.cnx, self.cnx._get_self())

    def test__send_cmd(self):
        """Send a command to MySQL"""
        cmd = constants.ServerCmd.QUERY
        arg = 'SELECT 1'.encode('utf-8')
        pktnr = 2

        self.cnx._socket.sock = None
        self.assertRaises(errors.OperationalError, self.cnx._send_cmd,
                          cmd, arg, pktnr)

        self.cnx._socket.sock = tests.DummySocket()
        exp = OK_PACKET
        self.cnx._socket.sock.add_packet(exp)
        res = self.cnx._send_cmd(cmd, arg, pktnr)
        self.assertEqual(exp, res)

        # Send an unknown command, the result should be an error packet
        exp = ERR_PACKET
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(exp)
        res = self.cnx._send_cmd(90, b'spam', 0)
        self.assertEqual(exp, res)

    def test__handle_server_status(self):
        """Handle the server/status flags"""
        cases = [
            # (serverflag, attribute_name, value when set, value when unset)
            (constants.ServerFlag.MORE_RESULTS_EXISTS,
             '_have_next_result', True, False),
            (constants.ServerFlag.STATUS_IN_TRANS,
             '_in_transaction', True, False),
        ]
        for (flag, attr, when_set, when_unset) in cases:
            setattr(self.cnx, attr, when_unset)
            self.cnx._handle_server_status(flag)
            self.assertEqual(when_set, getattr(self.cnx, attr))
            self.cnx._handle_server_status(0)
            self.assertEqual(when_unset, getattr(self.cnx, attr))

    def test__handle_ok(self):
        """Handle an OK-packet sent by MySQL"""
        self.assertEqual(OK_PACKET_RESULT, self.cnx._handle_ok(OK_PACKET))
        self.assertRaises(errors.ProgrammingError,
                          self.cnx._handle_ok, ERR_PACKET)
        self.assertRaises(errors.InterfaceError,
                          self.cnx._handle_ok, EOF_PACKET)

        # Test for multiple results
        self.cnx._have_next_result = False
        packet = OK_PACKET[:-4] + b'\x08' + OK_PACKET[-3:]
        self.cnx._handle_ok(packet)
        self.assertTrue(self.cnx._have_next_result)

    def test__handle_eof(self):
        """Handle an EOF-packet sent by MySQL"""
        self.assertEqual(EOF_PACKET_RESULT, self.cnx._handle_eof(EOF_PACKET))
        self.assertRaises(errors.ProgrammingError,
                          self.cnx._handle_eof, ERR_PACKET)
        self.assertRaises(errors.InterfaceError,
                          self.cnx._handle_eof, OK_PACKET)

        # Test for multiple results
        self.cnx._have_next_result = False
        packet = EOF_PACKET[:-2] + b'\x08' + EOF_PACKET[-1:]
        self.cnx._handle_eof(packet)
        self.assertTrue(self.cnx._have_next_result)

    def test__handle_result(self):
        """Handle the result after sending a command to MySQL"""
        self.assertRaises(errors.InterfaceError, self.cnx._handle_result,
                          '\x00')
        self.assertRaises(errors.InterfaceError, self.cnx._handle_result,
                          None)
        self.cnx._socket.sock = tests.DummySocket()
        eof_packet = EOF_PACKET
        eof_packet[3] = 3
        self.cnx._socket.sock.add_packets([COLUMNS_SINGLE, eof_packet])
        exp = {
            'eof': {'status_flag': 0, 'warning_count': 0},
            'columns': [('1', 8, None, None, None, None, 0, 129)]
        }
        res = self.cnx._handle_result(COLUMNS_SINGLE_COUNT)
        self.assertEqual(exp, res)

        self.assertEqual(EOF_PACKET_RESULT,
                         self.cnx._handle_result(EOF_PACKET))
        self.cnx._unread_result = False

        # Handle LOAD DATA INFILE
        self.cnx._socket.sock.reset()
        packet = bytearray(
            b'\x1A\x00\x00\x01\xfb'
            b'\x74\x65\x73\x74\x73\x2f\x64\x61\x74\x61\x2f\x6c\x6f'
            b'\x63\x61\x6c\x5f\x64\x61\x74\x61\x2e\x63\x73\x76')
        self.cnx._socket.sock.add_packet(bytearray(
            b'\x37\x00\x00\x04\x00\x06\x00\x01\x00\x00\x00\x2f\x52'
            b'\x65\x63\x6f\x72\x64\x73\x3a\x20\x36\x20\x20\x44\x65'
            b'\x6c\x65\x74\x65\x64\x3a\x20\x30\x20\x20\x53\x6b\x69'
            b'\x70\x70\x65\x64\x3a\x20\x30\x20\x20\x57\x61\x72\x6e'
            b'\x69\x6e\x67\x73\x3a\x20\x30'))
        exp = {
            'info_msg': 'Records: 6  Deleted: 0  Skipped: 0  Warnings: 0',
            'insert_id': 0, 'field_count': 0, 'warning_count': 0,
            'status_flag': 1, 'affected_rows': 6}
        self.assertEqual(exp, self.cnx._handle_result(packet))

        exp = [
            bytearray(b'\x47\x00\x00\x04\x31\x09\x63\x31\x5f\x31\x09\x63\x32'
                      b'\x5f\x31\x0a\x32\x09\x63\x31\x5f\x32\x09\x63\x32\x5f'
                      b'\x32\x0a\x33\x09\x63\x31\x5f\x33\x09\x63\x32\x5f\x33'
                      b'\x0a\x34\x09\x63\x31\x5f\x34\x09\x63\x32\x5f\x34\x0a'
                      b'\x35\x09\x63\x31\x5f\x35\x09\x63\x32\x5f\x35\x0a\x36'
                      b'\x09\x63\x31\x5f\x36\x09\x63\x32\x5f\x36'),
            bytearray(b'\x00\x00\x00\x05')
        ]
        self.assertEqual(exp, self.cnx._socket.sock._client_sends)

        # Column count is invalid ( more than 4096)
        self.cnx._socket.sock.reset()
        packet = bytearray(b'\x01\x00\x00\x01\xfc\xff\xff\xff')
        self.assertRaises(errors.InterfaceError,
                          self.cnx._handle_result, packet)

        # First byte in first packet is wrong
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x00\x00\x00\x02\x03\x64\x65\x66\x00\x00\x00\x01'
                      b'\x31\x00\x0c\x3f\x00\x01\x00\x00\x00\x08\x81\x00'
                      b'\x00\x00\x00'),
            bytearray(b'\x05\x00\x00\x03\xfe\x00\x00\x00\x00')])

        self.assertRaises(errors.InterfaceError,
                          self.cnx._handle_result, b'\x01\x00\x00\x01\x00')

    def __helper_get_rows_buffer(self, toggle_next_result=False):
        self.cnx._socket.sock.reset()

        packets = [
            bytearray(b'\x07\x00\x00\x04\x06\x4d\x79\x49\x53\x41\x4d'),
            bytearray(b'\x07\x00\x00\x05\x06\x49\x6e\x6e\x6f\x44\x42'),
            bytearray(b'\x0a\x00\x00\x06\x09\x42\x4c'
                      b'\x41\x43\x4b\x48\x4f\x4c\x45'),
            bytearray(b'\x04\x00\x00\x07\x03\x43\x53\x56'),
            bytearray(b'\x07\x00\x00\x08\x06\x4d\x45\x4d\x4f\x52\x59'),
            bytearray(b'\x0a\x00\x00\x09\x09\x46\x45'
                      b'\x44\x45\x52\x41\x54\x45\x44'),
            bytearray(b'\x08\x00\x00\x0a\x07\x41\x52\x43\x48\x49\x56\x45'),
            bytearray(b'\x0b\x00\x00\x0b\x0a\x4d\x52'
                      b'\x47\x5f\x4d\x59\x49\x53\x41\x4d'),
            bytearray(b'\x05\x00\x00\x0c\xfe\x00\x00\x20\x00'),
        ]

        if toggle_next_result:
            packets[-1] = packets[-1][:-2] + b'\x08' + packets[-1][-1:]

        self.cnx._socket.sock.add_packets(packets)
        self.cnx.unread_result = True

    def test_get_rows(self):
        """Get rows from the MySQL resultset"""
        self.cnx._socket.sock = tests.DummySocket()
        self.__helper_get_rows_buffer()
        exp = (
            [(b'MyISAM',), (b'InnoDB',), (b'BLACKHOLE',), (b'CSV',),
             (b'MEMORY',), (b'FEDERATED',), (b'ARCHIVE',), (b'MRG_MYISAM',)],
            {'status_flag': 32, 'warning_count': 0}
        )
        res = self.cnx.get_rows(raw=True)
        self.assertEqual(exp, res)

        self.__helper_get_rows_buffer()
        rows = exp[0]
        i = 0
        while i < len(rows):
            exp = (rows[i:i + 2], None)
            res = self.cnx.get_rows(count=2, raw=True)
            self.assertEqual(exp, res)
            i += 2
        exp = ([], {'status_flag': 32, 'warning_count': 0})
        self.assertEqual(exp, self.cnx.get_rows(raw=True))

        # Test unread results
        self.cnx.unread_result = False
        self.assertRaises(errors.InternalError, self.cnx.get_rows)

        # Test multiple results
        self.cnx._have_next_results = False
        self.__helper_get_rows_buffer(toggle_next_result=True)
        exp = {'status_flag': 8, 'warning_count': 0}
        self.assertEqual(exp, self.cnx.get_rows(raw=True)[-1])
        self.assertTrue(self.cnx._have_next_result)

    def test_get_row(self):
        """Get a row from the MySQL resultset"""
        self.cnx._socket.sock = tests.DummySocket()
        self.__helper_get_rows_buffer()
        expall = (
            [(b'MyISAM',), (b'InnoDB',), (b'BLACKHOLE',), (b'CSV',),
             (b'MEMORY',), (b'FEDERATED',), (b'ARCHIVE',), (b'MRG_MYISAM',)],
            {'status_flag': 32, 'warning_count': 0}
        )

        rows = expall[0]
        for row in rows:
            res = self.cnx.get_row(raw=True)
            exp = (row, None)
            self.assertEqual(exp, res)
        exp = ([], {'status_flag': 32, 'warning_count': 0})
        self.assertEqual(exp, self.cnx.get_rows(raw=True))

    def test_cmd_init_db(self):
        """Send the Init_db-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.cnx._socket.sock.add_packet(OK_PACKET)
        self.assertEqual(OK_PACKET_RESULT, self.cnx.cmd_init_db('test'))

        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(bytearray(
            b'\x2c\x00\x00\x01\xff\x19\x04\x23\x34\x32\x30\x30'
            b'\x30\x55\x6e\x6b\x6e\x6f\x77\x6e\x20\x64\x61\x74'
            b'\x61\x62\x61\x73\x65\x20\x27\x75\x6e\x6b\x6e\x6f'
            b'\x77\x6e\x5f\x64\x61\x74\x61\x62\x61\x73\x65\x27')
        )
        self.assertRaises(errors.ProgrammingError,
                          self.cnx.cmd_init_db, 'unknown_database')

    def test_cmd_query(self):
        """Send a query to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.cnx._socket.sock.add_packet(OK_PACKET)
        res = self.cnx.cmd_query("SET AUTOCOMMIT = OFF")
        self.assertEqual(OK_PACKET_RESULT, res)

        packets = [
            COLUMNS_SINGLE_COUNT,
            COLUMNS_SINGLE,
            bytearray(b'\x05\x00\x00\x03\xfe\x00\x00\x00\x00')
        ]

        # query = "SELECT 1"
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets(packets)
        exp = {
            'eof': {'status_flag': 0, 'warning_count': 0},
            'columns': [('1', 8, None, None, None, None, 0, 129)]
        }
        res = self.cnx.cmd_query("SELECT 1")
        self.assertEqual(exp, res)
        self.assertRaises(errors.InternalError,
                          self.cnx.cmd_query, 'SELECT 2')
        self.cnx.unread_result = False

        # Forge the packets so the multiple result flag is set
        packets[-1] = packets[-1][:-2] + b'\x08' + packets[-1][-1:]
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets(packets)
        self.assertRaises(errors.InterfaceError,
                          self.cnx.cmd_query, "SELECT 1")

    def test_cmd_query_iter(self):
        """Send queries to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.cnx._socket.sock.add_packet(OK_PACKET)
        res = next(self.cnx.cmd_query_iter(
            "SET AUTOCOMMIT = OFF"))
        self.assertEqual(OK_PACKET_RESULT, res)

        packets = [
            COLUMNS_SINGLE_COUNT,
            COLUMNS_SINGLE,
            bytearray(b'\x05\x00\x00\x03\xfe\x00\x00\x08\x00'),
            bytearray(b'\x02\x00\x00\x04\x01\x31'),
            bytearray(b'\x05\x00\x00\x05\xfe\x00\x00\x08\x00'),
            bytearray(b'\x07\x00\x00\x06\x00\x01\x00\x08\x00\x00\x00'),
            bytearray(b'\x01\x00\x00\x07\x01'),
            bytearray(b'\x17\x00\x00\x08\x03\x64\x65\x66\x00\x00\x00\x01'
                      b'\x32\x00\x0c\x3f\x00\x01\x00\x00\x00\x08\x81\x00'
                      b'\x00\x00\x00'),
            bytearray(b'\x05\x00\x00\x09\xfe\x00\x00\x00\x00'),
            bytearray(b'\x02\x00\x00\x0a\x01\x32'),
            bytearray(b'\x05\x00\x00\x0b\xfe\x00\x00\x00\x00'),
        ]
        exp = [
            {'columns': [('1', 8, None, None, None, None, 0, 129)],
             'eof': {'status_flag': 8, 'warning_count': 0}},
            ([(1,)], {'status_flag': 8, 'warning_count': 0}),
            {'affected_rows': 1,
             'field_count': 0,
             'insert_id': 0,
             'status_flag': 8,
             'warning_count': 0},
            {'columns': [('2', 8, None, None, None, None, 0, 129)],
             'eof': {'status_flag': 0, 'warning_count': 0}},
            ([(2,)], {'status_flag': 0, 'warning_count': 0}),
        ]
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets(packets)
        results = []
        stmt = "SELECT 1; SELECT 2".encode('utf-8')
        for result in self.cnx.cmd_query_iter(stmt):
            results.append(result)
            if 'columns' in result:
                results.append(self.cnx.get_rows())
        self.assertEqual(exp, results)

    def test_cmd_refresh(self):
        """Send the Refresh-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.cnx._socket.sock.add_packet(OK_PACKET)
        refresh = constants.RefreshOption.LOG | constants.RefreshOption.THREADS

        self.assertEqual(OK_PACKET_RESULT, self.cnx.cmd_refresh(refresh))

    def test_cmd_quit(self):
        """Send the Quit-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.assertEqual(b'\x01', self.cnx.cmd_quit())

    def test_cmd_statistics(self):
        """Send the Statistics-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        goodpkt = bytearray(
            b'\x88\x00\x00\x01\x55\x70\x74\x69\x6d\x65\x3a\x20'
            b'\x31\x34\x36\x32\x34\x35\x20\x20\x54\x68\x72\x65'
            b'\x61\x64\x73\x3a\x20\x32\x20\x20\x51\x75\x65\x73'
            b'\x74\x69\x6f\x6e\x73\x3a\x20\x33\x36\x33\x35\x20'
            b'\x20\x53\x6c\x6f\x77\x20\x71\x75\x65\x72\x69\x65'
            b'\x73\x3a\x20\x30\x20\x20\x4f\x70\x65\x6e\x73\x3a'
            b'\x20\x33\x39\x32\x20\x20\x46\x6c\x75\x73\x68\x20'
            b'\x74\x61\x62\x6c\x65\x73\x3a\x20\x31\x20\x20\x4f'
            b'\x70\x65\x6e\x20\x74\x61\x62\x6c\x65\x73\x3a\x20'
            b'\x36\x34\x20\x20\x51\x75\x65\x72\x69\x65\x73\x20'
            b'\x70\x65\x72\x20\x73\x65\x63\x6f\x6e\x64\x20\x61'
            b'\x76\x67\x3a\x20\x30\x2e\x32\x34'
        )
        self.cnx._socket.sock.add_packet(goodpkt)
        exp = {
            'Uptime': 146245,
            'Open tables': 64,
            'Queries per second avg': Decimal('0.24'),
            'Slow queries': 0,
            'Threads': 2,
            'Questions': 3635,
            'Flush tables': 1,
            'Opens': 392
        }
        self.assertEqual(exp, self.cnx.cmd_statistics())

        badpkt = bytearray(
            b'\x88\x00\x00\x01\x55\x70\x74\x69\x6d\x65\x3a\x20'
            b'\x31\x34\x36\x32\x34\x35\x20\x54\x68\x72\x65'
            b'\x61\x64\x73\x3a\x20\x32\x20\x20\x51\x75\x65\x73'
            b'\x74\x69\x6f\x6e\x73\x3a\x20\x33\x36\x33\x35\x20'
            b'\x20\x53\x6c\x6f\x77\x20\x71\x75\x65\x72\x69\x65'
            b'\x73\x3a\x20\x30\x20\x20\x4f\x70\x65\x6e\x73\x3a'
            b'\x20\x33\x39\x32\x20\x20\x46\x6c\x75\x73\x68\x20'
            b'\x74\x61\x62\x6c\x65\x73\x3a\x20\x31\x20\x20\x4f'
            b'\x70\x65\x6e\x20\x74\x61\x62\x6c\x65\x73\x3a\x20'
            b'\x36\x34\x20\x20\x51\x75\x65\x72\x69\x65\x73\x20'
            b'\x70\x65\x72\x20\x73\x65\x63\x6f\x6e\x64\x20\x61'
            b'\x76\x67\x3a\x20\x30\x2e\x32\x34'
        )
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(badpkt)
        self.assertRaises(errors.InterfaceError, self.cnx.cmd_statistics)

        badpkt = bytearray(
            b'\x88\x00\x00\x01\x55\x70\x74\x69\x6d\x65\x3a\x20'
            b'\x55\x70\x36\x32\x34\x35\x20\x20\x54\x68\x72\x65'
            b'\x61\x64\x73\x3a\x20\x32\x20\x20\x51\x75\x65\x73'
            b'\x74\x69\x6f\x6e\x73\x3a\x20\x33\x36\x33\x35\x20'
            b'\x20\x53\x6c\x6f\x77\x20\x71\x75\x65\x72\x69\x65'
            b'\x73\x3a\x20\x30\x20\x20\x4f\x70\x65\x6e\x73\x3a'
            b'\x20\x33\x39\x32\x20\x20\x46\x6c\x75\x73\x68\x20'
            b'\x74\x61\x62\x6c\x65\x73\x3a\x20\x31\x20\x20\x4f'
            b'\x70\x65\x6e\x20\x74\x61\x62\x6c\x65\x73\x3a\x20'
            b'\x36\x34\x20\x20\x51\x75\x65\x72\x69\x65\x73\x20'
            b'\x70\x65\x72\x20\x73\x65\x63\x6f\x6e\x64\x20\x61'
            b'\x76\x67\x3a\x20\x30\x2e\x32\x34'
        )
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(badpkt)
        self.assertRaises(errors.InterfaceError, self.cnx.cmd_statistics)

    def test_cmd_process_info(self):
        """Send the Process-Info-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.assertRaises(errors.NotSupportedError, self.cnx.cmd_process_info)

    def test_cmd_process_kill(self):
        """Send the Process-Kill-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.cnx._socket.sock.add_packet(OK_PACKET)
        self.assertEqual(OK_PACKET_RESULT, self.cnx.cmd_process_kill(1))

        pkt = bytearray(
            b'\x1f\x00\x00\x01\xff\x46\x04\x23\x48\x59\x30\x30'
            b'\x30\x55\x6e\x6b\x6e\x6f\x77\x6e\x20\x74\x68\x72'
            b'\x65\x61\x64\x20\x69\x64\x3a\x20\x31\x30\x30'
        )
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(pkt)
        self.assertRaises(errors.DatabaseError,
                          self.cnx.cmd_process_kill, 100)

        pkt = bytearray(
            b'\x29\x00\x00\x01\xff\x47\x04\x23\x48\x59\x30\x30'
            b'\x30\x59\x6f\x75\x20\x61\x72\x65\x20\x6e\x6f\x74'
            b'\x20\x6f\x77\x6e\x65\x72\x20\x6f\x66\x20\x74\x68'
            b'\x72\x65\x61\x64\x20\x31\x36\x30\x35'
        )
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(pkt)
        self.assertRaises(errors.DatabaseError,
                          self.cnx.cmd_process_kill, 1605)

    def test_cmd_debug(self):
        """Send the Debug-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        pkt = bytearray(b'\x05\x00\x00\x01\xfe\x00\x00\x00\x00')
        self.cnx._socket.sock.add_packet(pkt)
        exp = {
            'status_flag': 0,
            'warning_count': 0
        }
        self.assertEqual(exp, self.cnx.cmd_debug())

        pkt = bytearray(
            b'\x47\x00\x00\x01\xff\xcb\x04\x23\x34\x32\x30\x30'
            b'\x30\x41\x63\x63\x65\x73\x73\x20\x64\x65\x6e\x69'
            b'\x65\x64\x3b\x20\x79\x6f\x75\x20\x6e\x65\x65\x64'
            b'\x20\x74\x68\x65\x20\x53\x55\x50\x45\x52\x20\x70'
            b'\x72\x69\x76\x69\x6c\x65\x67\x65\x20\x66\x6f\x72'
            b'\x20\x74\x68\x69\x73\x20\x6f\x70\x65\x72\x61\x74'
            b'\x69\x6f\x6e'
        )
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(pkt)
        self.assertRaises(errors.ProgrammingError, self.cnx.cmd_debug)

    def test_cmd_ping(self):
        """Send the Ping-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.cnx._socket.sock.add_packet(OK_PACKET)
        self.assertEqual(OK_PACKET_RESULT, self.cnx.cmd_ping())

        self.assertRaises(errors.Error, self.cnx.cmd_ping)

    def test_cmd_change_user(self):
        """Send the Change-User-command to MySQL"""
        self.cnx._socket.sock = tests.DummySocket()
        self.cnx._handshake = {
            'protocol': 10,
            'server_version_original': '5.0.30-enterprise-gpl-log',
            'charset': 8,
            'server_threadid': 265,
            'capabilities': 41516,
            'server_status': 2,
            'auth_data': b'h4i6oP!OLng9&PD@WrYH',
            'auth_plugin': 'mysql_native_password',
        }

        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(bytearray(
            b'\x45\x00\x00\x01\xff\x14\x04\x23\x34\x32\x30\x30'
            b'\x30\x41\x63\x63\x65\x73\x73\x20\x64\x65\x6e\x69'
            b'\x65\x64\x20\x66\x6f\x72\x20\x75\x73\x65\x72\x20'
            b'\x27\x68\x61\x6d\x27\x40\x27\x6c\x6f\x63\x61\x6c'
            b'\x68\x6f\x73\x74\x27\x20\x74\x6f\x20\x64\x61\x74'
            b'\x61\x62\x61\x73\x65\x20\x27\x6d\x79\x73\x71\x6c'
            b'\x27'))
        self.assertRaises(errors.ProgrammingError, self.cnx.cmd_change_user,
                          username='ham', password='spam', database='mysql')

    def test__do_handshake(self):
        """Handle the handshake-packet sent by MySQL"""
        config = tests.get_mysql_config()
        config['connection_timeout'] = 1
        cnx = connection.MySQLConnection()
        self.assertEqual(None, cnx._handshake)

        cnx.connect(**config)

        exp = {
            'protocol': int,
            'server_version_original': catch23.STRING_TYPES,
            'charset': int,
            'server_threadid': int,
            'capabilities': catch23.INT_TYPES,
            'server_status': int,
            'auth_data': catch23.BYTE_TYPES,
            'auth_plugin': catch23.STRING_TYPES,
        }

        self.assertEqual(len(exp), len(cnx._handshake))
        for key, type_ in exp.items():
            self.assertTrue(key in cnx._handshake)
            self.assertTrue(
                isinstance(cnx._handshake[key], type_),
                "type check failed for '{0}', expected {1}, we got {2}".format(
                    key, type_, type(cnx._handshake[key])))
        self.assertRaises(errors.OperationalError, cnx._do_handshake)

        class FakeSocket(object):
            def __init__(self, packet):
                self.packet = packet
            def recv(self):
                return self.packet

        correct_handshake =  bytearray(
            b'\x47\x00\x00\x00\x0a\x35\x2e\x30\x2e\x33\x30\x2d'
            b'\x65\x6e\x74\x65\x72\x70\x72\x69\x73\x65\x2d\x67'
            b'\x70\x6c\x2d\x6c\x6f\x67\x00\x09\x01\x00\x00\x68'
            b'\x34\x69\x36\x6f\x50\x21\x4f\x00\x2c\xa2\x08\x02'
            b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
            b'\x00\x00\x4c\x6e\x67\x39\x26\x50\x44\x40\x57\x72'
            b'\x59\x48\x00'
        )

        cnx = connection.MySQLConnection(**config)
        cnx._socket = FakeSocket(correct_handshake)
        exp = {
            'protocol': 10,
            'server_version_original': '5.0.30-enterprise-gpl-log',
            'charset': 8,
            'server_threadid': 265,
            'capabilities': 41516,
            'server_status': 2,
            'auth_data': bytearray(b'h4i6oP!OLng9&PD@WrYH'),
            'auth_plugin': u'mysql_native_password',
        }

        cnx._do_handshake()
        self.assertEqual(exp, cnx._handshake)

        # Handshake with version set to Z.Z.ZZ to simulate bad version
        false_handshake = bytearray(
            b'\x47\x00\x00\x00\x0a\x5a\x2e\x5a\x2e\x5a\x5a\x2d'
            b'\x65\x6e\x74\x65\x72\x70\x72\x69\x73\x65\x2d\x67'
            b'\x70\x6c\x2d\x6c\x6f\x67\x00\x09\x01\x00\x00\x68'
            b'\x34\x69\x36\x6f\x50\x21\x4f\x00\x2c\xa2\x08\x02'
            b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
            b'\x00\x00\x4c\x6e\x67\x39\x26\x50\x44\x40\x57\x72'
            b'\x59\x48\x00'
        )
        cnx._socket = FakeSocket(false_handshake)
        self.assertRaises(errors.InterfaceError, cnx._do_handshake)

        # Handshake with version set to 4.0.23
        unsupported_handshake = bytearray(
            b'\x47\x00\x00\x00\x0a\x34\x2e\x30\x2e\x32\x33\x2d'
            b'\x65\x6e\x74\x65\x72\x70\x72\x69\x73\x65\x2d\x67'
            b'\x70\x6c\x2d\x6c\x6f\x67\x00\x09\x01\x00\x00\x68'
            b'\x34\x69\x36\x6f\x50\x21\x4f\x00\x2c\xa2\x08\x02'
            b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
            b'\x00\x00\x4c\x6e\x67\x39\x26\x50\x44\x40\x57\x72'
            b'\x59\x48\x00'
        )
        cnx._socket = FakeSocket(unsupported_handshake)
        self.assertRaises(errors.InterfaceError, cnx._do_handshake)

    def test__do_auth(self):
        """Authenticate with the MySQL server"""
        self.cnx._socket.sock = tests.DummySocket()
        self.cnx._handshake["auth_plugin"] = "mysql_native_password"
        flags = constants.ClientFlag.get_default()
        kwargs = {
            'username': 'ham',
            'password': 'spam',
            'database': 'test',
            'charset': 45,
            'client_flags': flags,
        }

        self.cnx._socket.sock.add_packet(OK_PACKET)
        self.assertEqual(True, self.cnx._do_auth(**kwargs))

        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(bytearray(b'\x01\x00\x00\x02\xfe'))
        self.assertRaises(errors.NotSupportedError,
                          self.cnx._do_auth, **kwargs)

        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x07\x00\x00\x01\x00\x00\x00\x02\x00\x00\x00'),
            bytearray(b'\x07\x00\x00\x02\x00\x00\x00\x02\x00\x00\x00')
        ])
        self.cnx.set_client_flags([-constants.ClientFlag.CONNECT_WITH_DB])
        self.assertEqual(True, self.cnx._do_auth(**kwargs))

        # Using an unknown database should raise an error
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x07\x00\x00\x01\x00\x00\x00\x02\x00\x00\x00'),
            bytearray(b'\x24\x00\x00\x01\xff\x19\x04\x23\x34\x32\x30\x30'
                      b'\x30\x55\x6e\x6b\x6e\x6f\x77\x6e\x20\x64\x61\x74'
                      b'\x61\x62\x61\x73\x65\x20\x27\x61\x73\x64\x66\x61'
                      b'\x73\x64\x66\x27')
        ])
        flags &= ~constants.ClientFlag.CONNECT_WITH_DB
        kwargs['client_flags'] = flags
        self.assertRaises(errors.ProgrammingError,
                          self.cnx._do_auth, **kwargs)

    @unittest.skipIf(not tests.SSL_AVAILABLE, "Python has no SSL support")
    @unittest.skipIf(tests.MYSQL_VERSION < (8, 0, 3),
                     "caching_sha2_password plugin not supported by server.")
    def test_caching_sha2_password(self):
        """Authenticate with the MySQL server using caching_sha2_password"""
        self.cnx._socket.sock = tests.DummySocket()
        flags = constants.ClientFlag.get_default()
        flags |= constants.ClientFlag.SSL
        kwargs = {
            'username': 'ham',
            'password': 'spam',
            'database': 'test',
            'charset': 45,
            'client_flags': flags,
            'ssl_options': {
                'ca': os.path.join(tests.SSL_DIR, 'tests_CA_cert.pem'),
                'cert': os.path.join(tests.SSL_DIR, 'tests_client_cert.pem'),
                'key': os.path.join(tests.SSL_DIR, 'tests_client_key.pem'),
            },
        }

        self.cnx._handshake['auth_plugin'] = 'caching_sha2_password'
        self.cnx._handshake['auth_data'] = b'h4i6oP!OLng9&PD@WrYH'
        self.cnx._socket.switch_to_ssl = \
            lambda ca, cert, key, verify_cert, verify_identity, cipher, \
                ssl_version: None

        # Test perform_full_authentication
        # Exchange:
        # Client              Server
        # ------              ------
        # make_ssl_auth
        # first_auth
        #                     full_auth
        # second_auth
        #                     OK
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x02\x00\x00\x03\x01\x04'), # full_auth request
            bytearray(b'\x07\x00\x00\x05\x00\x00\x00\x02\x00\x00\x00') # OK
        ])
        self.cnx._do_auth(**kwargs)
        packets = self.cnx._socket.sock._client_sends
        self.assertEqual(3, len(packets))
        ssl_pkt = self.cnx._protocol.make_auth_ssl(
                charset=kwargs['charset'], client_flags=kwargs['client_flags'])
        # Check the SSL request packet
        self.assertEqual(packets[0][4:], ssl_pkt)
        auth_pkt = self.cnx._protocol.make_auth(
            self.cnx._handshake, kwargs['username'],
            kwargs['password'], kwargs['database'],
            charset=kwargs['charset'],
            client_flags=kwargs['client_flags'],
            ssl_enabled=True)
        # Check the first_auth packet
        self.assertEqual(packets[1][4:], auth_pkt)
        # Check the second_auth packet
        self.assertEqual(packets[2][4:],
                bytearray(kwargs["password"].encode('utf-8') + b"\x00"))

        # Test fast_auth_success
        # Exchange:
        # Client              Server
        # ------              ------
        # make_ssl_auth
        # first_auth
        #                     fast_auth
        #                     OK
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x02\x00\x00\x03\x01\x03'), # fast_auth success
            bytearray(b'\x07\x00\x00\x05\x00\x00\x00\x02\x00\x00\x00') # OK
        ])
        self.cnx._do_auth(**kwargs)
        packets = self.cnx._socket.sock._client_sends
        self.assertEqual(2, len(packets))
        ssl_pkt = self.cnx._protocol.make_auth_ssl(
                charset=kwargs['charset'], client_flags=kwargs['client_flags'])
        # Check the SSL request packet
        self.assertEqual(packets[0][4:], ssl_pkt)
        auth_pkt = self.cnx._protocol.make_auth(
            self.cnx._handshake, kwargs['username'],
            kwargs['password'], kwargs['database'],
            charset=kwargs['charset'],
            client_flags=kwargs['client_flags'],
            ssl_enabled=True)
        # Check the first auth packet
        self.assertEqual(packets[1][4:], auth_pkt)

    @unittest.skipIf(not tests.SSL_AVAILABLE, "Python has no SSL support")
    def test__do_auth_ssl(self):
        """Authenticate with the MySQL server using SSL"""
        self.cnx._socket.sock = tests.DummySocket()
        flags = constants.ClientFlag.get_default()
        flags |= constants.ClientFlag.SSL
        kwargs = {
            'username': 'ham',
            'password': 'spam',
            'database': 'test',
            'charset': 45,
            'client_flags': flags,
            'ssl_options': {
                'ca': os.path.join(tests.SSL_DIR, 'tests_CA_cert.pem'),
                'cert': os.path.join(tests.SSL_DIR, 'tests_client_cert.pem'),
                'key': os.path.join(tests.SSL_DIR, 'tests_client_key.pem'),
            },
        }

        self.cnx._handshake['auth_data'] = b'h4i6oP!OLng9&PD@WrYH'

        # We check if do_auth send the autherization for SSL and the
        # normal authorization.
        exp = [
            self.cnx._protocol.make_auth_ssl(
                charset=kwargs['charset'],
                client_flags=kwargs['client_flags']),
            self.cnx._protocol.make_auth(
                self.cnx._handshake, kwargs['username'],
                kwargs['password'], kwargs['database'],
                charset=kwargs['charset'],
                client_flags=kwargs['client_flags'],
                ssl_enabled=True),
        ]
        self.cnx._socket.switch_to_ssl = \
            lambda ca, cert, key, verify_cert, verify_identity, cipher, \
                ssl_version: None
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x07\x00\x00\x01\x00\x00\x00\x02\x00\x00\x00'),
            bytearray(b'\x07\x00\x00\x02\x00\x00\x00\x02\x00\x00\x00')
        ])
        self.cnx._do_auth(**kwargs)
        self.assertEqual(
            exp, [p[4:] for p in self.cnx._socket.sock._client_sends])

    def test_config(self):
        """Configure the MySQL connection

        These tests do not actually connect to MySQL, but they make
        sure everything is setup before calling _open_connection() and
        _post_connection().
        """
        cnx = _DummyMySQLConnection()
        default_config = abstracts.DEFAULT_CONFIGURATION.copy()

        # Should fail because 'dsn' is given
        self.assertRaises(errors.NotSupportedError, cnx.config,
                          **default_config)

        # Remove unsupported arguments
        del default_config['dsn']
        default_config.update({
            'ssl_ca': 'CACert',
            'ssl_cert': 'ServerCert',
            'ssl_key': 'ServerKey',
            'ssl_verify_cert': False,
            'ssl_verify_identity': False
        })
        default_config['converter_class'] = MySQLConverter
        try:
            cnx.config(**default_config)
        except AttributeError as err:
            self.fail("Config does not accept a supported"
                      " argument: {}".format(str(err)))

        # Add an argument which we don't allow
        default_config['spam'] = 'SPAM'
        self.assertRaises(AttributeError, cnx.config, **default_config)

        # We do not support dsn
        self.assertRaises(errors.NotSupportedError, cnx.connect, dsn='ham')

        exp = {
            'host': 'localhost.local',
            'port': 3306,
            'unix_socket': '/tmp/mysql.sock'
        }
        cnx.config(**exp)
        self.assertEqual(exp['port'], cnx._port)
        self.assertEqual(exp['host'], cnx._host)
        self.assertEqual(exp['unix_socket'], cnx._unix_socket)

        exp = (None, 'test', 'mysql  ')
        for database in exp:
            cnx.config(database=database)
            if database is not None:
                database = database.strip()
            failmsg = "Failed setting database to '{0}'".format(database)
            self.assertEqual(database, cnx._database, msg=failmsg)
        cnx.config(user='ham')
        self.assertEqual('ham', cnx._user)

        cnx.config(raise_on_warnings=True)
        self.assertEqual(True, cnx._raise_on_warnings)
        cnx.config(get_warnings=False)
        self.assertEqual(False, cnx._get_warnings)
        cnx.config(connection_timeout=123)
        self.assertEqual(123, cnx._connection_timeout)
        for toggle in [True, False]:
            cnx.config(buffered=toggle)
            self.assertEqual(toggle, cnx._buffered)
            cnx.config(raw=toggle)
            self.assertEqual(toggle, cnx._raw)
        for toggle in [False, True]:
            cnx.config(use_unicode=toggle)
            self.assertEqual(toggle, cnx._use_unicode)

        # Test client flags
        cnx = _DummyMySQLConnection()
        cnx.set_client_flags(constants.ClientFlag.get_default())
        flag = exp = constants.ClientFlag.COMPRESS
        cnx.config(client_flags=flag)
        self.assertEqual(exp, cnx._client_flags)

        # Setting client flags using a list
        cnx = _DummyMySQLConnection()
        cnx.set_client_flags(constants.ClientFlag.get_default())
        flags = [constants.ClientFlag.COMPRESS,
                 constants.ClientFlag.FOUND_ROWS]
        exp = constants.ClientFlag.get_default()
        for flag in flags:
            exp |= flag
        cnx.config(client_flags=flags)
        self.assertEqual(exp, cnx._client_flags)

        # and unsetting client flags again
        exp = constants.ClientFlag.get_default()
        flags = [-constants.ClientFlag.COMPRESS,
                 -constants.ClientFlag.FOUND_ROWS]
        cnx.config(client_flags=flags)
        self.assertEqual(exp, cnx._client_flags)

        # Test compress argument
        cnx.config(compress=True)
        exp = constants.ClientFlag.COMPRESS
        self.assertEqual(exp, cnx._client_flags &
                              constants.ClientFlag.COMPRESS)

        # Test character set
        # utf8mb4 is default, which is mapped to 45
        self.assertEqual(45, cnx._charset_id)
        cnx.config(charset='latin1')
        self.assertEqual(8, cnx._charset_id)
        cnx.config(charset='latin1', collation='latin1_general_ci')
        self.assertEqual(48, cnx._charset_id)
        cnx.config(collation='latin1_general_ci')
        self.assertEqual(48, cnx._charset_id)

        # Test converter class
        class TestConverter(MySQLConverterBase):

            def __init__(self, charset=None, unicode=True):
                pass

        self.cnx.config(converter_class=TestConverter)
        self.assertTrue(isinstance(self.cnx.converter, TestConverter))
        self.assertEqual(self.cnx._converter_class, TestConverter)

        class TestConverterWrong(object):

            def __init__(self, charset, unicode):
                pass

        self.assertRaises(AttributeError,
                          self.cnx.config, converter_class=TestConverterWrong)

        # Test SSL configuration
        exp = {
            'ca': 'CACert',
            'cert': 'ServerCert',
            'key': 'ServerKey',
            'verify_cert': False,
            'verify_identity': False
        }
        cnx.config(ssl_ca=exp['ca'], ssl_cert=exp['cert'], ssl_key=exp['key'])
        self.assertEqual(exp, cnx._ssl)

        exp['verify_cert'] = True

        cnx.config(ssl_ca=exp['ca'], ssl_cert=exp['cert'],
                   ssl_key=exp['key'], ssl_verify_cert=exp['verify_cert'],
                   ssl_verify_identity=exp['verify_identity'])
        self.assertEqual(exp, cnx._ssl)

        # Missing SSL configuration should raise an AttributeError
        cnx._ssl = {}
        cases = [
            {'ssl_key': exp['key']},
            {'ssl_cert': exp['cert']},
        ]
        for case in cases:
            cnx._ssl = {}
            try:
                cnx.config(ssl_ca=exp['ca'], **case)
            except AttributeError as err:
                errmsg = str(err)
                self.assertTrue(list(case.keys())[0] in errmsg)
            else:
                self.fail("Testing SSL attributes failed.")

        # Compatibility tests: MySQLdb
        cnx = _DummyMySQLConnection()
        cnx.connect(db='mysql', passwd='spam', connect_timeout=123)
        self.assertEqual('mysql', cnx._database)
        self.assertEqual('spam', cnx._password)
        self.assertEqual(123, cnx._connection_timeout)

        # Option Files tests
        option_file_dir = os.path.join('tests', 'data', 'option_files')
        cfg = read_option_files(option_files=os.path.join(option_file_dir,
                                                          'my.cnf'))
        cnx.config(**cfg)
        self.assertEqual(cnx._port, 1000)
        self.assertEqual(cnx._unix_socket, '/var/run/mysqld/mysqld.sock')

        cfg = read_option_files(option_files=os.path.join(option_file_dir,
                                                          'my.cnf'), port=2000)
        cnx.config(**cfg)
        self.assertEqual(cnx._port, 2000)

        cfg = read_option_files(option_files=os.path.join(
            option_file_dir, 'my.cnf'), option_groups=[ 'client', 'mysqld'])
        cnx.config(**cfg)

        self.assertEqual(cnx._port, 1001)
        self.assertEqual(cnx._unix_socket, '/var/run/mysqld/mysqld2.sock')
        self.assertEqual(cnx._ssl['ca'], 'dummyCA')
        self.assertEqual(cnx._ssl['cert'], 'dummyCert')
        self.assertEqual(cnx._ssl['key'], 'dummyKey')

    def test__get_connection(self):
        """Get connection based on configuration"""
        if os.name != 'nt':
            res = self.cnx._get_connection()
            self.assertTrue(isinstance(res, network.MySQLUnixSocket))

        self.cnx._unix_socket = None
        self.cnx._connection_timeout = 123
        res = self.cnx._get_connection()
        self.assertTrue(isinstance(res, network.MySQLTCPSocket))
        self.assertEqual(self.cnx._connection_timeout,
                         res._connection_timeout)

    def test__open_connection(self):
        """Open the connection to the MySQL server"""
        # Force TCP Connection
        self.cnx._unix_socket = None
        self.cnx._open_connection()
        self.assertTrue(isinstance(self.cnx._socket,
                                   network.MySQLTCPSocket))
        self.cnx.close()

        self.cnx._client_flags |= constants.ClientFlag.COMPRESS
        self.cnx._open_connection()
        self.assertEqual(self.cnx._socket.recv_compressed,
                         self.cnx._socket.recv)
        self.assertEqual(self.cnx._socket.send_compressed,
                         self.cnx._socket.send)

    def test__post_connection(self):
        """Executes commands after connection has been established"""
        self.cnx._charset_id = 45
        self.cnx._autocommit = True
        self.cnx._time_zone = "-09:00"
        self.cnx._sql_mode = "STRICT_ALL_TABLES"
        self.cnx._post_connection()
        self.assertEqual('utf8mb4', self.cnx.charset)
        self.assertEqual(self.cnx._autocommit, self.cnx.autocommit)
        self.assertEqual(self.cnx._time_zone, self.cnx.time_zone)
        self.assertEqual(self.cnx._sql_mode, self.cnx.sql_mode)

    def test_connect(self):
        """Connect to the MySQL server"""
        config = tests.get_mysql_config()
        config['unix_socket'] = None
        config['connection_timeout'] = 1

        cnx = connection.MySQLConnection()
        config['host'] = tests.get_mysql_config()['host']
        try:
            cnx.connect(**config)
            cnx.close()
        except errors.Error as err:
            self.fail("Failed connecting to '{}': {}".format(
                config['host'], str(err)))

        config['host'] = tests.fake_hostname()
        self.assertRaises(errors.InterfaceError, cnx.connect, **config)

    def test_reconnect(self):
        """Reconnecting to the MySQL Server"""
        supported_arguments = {
            'attempts': 1,
            'delay': 0,
        }
        self.check_args(self.cnx.reconnect, supported_arguments)

        def _test_reconnect_delay():
            config = {
                'unix_socket': None,
                'host': tests.fake_hostname(),
                'connection_timeout': 1,
            }
            cnx = connection.MySQLConnection()
            cnx.config(**config)
            try:
                cnx.reconnect(attempts=2, delay=3)
            except:
                pass

        # Check the delay
        timer = timeit.Timer(_test_reconnect_delay)
        result = timer.timeit(number=1)
        self.assertTrue(result > 3 and result < 12,
                        "3 <= result < 12, was {0}".format(result))

        # Check reconnect stops when successful
        config = tests.get_mysql_config()
        cnx = connection.MySQLConnection()
        cnx.connect(**config)
        conn_id = cnx.connection_id
        cnx.reconnect(attempts=5)
        exp = conn_id + 1
        self.assertGreaterEqual(cnx.connection_id, exp)

    def test_ping(self):
        """Ping the MySQL server"""
        supported_arguments = {
            'reconnect': False,
            'attempts': 1,
            'delay': 0,
        }
        self.check_args(self.cnx.ping, supported_arguments)

        try:
            self.cnx.ping()
        except errors.InterfaceError:
            self.fail("Ping should have not raised an error")

        self.cnx.disconnect()
        self.assertRaises(errors.InterfaceError, self.cnx.ping)

    def test_set_converter_class(self):
        """Set the converter class"""

        class TestConverterWrong(object):
            def __init__(self, charset, unicode):
                pass

        self.assertRaises(TypeError,
                          self.cnx.set_converter_class, TestConverterWrong)

        class TestConverter(MySQLConverterBase):
            def __init__(self, charset, unicode):
                pass

        self.cnx.set_converter_class(TestConverter)
        self.assertTrue(isinstance(self.cnx.converter, TestConverter))
        self.assertEqual(self.cnx._converter_class, TestConverter)

    def test_get_server_version(self):
        """Get the MySQL version"""
        self.assertEqual(self.cnx._server_version,
                         self.cnx.get_server_version())

    def test_get_server_info(self):
        """Get the original MySQL version information"""
        self.assertEqual(self.cnx._handshake['server_version_original'],
                         self.cnx.get_server_info())

        del self.cnx._handshake['server_version_original']
        self.assertEqual(None, self.cnx.get_server_info())

    def test_connection_id(self):
        """MySQL connection ID"""
        self.assertEqual(self.cnx._handshake['server_threadid'],
                         self.cnx.connection_id)

        del self.cnx._handshake['server_threadid']
        self.assertEqual(None, self.cnx.connection_id)

    def test_set_login(self):
        """Set login information for MySQL"""
        exp = ('Ham ', ' Spam ')
        self.cnx.set_login(*exp)
        self.assertEqual(exp[0].strip(), self.cnx._user)
        self.assertEqual(exp[1], self.cnx._password)

        self.cnx.set_login()
        self.assertEqual('', self.cnx._user)
        self.assertEqual('', self.cnx._password)

    def test_unread_results(self):
        """Check and toggle unread result using property"""
        self.cnx.unread_result = True
        self.assertEqual(True, self.cnx._unread_result)
        self.assertEqual(True, self.cnx.unread_result)
        self.cnx.unread_result = False
        self.assertEqual(False, self.cnx._unread_result)
        self.assertEqual(False, self.cnx.unread_result)

        try:
            self.cnx.unread_result = 1
        except ValueError:
            pass  # Expected
        except:
            self.fail("Expected ValueError to be raised")

    def test_get_warnings(self):
        """Check and toggle the get_warnings property"""
        self.cnx.get_warnings = True
        self.assertEqual(True, self.cnx._get_warnings)
        self.assertEqual(True, self.cnx.get_warnings)
        self.cnx.get_warnings = False
        self.assertEqual(False, self.cnx._get_warnings)
        self.assertEqual(False, self.cnx.get_warnings)

        try:
            self.cnx.get_warnings = 1
        except ValueError:
            pass  # Expected
        except:
            self.fail("Expected ValueError to be raised")

    def test_set_charset_collation(self):
        """Set the character set and collation"""
        self.cnx.set_charset_collation('latin1')
        self.assertEqual(8, self.cnx._charset_id)
        self.cnx.set_charset_collation('latin1', 'latin1_general_ci')
        self.assertEqual(48, self.cnx._charset_id)
        self.cnx.set_charset_collation('latin1', None)
        self.assertEqual(8, self.cnx._charset_id)

        self.cnx.set_charset_collation(collation='greek_bin')
        self.assertEqual(70, self.cnx._charset_id)

        self.assertRaises(errors.ProgrammingError,
                          self.cnx.set_charset_collation, 666)
        self.assertRaises(errors.ProgrammingError,
                          self.cnx.set_charset_collation, 'spam')
        self.assertRaises(errors.ProgrammingError,
                          self.cnx.set_charset_collation, 'latin1', 'spam')
        self.assertRaises(errors.ProgrammingError,
                          self.cnx.set_charset_collation, None, 'spam')
        self.assertRaises(ValueError,
                          self.cnx.set_charset_collation, object())

    def test_charset(self):
        """Get character set name"""
        self.cnx.set_charset_collation('latin1', 'latin1_general_ci')
        self.assertEqual('latin1', self.cnx.charset)
        self.cnx._charset_id = 70
        self.assertEqual('greek', self.cnx.charset)
        self.cnx._charset_id = 9
        self.assertEqual('latin2', self.cnx.charset)

        self.cnx._charset_id = 1234567
        try:
            self.cnx.charset
        except errors.ProgrammingError:
            pass  # This is expected
        except:
            self.fail("Expected errors.ProgrammingError to be raised")

    def test_collation(self):
        """Get collation name"""
        exp = 'latin2_general_ci'
        self.cnx.set_charset_collation(collation=exp)
        self.assertEqual(exp, self.cnx.collation)
        self.cnx._charset_id = 70
        self.assertEqual('greek_bin', self.cnx.collation)
        self.cnx._charset_id = 9
        self.assertEqual('latin2_general_ci', self.cnx.collation)

        self.cnx._charset_id = 1234567
        try:
            self.cnx.collation
        except errors.ProgrammingError:
            pass  # This is expected
        except:
            self.fail("Expected errors.ProgrammingError to be raised")

    def test_set_client_flags(self):
        """Set the client flags"""
        self.assertRaises(errors.ProgrammingError,
                          self.cnx.set_client_flags, 'Spam')
        self.assertRaises(errors.ProgrammingError,
                          self.cnx.set_client_flags, 0)

        default_flags = constants.ClientFlag.get_default()

        exp = default_flags
        self.assertEqual(exp, self.cnx.set_client_flags(exp))
        self.assertEqual(exp, self.cnx._client_flags)

        exp = default_flags
        exp |= constants.ClientFlag.SSL
        exp |= constants.ClientFlag.FOUND_ROWS
        exp &= ~constants.ClientFlag.MULTI_RESULTS
        flags = [
            constants.ClientFlag.SSL,
            constants.ClientFlag.FOUND_ROWS,
            -constants.ClientFlag.MULTI_RESULTS
        ]
        self.assertEqual(exp, self.cnx.set_client_flags(flags))
        self.assertEqual(exp, self.cnx._client_flags)

    def test_user(self):
        exp = 'ham'
        self.cnx._user = exp
        self.assertEqual(exp, self.cnx.user)

    def test_host(self):
        exp = 'ham'
        self.cnx._host = exp
        self.assertEqual(exp, self.cnx.server_host)

    def test_port(self):
        exp = 'ham'
        self.cnx._port = exp
        self.assertEqual(exp, self.cnx.server_port)

    def test_unix_socket(self):
        exp = 'ham'
        self.cnx._unix_socket = exp
        self.assertEqual(exp, self.cnx.unix_socket)

    def test_database(self):
        exp = self.cnx.info_query("SELECT DATABASE()")[0]
        self.assertEqual(exp, self.cnx.database)
        exp = 'mysql'
        self.cnx.database = exp
        self.assertEqual(exp, self.cnx.database)

    def test_autocommit(self):
        for exp in [False, True]:
            self.cnx.autocommit = exp
            self.assertEqual(exp, self.cnx.autocommit)

    def test_raise_on_warnings(self):
        """Check and toggle the get_warnings property"""
        self.cnx.raise_on_warnings = True
        self.assertEqual(True, self.cnx._raise_on_warnings)
        self.assertEqual(True, self.cnx.raise_on_warnings)
        self.cnx.raise_on_warnings = False
        self.assertEqual(False, self.cnx._raise_on_warnings)
        self.assertEqual(False, self.cnx.raise_on_warnings)

        try:
            self.cnx.raise_on_warnings = 1
        except ValueError:
            pass  # Expected
        except:
            self.fail("Expected ValueError to be raised")

    def test_cursor(self):
        class FalseCursor(object):
            pass

        class TrueCursor(cursor.CursorBase):

            def __init__(self, cnx=None):
                if PY2:
                    super(TrueCursor, self).__init__()
                else:
                    super().__init__()

        self.assertRaises(errors.ProgrammingError, self.cnx.cursor,
                          cursor_class=FalseCursor)
        self.assertTrue(isinstance(self.cnx.cursor(cursor_class=TrueCursor),
                                   TrueCursor))

        cases = [
            ({}, cursor.MySQLCursor),
            ({'buffered': True}, cursor.MySQLCursorBuffered),
            ({'raw': True}, cursor.MySQLCursorRaw),
            ({'buffered': True, 'raw': True}, cursor.MySQLCursorBufferedRaw),
            ({'prepared': True}, cursor.MySQLCursorPrepared),
            ({'dictionary': True}, cursor.MySQLCursorDict),
            ({'named_tuple': True}, cursor.MySQLCursorNamedTuple),
            ({'dictionary': True, 'buffered': True},
             cursor.MySQLCursorBufferedDict),
            ({'named_tuple': True, 'buffered': True},
             cursor.MySQLCursorBufferedNamedTuple)
        ]
        for kwargs, exp in cases:
            self.assertTrue(isinstance(self.cnx.cursor(**kwargs), exp))

        self.assertRaises(ValueError, self.cnx.cursor, prepared=True,
                          buffered=True)
        self.assertRaises(ValueError, self.cnx.cursor, dictionary=True,
                          raw=True)
        self.assertRaises(ValueError, self.cnx.cursor, named_tuple=True,
                          raw=True)

        # Test when connection is closed
        self.cnx.close()
        self.assertRaises(errors.OperationalError, self.cnx.cursor)

    def test__send_data(self):
        self.assertRaises(ValueError, self.cnx._send_data, 'spam')

        self.cnx._socket.sock = tests.DummySocket()

        data = b"1\tham\t'ham spam'\n2\tfoo\t'foo bar'"

        fp = io.BytesIO(data)
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(OK_PACKET)
        exp = [bytearray(
            b"\x20\x00\x00\x02\x31\x09\x68\x61\x6d\x09\x27\x68\x61\x6d"
            b"\x20\x73\x70\x61\x6d\x27\x0a\x32\x09\x66\x6f\x6f\x09\x27"
            b"\x66\x6f\x6f\x20\x62\x61\x72\x27"
        )]

        self.assertEqual(OK_PACKET, self.cnx._send_data(fp, False))
        self.assertEqual(exp, self.cnx._socket.sock._client_sends)

        fp = io.BytesIO(data)
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packet(OK_PACKET)
        exp.append(bytearray(b'\x00\x00\x00\x03'))
        self.assertEqual(OK_PACKET, self.cnx._send_data(fp, True))
        self.assertEqual(exp, self.cnx._socket.sock._client_sends)

        fp = io.BytesIO(data)
        self.cnx._socket = None
        self.assertRaises(errors.OperationalError,
                          self.cnx._send_data, fp, False)
        # Nothing to read, but try to send empty packet
        self.assertRaises(errors.OperationalError,
                          self.cnx._send_data, fp, True)

    def test__handle_binary_ok(self):
        """Handle a Binary OK packet"""
        packet = bytearray(
            b'\x0c\x00\x00\x01'
            b'\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
        )

        exp = {
            'num_params': 0,
            'statement_id': 1,
            'warning_count': 0,
            'num_columns': 0
        }
        self.assertEqual(exp, self.cnx._handle_binary_ok(packet))

        # Raise an error
        packet = bytearray(
            b'\x2a\x00\x00\x01\xff\x19\x05\x23\x34\x32\x30\x30\x30\x46\x55'
            b'\x4e\x43\x54\x49\x4f\x4e\x20\x74\x65\x73\x74\x2e\x53\x50\x41'
            b'\x4d\x20\x64\x6f\x65\x73\x20\x6e\x6f\x74\x20\x65\x78\x69\x73\x74'
        )
        self.assertRaises(errors.ProgrammingError,
                          self.cnx._handle_binary_ok, packet)

    def test_cmd_stmt_prepare(self):
        """Prepare a MySQL statement"""
        self.cnx._socket.sock = tests.DummySocket()

        stmt = b"SELECT CONCAT(?, ?) AS c1"
        self.cnx._socket.sock.add_packets([
            bytearray(
                b'\x0c\x00\x00\x01\x00\x01\x00\x00\x00\x01'
                b'\x00\x02\x00\x00\x00\x00'),
            bytearray(
                b'\x17\x00\x00\x02\x03\x64\x65\x66\x00\x00\x00\x01\x3f\x00\x0c'
                b'\x3f\x00\x00\x00\x00\x00\xfd\x80\x00\x00\x00\x00'),
            bytearray(
                b'\x17\x00\x00\x03\x03\x64\x65\x66\x00\x00\x00\x01\x3f\x00\x0c'
                b'\x3f\x00\x00\x00\x00\x00\xfd\x80\x00\x00\x00\x00'),
            EOF_PACKET,
            bytearray(
                b'\x18\x00\x00\x05\x03\x64\x65\x66\x00\x00\x00\x02\x63\x31\x00'
                b'\x0c\x3f\x00\x00\x00\x00\x00\xfd\x80\x00\x1f\x00\x00'),
            EOF_PACKET
        ])
        exp = {
            'num_params': 2,
            'statement_id': 1,
            'parameters': [
                ('?', 253, None, None, None, None, 1, 128),
                ('?', 253, None, None, None, None, 1, 128)
            ],
            'warning_count': 0,
            'num_columns': 1,
            'columns': [('c1', 253, None, None, None, None, 1, 128)]
        }
        self.assertEqual(exp, self.cnx.cmd_stmt_prepare(stmt))

        stmt = b"DO 1"
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x0c\x00\x00\x01\x00\x01\x00\x00\x00'
                      b'\x00\x00\x00\x00\x00\x00\x00')
        ])
        exp = {
            'num_params': 0,
            'statement_id': 1,
            'parameters': [],
            'warning_count': 0,
            'num_columns': 0,
            'columns': []
        }
        self.assertEqual(exp, self.cnx.cmd_stmt_prepare(stmt))

        # Raise an error using illegal SPAM() MySQL function
        stmt = b"SELECT SPAM(?) AS c1"
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x2a\x00\x00\x01\xff\x19\x05\x23\x34\x32\x30\x30\x30'
                      b'\x46\x55\x4e\x43\x54\x49\x4f\x4e\x20\x74\x65\x73\x74'
                      b'\x2e\x53\x50\x41\x4d\x20\x64\x6f\x65\x73\x20\x6e\x6f'
                      b'\x74\x20\x65\x78\x69\x73\x74')
        ])
        self.assertRaises(errors.ProgrammingError,
                          self.cnx.cmd_stmt_prepare, stmt)

    def test__handle_binary_result(self):
        self.cnx._socket.sock = tests.DummySocket()

        self.assertRaises(errors.InterfaceError,
                          self.cnx._handle_binary_result, None)
        self.assertRaises(errors.InterfaceError,
                          self.cnx._handle_binary_result,
                          bytearray(b'\x00\x00\x00'))

        self.assertEqual(OK_PACKET_RESULT,
                         self.cnx._handle_binary_result(OK_PACKET))
        self.assertEqual(EOF_PACKET_RESULT,
                         self.cnx._handle_binary_result(EOF_PACKET))

        self.assertRaises(errors.ProgrammingError,
                          self.cnx._handle_binary_result, ERR_PACKET)

        # handle result set
        self.cnx._socket.sock.reset()
        self.cnx._socket.sock.add_packets([
            bytearray(b'\x18\x00\x00\x02\x03\x64\x65\x66\x00\x00\x00\x02\x63'
                      b'\x31\x00\x0c\x21\x00\x09\x00\x00\x00\xfd\x01\x00\x00'
                      b'\x00\x00'),
            EOF_PACKET,
        ])
        exp = (
            1,
            [('c1', 253, None, None, None, None, 0, 1)],
            {'status_flag': 0, 'warning_count': 0}
        )
        self.assertEqual(
            exp, self.cnx._handle_binary_result(
                bytearray(b'\x01\x00\x00\x01\x01'))
        )

    def test_cmd_stmt_execute(self):
        stmt = b"SELECT ? as c1"
        params = (
            1,
            ('ham',),
            [('c1', 253, None, None, None, None, 1, 128)],
            0
        )

        # statement does not exists
        self.assertRaises(errors.DatabaseError, self.cnx.cmd_stmt_execute,
                          *params)

        # prepare and execute
        self.cnx.cmd_stmt_prepare(stmt)
        exp = (
            1,
            [('c1', 253, None, None, None, None, 0, 1)],
            {'status_flag': 0, 'warning_count': 0}
        )
        self.assertEqual(exp, self.cnx.cmd_stmt_execute(*params))

    def test_cmd_stmt_close(self):
        # statement does not exists, does not return or raise anything
        try:
            self.cnx.cmd_stmt_close(99)
        except errors.Error as err:
            self.fail("cmd_stmt_close raised: {0}".format(err))

        # after closing, should not be able to execute
        stmt_info = self.cnx.cmd_stmt_prepare(b"SELECT ? as c1")
        self.cnx.cmd_stmt_close(stmt_info['statement_id'])
        params = (
            stmt_info['statement_id'],
            ('ham',),
            stmt_info['parameters'],
            0
        )
        self.assertRaises(errors.ProgrammingError, self.cnx.cmd_stmt_execute,
                          *params)

    def test_cmd_reset_connection(self):
        """Resets session without re-authenticating"""
        if tests.MYSQL_VERSION < (5, 7, 3):
            self.assertRaises(errors.NotSupportedError,
                              self.cnx.cmd_reset_connection)
        else:
            exp_session_id = self.cnx.connection_id
            self.cnx.cmd_query("SET @ham = 2")
            self.cnx.cmd_reset_connection()

            self.cnx.cmd_query("SELECT @ham")
            self.assertEqual(exp_session_id, self.cnx.connection_id)
            if PY2:
                self.assertNotEqual(('2',), self.cnx.get_rows()[0][0])
            else:
                self.assertNotEqual((b'2',), self.cnx.get_rows()[0][0])

    @unittest.skipIf(os.environ.get("PB2WORKDIR"), "Do not run on PB2")
    @unittest.skipIf(tests.MYSQL_VERSION <= (5, 7, 1), "Shutdown CMD "
                     "not tested with MySQL version 5.6 (BugOra17422299)")
    def test_shutdown(self):
        """Shutting down a connection"""
        config = tests.get_mysql_config()
        cnx = connection.MySQLConnection(**config)
        sql = "SHOW GLOBAL STATUS WHERE variable_name LIKE 'Aborted_clients'"
        cur = cnx.cursor()
        cur.execute(sql)
        aborted_clients = cur.fetchone()[1]

        test_close_cnx = connection.MySQLConnection(**config)
        test_shutdown_cnx = connection.MySQLConnection(**config)

        test_close_cnx.close()
        cur.execute(sql)
        self.assertEqual(aborted_clients, cur.fetchone()[1])

        test_shutdown_cnx.shutdown()
        self.assertRaises(socket.error,
                          test_shutdown_cnx._socket.sock.getsockname)
        cur.execute(sql)
        self.assertEqual(str(int(aborted_clients) + 1), cur.fetchone()[1])

        cur.close()
        cnx.close()

    def test_handle_unread_result(self):
        config = tests.get_mysql_config()
        config['consume_results'] = True
        cnx = connection.MySQLConnection(**config)
        cur = cnx.cursor()
        cur.execute("SELECT 1,2,3")
        cur.execute("SELECT 1,2,3")
        cnx.handle_unread_result()
        self.assertEqual(False, cnx.unread_result)
        cur.close()

        config['consume_results'] = False
        cnx = connection.MySQLConnection(**config)
        cur = cnx.cursor()
        cur.execute("SELECT 1,2,3")
        self.assertRaises(errors.InternalError, cur.execute, "SELECT 1,2,3")
        cnx.consume_results()
        cur.close()
        cnx.close()

    @unittest.skipIf(tests.MYSQL_VERSION < (8, 0, 0),
                     "The local_infile option is disabled only in MySQL 8.0.")
    @tests.foreach_cnx()
    def test_default_allow_local_infile(self):
        cur = self.cnx.cursor()
        cur.execute("DROP TABLE IF EXISTS local_data")
        cur.execute("CREATE TABLE local_data "
                    "(id int, c1 VARCHAR(6), c2 VARCHAR(6))")
        data_file = os.path.join("tests", "data", "local_data.csv")
        cur = self.cnx.cursor()
        sql = "LOAD DATA LOCAL INFILE %s INTO TABLE local_data"
        self.assertRaises(errors.ProgrammingError,
                          cur.execute, sql, (data_file,))
        cur.execute("DROP TABLE IF EXISTS local_data")
        cur.close()

    @tests.foreach_cnx(allow_local_infile=True)
    def test_allow_local_infile(self):
        cur = self.cnx.cursor()
        cur.execute("DROP TABLE IF EXISTS local_data")
        cur.execute("CREATE TABLE local_data "
                    "(id int, c1 VARCHAR(6), c2 VARCHAR(6))")
        data_file = os.path.join("tests", "data", "local_data.csv")
        cur = self.cnx.cursor()
        sql = "LOAD DATA LOCAL INFILE %s INTO TABLE local_data"
        cur.execute(sql, (data_file, ))
        cur.execute("SELECT * FROM local_data")

        exp = [(1, "c1_1", "c2_1"), (2, "c1_2", "c2_2"),
               (3, "c1_3", "c2_3"), (4, "c1_4", "c2_4"),
               (5, "c1_5", "c2_5"), (6, "c1_6", "c2_6")]
        self.assertEqual(exp, cur.fetchall())
        cur.execute("DROP TABLE IF EXISTS local_data")
        cur.close()

    @tests.foreach_cnx()
    def test_connection_attributes_defaults(self):
        """Test default connection attributes"""
        if os.name == "nt":
            if "64" in platform.architecture()[0]:
                platform_arch = "x86_64"
            elif "32" in platform.architecture()[0]:
                platform_arch = "i386"
            else:
                platform_arch = platform.architecture()
            os_ver = "Windows-{}".format(platform.win32_ver()[1])
        else:
            platform_arch = platform.machine()
            if platform.system() == "Darwin":
                os_ver = "{}-{}".format("macOS", platform.mac_ver()[0])
            else:
                os_ver = "-".join(linux_distribution()[0:2])

        license_chunks = LICENSE.split(" ")
        if license_chunks[0] == "GPLv2":
            client_license = "GPL-2.0"
        else:
            client_license = "Commercial"

        cur = self.cnx.cursor()
        # Verify user defined session-connection-attributes are in the server
        cur.execute("SHOW VARIABLES LIKE \"pseudo_thread_id\"")
        pseudo_thread_id = cur.fetchall()[0][1]
        get_attrs = ("SELECT ATTR_NAME, ATTR_VALUE FROM "
                    "performance_schema.session_account_connect_attrs "
                    "where PROCESSLIST_ID = \"{}\"")
        cur.execute(get_attrs.format(pseudo_thread_id))
        rows = cur.fetchall()
        res_dict = dict(rows)
        if CMySQLConnection is not None and \
           isinstance(self.cnx, CMySQLConnection):
            expected_attrs = {
                "_source_host": socket.gethostname(),
                "_connector_name": "mysql-connector-python",
                "_connector_license": client_license,
                "_connector_version": ".".join([str(x) for x in VERSION[0:3]]),
            }
        else:
            expected_attrs = {
                "_pid": str(os.getpid()),
                "_platform": platform_arch,
                "_source_host": socket.gethostname(),
                "_client_name": "mysql-connector-python",
                "_client_license": client_license,
                "_client_version": ".".join([str(x) for x in VERSION[0:3]]),
                "_os": os_ver
            }
        # Note that for an empty string "" value the server stores a Null value
        for attr_name in expected_attrs:
            self.assertEqual(expected_attrs[attr_name], res_dict[attr_name],
                             "Attribute {} with value {} differs of {}"
                             "".format(attr_name, res_dict[attr_name],
                                       expected_attrs[attr_name]))

    @unittest.skipIf(sys.version_info < (2, 7, 9), "The support for SSL is "
                     "not available for Python versions < 2.7.9.")
    @unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 40), "TLSv1.1 incompatible")
    @tests.foreach_cnx()
    def test_get_connection_with_tls_version(self):
        if isinstance(self.cnx, connection.MySQLConnection):
            connector_class = connection.MySQLConnection
        else:
            connector_class = CMySQLConnection

        # Test None value is returned if no schema name is specified
        settings = tests.get_mysql_config()
        if "unix_socket" in settings:
            settings.pop("unix_socket")

        # tls_versions and ssl_disabled=True
        settings["tls_versions"] = ["TLSv1"]
        settings["ssl_disabled"] = True
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)
        self.assertTrue(("along with ssl_disabled" in str(context.exception)),
                        "Unexpected exception message found: {}"
                        "".format(context.exception))

        # tls_versions and ssl_disabled=True
        settings.pop("tls_versions")
        settings["tls_ciphersuites"] = ["DHE-RSA-AES256-SHA"]
        settings["ssl_disabled"] = True
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)
        self.assertTrue(("along with ssl_disabled" in str(context.exception)),
                        "Unexpected exception message found: {}"
                        "".format(context.exception))

        # Empty tls_version list
        settings.pop("ssl_disabled")
        settings["tls_ciphersuites"] = ["DHE-RSA-AES256-SHA"]
        settings["tls_versions"] = []
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)
        self.assertTrue(("At least one" in str(context.exception)),
                        "Unexpected exception message found: {}"
                        "".format(context.exception))

        # Empty tls_ciphersuites list using dict settings
        settings["tls_ciphersuites"] = []
        settings["tls_versions"] = ["TLSv1"]
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)
        self.assertTrue(("No valid cipher suite" in str(context.exception)),
                        "Unexpected exception message found: {}"
                        "".format(context.exception))

        # Empty tls_ciphersuites list without tls-versions
        settings["tls_ciphersuites"] = []
        settings.pop("tls_versions")
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)
        self.assertTrue(("No valid cipher suite" in str(context.exception)),
                        "Unexpected exception message found: {}"
                        "".format(context.exception))

        # Given tls-version not in ["TLSv1.1", "TLSv1.2", "TLSv1.3"]
        settings["tls_ciphersuites"] = ["DHE-RSA-AES256-SHA"]
        settings["tls_versions"] = ["TLSv0.2", "TLSv1.7", "TLSv10.2"]
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)

        # Repeated values in tls_versions on dict settings
        settings["tls_ciphersuites"] = ["DHE-RSA-AES256-SHA"]
        settings["tls_versions"] = ["TLSv1.2", "TLSv1.1", "TLSv1.2"]
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)

        # Empty tls_versions on dict settings
        settings["tls_ciphersuites"] = ["DHE-RSA-AES256-SHA"]
        settings["tls_versions"] = []
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)
        self.assertTrue(("At least one TLS" in str(context.exception)),
                        "Unexpected exception message found: {}"
                        "".format(context.exception))

        # Verify unkown cipher suite case?
        settings["tls_ciphersuites"] = ["NOT-KNOWN"]
        settings["tls_versions"] = ["TLSv1.2"]
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)

        # Verify AttributeError exception is raised With invalid TLS version
        settings["tls_ciphersuites"] = ["DHE-RSA-AES256-SHA"]
        settings["tls_versions"] = ["TLSv8"]
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)
        self.assertTrue(("not recognized" in str(context.exception)),
                        "Unexpected exception message found: {}"
                        "".format(context.exception))

        # Verify unkown cipher suite case?
        settings["tls_ciphersuites"] = ["NOT-KNOWN"]
        settings["tls_versions"] = ["TLSv1.2"]
        with self.assertRaises(AttributeError) as context:
            _ = connector_class(**settings)

        # Verify unsupported TLSv1.3 version is accepted (connection success)
        settings["tls_ciphersuites"] = None
        settings["tls_versions"] = ["TLSv1.3", "TLSv1.2"]
        # Connection must be successfully
        _ = connector_class(**settings)

        err_msg = ("Not using the expected TLS version: {}, instead the "
                   "connection used: {}.")

        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))

        # Following tests requires TLSv1.2
        if tests.MYSQL_VERSION < (8, 0, 17):
            return

        if len(supported_tls) > 1:
            # Verify given TLS version is used
            settings["tls_ciphersuites"] = ["DHE-RSA-AES256-SHA"]
            for test_ver in supported_tls:
                expected_ssl_version = test_ver
                settings["tls_versions"] = [test_ver]
                cnx = connector_class(**settings)
                cur = cnx.cursor()
                cur.execute("SHOW STATUS LIKE 'Ssl_version%'")
                res = cur.fetchall()
    
                self.assertEqual(res[0][1], expected_ssl_version,
                                 err_msg.format(expected_ssl_version, res))

        # Verify the newest TLS version is used from the given list
        exp_res = ["TLSv1.2", "TLSv1.2", "TLSv1.2"]
        test_vers = [["TLSv1", "TLSv1.2", "TLSv1.1"], ["TLSv1.1", "TLSv1.2"],
                     ["TLSv1.2", "TLSv1.1"]]
        for test_ver, exp_ver in zip(test_vers, exp_res):
            settings["tls_versions"] = test_ver
            cnx = connector_class(**settings)
            cur = cnx.cursor()
            res = cur.execute("SHOW STATUS LIKE 'ssl_version%'")
            res = cur.fetchall()
            self.assertEqual(res[0][1], exp_ver, "On test case {}, {}"
                             "".format(test_ver, err_msg.format(exp_ver, res)))

        # Verify given TLS cipher suite is used
        exp_res = ["DHE-RSA-AES128-GCM-SHA256", "DHE-RSA-AES128-GCM-SHA256",
                   "DHE-RSA-AES128-GCM-SHA256"]
        test_ciphers = [["DHE-RSA-AES128-GCM-SHA256"],
                        ["DHE-RSA-AES128-GCM-SHA256"],
                        ["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
            cnx = connector_class(**settings)
            cur = cnx.cursor()
            cur.execute("SHOW STATUS LIKE 'Ssl_cipher%'")
            res = cur.fetchall()
            self.assertEqual(res[0][1], exp_ver, "Unexpected TLS version found:"
                             " {} for: {}".format(res[0][1], test_cipher))

        # 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
        cnx = connector_class(**settings)
        cur = cnx.cursor()
        cur.execute("SHOW STATUS LIKE 'Ssl_cipher%'")
        res = cur.fetchall()
        self.assertIn(res[0][1], exp_res, "Unexpected TLS version found"
                      ": {} not in {}".format(res[0][1], exp_res))

        if "TLSv1.1" in supported_tls:
            # Verify connection success with either TLS given version when.
            # TLSv1.3 is not supported.
            settings["tls_ciphersuites"] = ["DHE-RSA-AES256-SHA"]
            settings["tls_versions"] = ["TLSv1.3", "TLSv1.1"]

            cnx = connector_class(**settings)
            cur = cnx.cursor()
            cur.execute("SHOW STATUS LIKE 'ssl_version%'")
            res = cur.fetchall()
            self.assertNotEqual(res[0][1], 'TLSv1.2', "Unexpected TLS version "
                                "found: {}".format(res[0][1]))

        # Verify error when TLSv1.3 is not supported.
        if not TLS_V1_3_SUPPORTED:
            settings["tls_versions"] = ["TLSv1.3"]
            with self.assertRaises(AttributeError) as context:
                _ = connector_class(**settings)

    def test_connection_attributes_user_defined(self):
        """Tests defined connection attributes"""
        config = tests.get_mysql_config()
        use_pure_values = [True]
        if HAVE_CMYSQL:
            use_pure_values.append(False)
        for use_pure in use_pure_values:
            config['use_pure'] = use_pure

            # Validate an error is raised if user defined connection attributes
            # are invalid
            invalid_conn_attrs = [{1:"1"}, {1:2}, {"_invalid":""}, {"_": ""},
                                  123, 123.45, "text", {"_invalid"}, ['_a1=2',]]

            for invalid_attr in invalid_conn_attrs:
                test_config = copy.deepcopy(config)
                test_config['conn_attrs'] = invalid_attr
                with self.assertRaises(errors.InterfaceError) as _:
                    _ = connect(**test_config)
                    LOGGER.error("InterfaceError not raised while testing "
                                 "invalid attribute: {}".format(invalid_attr))

            # Test error is raised for attribute name starting with '_'
            connection_attributes = [
                {"foo": "bar", "_baz": "zoom"},
                {"_baz": "zoom"},
                {"foo": "bar", "_baz": "zoom", "puuuuum": "kaplot"}
            ]
            for invalid_attr in connection_attributes:
                test_config = copy.deepcopy(config)
                test_config['conn_attrs'] = invalid_attr
                with self.assertRaises(errors.InterfaceError) as _:
                    _ = connect(**test_config)
                    LOGGER.error("InterfaceError not raised while testing "
                                 "invalid attribute: {}".format(invalid_attr))

            # 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 invalid_attr in connection_attributes:
                test_config = copy.deepcopy(config)
                test_config['conn_attrs'] = invalid_attr
                with self.assertRaises(errors.InterfaceError) as context:
                    _ = connect(**test_config)
                    LOGGER.error("InterfaceError not raised while testing "
                                 "invalid attribute: {}".format(invalid_attr))

                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 invalid_attr in connection_attributes:
                test_config = copy.deepcopy(config)
                test_config['conn_attrs'] = invalid_attr
                with self.assertRaises(errors.InterfaceError) as context:
                    _ = connect(**test_config)
                    LOGGER.error("InterfaceError not raised while testing "
                                 "invalid attribute: {}".format(invalid_attr))

            self.assertTrue("exceeds 1024 characters limit size" in
                            context.exception.msg)

            # Validate the user defined attributes are created in the server
            connection_attributes = {
                "foo": "bar",
                "baz": "zoom",
                "quash": "",
                "puuuuum": "kaplot"
            }
            test_config = copy.deepcopy(config)
            test_config['conn_attrs'] = connection_attributes
            cnx = connect(**test_config)
            cur = cnx.cursor()
            cur.execute("SHOW VARIABLES LIKE \"pseudo_thread_id\"")
            pseudo_thread_id = cur.fetchall()[0][1]
            get_attrs = ("SELECT ATTR_NAME, ATTR_VALUE FROM "
                        "performance_schema.session_account_connect_attrs "
                        "where PROCESSLIST_ID = \"{}\"")
            cur.execute(get_attrs.format(pseudo_thread_id))
            rows = cur.fetchall()
            res_dict = dict(rows)
            expected_attrs = copy.deepcopy(connection_attributes)
            # Note: For an empty string "" value the server stores a Null value
            expected_attrs["quash"] = None
            for attr_name in expected_attrs:
                self.assertEqual(expected_attrs[attr_name], res_dict[attr_name],
                                 "Attribute {} with value {} differs of {}"
                                 "".format(attr_name, res_dict[attr_name],
                                           expected_attrs[attr_name]))

    @unittest.skipIf(tests.MYSQL_VERSION < (8, 0, 19),
                     "MySQL 8.0.19+ is required for DNS SRV")
    @unittest.skipIf(not HAVE_DNSPYTHON,
                     "dnspython module is required for DNS SRV")
    def test_dns_srv(self):
        config = tests.get_mysql_config().copy()
        config.pop("unix_socket", None)
        config.pop("port", None)

        # The value of 'dns-srv' must be a boolean
        config["dns_srv"] = "true"
        self.assertRaises(InterfaceError, connect, **config)
        config["dns_srv"] = "false"
        self.assertRaises(InterfaceError, connect, **config)
        config["dns_srv"] = 0
        self.assertRaises(InterfaceError, connect, **config)
        config["dns_srv"] = 1
        self.assertRaises(InterfaceError, connect, **config)
        config["dns_srv"] = True

        # Specifying a port number with DNS SRV lookup is not allowed
        config["port"] = 3306
        self.assertRaises(InterfaceError, connect, **config)
        del config["port"]

        # Using Unix domain sockets with DNS SRV lookup is not allowed
        config["unix_socket"] = "/tmp/mysql.sock"
        self.assertRaises(InterfaceError, connect, **config)
        del config["unix_socket"]


class WL13335(tests.MySQLConnectorTests):
    """WL#13335: Avoid set config values whit flag CAN_HANDLE_EXPIRED_PASSWORDS
    """
    def setUp(self):
        self.config = tests.get_mysql_config()
        cnx = connection.MySQLConnection(**self.config)
        self.user = "expired_pw_user"
        self.passw = "i+QEqGkFr8h5"
        self.hosts = ["localhost", "127.0.0.1"]
        for host in self.hosts:
            cnx.cmd_query("CREATE USER '{}'@'{}' IDENTIFIED BY "
                          "'{}'".format(self.user, host, self.passw))
            cnx.cmd_query("GRANT ALL ON *.* TO '{}'@'{}'"
                          "".format(self.user, host))
            cnx.cmd_query("ALTER USER '{}'@'{}' PASSWORD EXPIRE"
                          "".format(self.user, host))
        cnx.close()

    def tearDown(self):
        cnx = connection.MySQLConnection(**self.config)
        for host in self.hosts:
            cnx.cmd_query("DROP USER '{}'@'{}'".format(self.user, host))
        cnx.close()

    @tests.foreach_cnx()
    def test_connect_with_can_handle_expired_pw_flag(self):
        cnx_config = self.config.copy()
        cnx_config['user'] = self.user
        cnx_config['password'] = self.passw
        flags = constants.ClientFlag.get_default()
        flags |= constants.ClientFlag.CAN_HANDLE_EXPIRED_PASSWORDS
        cnx_config['client_flags'] =flags
        # connection must be successful
        _ = self.cnx.__class__(**cnx_config)


class WL13334(tests.MySQLConnectorTests):
    """WL#13334: Failover and multihost
    """
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_failover_init(self):
        cnx_config = tests.get_mysql_config()
        user = cnx_config['user']
        passw = cnx_config['password']
        host = cnx_config['host']
        port = cnx_config['port']

        # check invalid priority values
        settings_org = {
            "failover": [
                {"host":"127.0.1.2", "port":port, "user": user,
                 "priority":80},
                {"host":"127.0.1.5", "port":port, "user": user,
                 "priority":80},
                {"host":"127.0.1.7", "port":port, "user": user,
                 "priority":10},
                {"host":"127.0.1.10", "port":port, "user": user,
                 "priority":10}],
            "user": "root",
            "passwd": passw,
        }

        # check invalid priority values: negative
        settings = settings_org.copy()
        settings["failover"].append({"host":host, "port":port, "user": user,
                 "priority":-1})
        with self.assertRaises(InterfaceError):
            _ = connect(**settings)

        # check invalid priority values: out of range
        settings = settings_org.copy()
        settings["failover"].append({"host":host, "port":port, "user": user,
                 "priority":101})
        with self.assertRaises(InterfaceError):
            _ = connect(**settings)

        # check invalid priority values: not int type
        settings = settings_org.copy()
        settings["failover"].append({"host":host, "port":port, "user": user,
                 "priority":"A"})
        with self.assertRaises(InterfaceError):
            _ = connect(**settings)

        # check all servers has priority or none error
        settings = settings_org.copy()
        settings["failover"].append({"host":host, "port":port, "user": user})
        with self.assertRaises(InterfaceError):
            _ = connect(**settings)

    def test_failover(self):
        cnx_config = tests.get_mysql_config()
        user = cnx_config['user']
        passw = cnx_config['password']
        host = cnx_config['host']
        port = cnx_config['port']

        settings = {
            "failover": [
                {"host":"127.0.1.2", "port":port, "user": user,
                 "priority":80},
                {"host":"127.0.1.5", "port":port, "user": user,
                 "priority":80},
                {"host":"127.0.1.7", "port":port, "user": user,
                 "priority":10},
                {"host":"127.0.1.10", "port":port, "user": user,
                 "priority":10}],
            "user": "root",
            "passwd": passw,
        }

        # connection must fail with error: "Unable to connect to any of the target hosts"
        with self.assertRaises(InterfaceError) as context:
            _ = connect(**settings)

        self.assertTrue(
            ("Unable to connect to any of the target hosts" in context.exception.msg),
            "Unexpected exception message found: {}".format(context.exception.msg))

        # Verify connection is successful
        settings["failover"].append({"host":host, "port":port, "user": user,
                 "priority":100})
        # connection must be successful
        _ = connect(**cnx_config)
