# Copyright (c) 2009, 2019, 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.protocol
"""

import struct
import datetime
import decimal

import tests
from mysql.connector import (protocol, errors)
from mysql.connector.constants import (ClientFlag, FieldType, FieldFlag)

OK_PACKET = bytearray(b'\x07\x00\x00\x01\x00\x01\x00\x00\x00\x01\x00')
OK_PACKET_RESULT = {
    'insert_id': 0,
    'affected_rows': 1,
    'field_count': 0,
    'warning_count': 1,
    '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'\x01\x00\x00\x00\xfe\x00\x00\x00\x00')
EOF_PACKET_RESULT = {'status_flag': 0, 'warning_count': 0}

SEED = bytearray(
    b'\x66\x5e\x25\x3d\x40\x6c\x7c\x4f\x53\x32'
    b'\x41\x2f\x68\x3e\x3b\x4f\x5a\x56\x23\x46'
)


class MySQLProtocolTests(tests.MySQLConnectorTests):
    def setUp(self):
        self._protocol = protocol.MySQLProtocol()

    def test_make_auth(self):
        """Make a MySQL authentication packet"""
        exp = {
            'allset': bytearray(
                b'\x0d\xa2\x13\x00\x00\x00\x00@!'
                b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
                b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00ham'
                b'\x00\x143\xaa_\x92\\d\x94Z\xb2\xf8;U\x8a\x0c'
                b'\xfb\x8a\xfc\xa7\xd0\xe3test\x00'),
            'nopass': bytearray(
                b'\x0d\xa2\x13\x00\x00\x00\x00@!'
                b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
                b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00ham'
                b'\x00\x00test\x00'),
            'nouser': bytearray(
                b'\x0d\xa2\x13\x00\x00\x00\x00@!'
                b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
                b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
                b'\x143\xaa_\x92\\d\x94Z\xb2\xf8;U\x8a\x0c'
                b'\xfb\x8a\xfc\xa7\xd0\xe3test\x00'),
            'nodb': bytearray(
                b'\x0d\xa2\x13\x00\x00\x00\x00@!'
                b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
                b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00ham'
                b'\x00\x143\xaa_\x92\\d\x94Z\xb2\xf8;U\x8a\x0c'
                b'\xfb\x8a\xfc\xa7\xd0\xe3\x00'),
        }
        flags = ClientFlag.get_default()
        kwargs = {
            'handshake': None,
            'username': 'ham',
            'password': 'spam',
            'database': 'test',
            'charset': 33,
            'client_flags': flags
        }

        self.assertRaises(errors.ProgrammingError,
                          self._protocol.make_auth, **kwargs)

        kwargs['handshake'] = {'auth_data': SEED}
        self.assertRaises(errors.ProgrammingError,
                          self._protocol.make_auth, **kwargs)

        kwargs['handshake'] = {
            'auth_data': SEED,
            'auth_plugin': 'mysql_native_password'
        }
        res = self._protocol.make_auth(**kwargs)
        self.assertEqual(exp['allset'], res)

        kwargs['password'] = None
        res = self._protocol.make_auth(**kwargs)
        self.assertEqual(exp['nopass'], res)

        kwargs['password'] = 'spam'
        kwargs['database'] = None
        res = self._protocol.make_auth(**kwargs)
        self.assertEqual(exp['nodb'], res)

        kwargs['username'] = None
        kwargs['database'] = 'test'
        res = self._protocol.make_auth(**kwargs)
        self.assertEqual(exp['nouser'], res)

    def test_make_auth_ssl(self):
        """Make a SSL authentication packet"""
        cases = [
            ({},
             b'\x00\x00\x00\x00\x00\x00\x00@-\x00\x00\x00\x00\x00\x00'
             b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
             b'\x00\x00\x00'),
            ({'charset': 8},
             b'\x00\x00\x00\x00\x00\x00\x00@\x08\x00\x00\x00\x00\x00\x00'
             b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
             b'\x00\x00\x00'),
            ({'client_flags': 240141},
             b'\r\xaa\x03\x00\x00\x00\x00@-\x00\x00\x00\x00\x00\x00\x00'
             b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
             b'\x00\x00'),
            ({'charset': 8, 'client_flags': 240141,
              'max_allowed_packet': 2147483648},
             b'\r\xaa\x03\x00\x00\x00\x00\x80\x08\x00\x00\x00\x00\x00'
             b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
             b'\x00\x00\x00\x00'),
        ]
        for kwargs, exp in cases:
            self.assertEqual(exp, self._protocol.make_auth_ssl(**kwargs))

    def test_make_command(self):
        """Make a generic MySQL command packet"""
        exp = bytearray(b'\x01\x68\x61\x6d')
        arg = 'ham'.encode('utf-8')
        res = self._protocol.make_command(1, arg)
        self.assertEqual(exp, res)
        res = self._protocol.make_command(1, argument=arg)
        self.assertEqual(exp, res)

        exp = b'\x03'
        res = self._protocol.make_command(3)
        self.assertEqual(exp, res)

    def test_make_changeuser(self):
        """Make a change user MySQL packet"""
        exp = {
            'allset': bytearray(
                b'\x11ham\x00\x143\xaa_'
                b'\x92\\d\x94Z\xb2\xf8;U\x8a\x0c\xfb'
                b'\x8a\xfc\xa7\xd0\xe3test\x00\x08'
                b'\x00'),
            'nopass': bytearray(
                b'\x11\x68\x61\x6d\x00\x00\x74\x65'
                b'\x73\x74\x00\x08\x00'),
        }
        kwargs = {
            'handshake': None,
            'username': 'ham',
            'password': 'spam',
            'database': 'test',
            'charset': 8,
            'client_flags': ClientFlag.get_default()
        }
        self.assertRaises(errors.ProgrammingError,
                          self._protocol.make_change_user, **kwargs)

        kwargs['handshake'] = {'auth_data': SEED}
        self.assertRaises(errors.ProgrammingError,
                          self._protocol.make_change_user, **kwargs)

        kwargs['handshake'] = {
            'auth_data': SEED,
            'auth_plugin': 'mysql_native_password'
        }
        res = self._protocol.make_change_user(**kwargs)
        self.assertEqual(exp['allset'], res)

        kwargs['password'] = None
        res = self._protocol.make_change_user(**kwargs)
        self.assertEqual(exp['nopass'], res)

    def test_parse_handshake(self):
        """Parse handshake-packet sent by MySQL"""
        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'
        )
        exp = {
            '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',
        }

        res = self._protocol.parse_handshake(handshake)
        self.assertEqual(exp, res)

        # Test when end byte \x00 is not present for server 5.5.8
        handshake = handshake[:-1]
        res = self._protocol.parse_handshake(handshake)
        self.assertEqual(exp, res)

    def test_parse_ok(self):
        """Parse OK-packet sent by MySQL"""
        res = self._protocol.parse_ok(OK_PACKET)
        self.assertEqual(OK_PACKET_RESULT, res)

        okpkt = OK_PACKET + b'\x04spam'
        exp = OK_PACKET_RESULT.copy()
        exp['info_msg'] = 'spam'
        res = self._protocol.parse_ok(okpkt)
        self.assertEqual(exp, res)

    def test_parse_column_count(self):
        """Parse the number of columns"""
        packet = bytearray(b'\x01\x00\x00\x01\x03')
        res = self._protocol.parse_column_count(packet)
        self.assertEqual(3, res)

        packet = bytearray(b'\x01\x00')
        self.assertRaises(errors.InterfaceError,
                          self._protocol.parse_column_count, packet)

    def test_parse_column(self):
        """Parse field-packet sent by MySQL"""
        column_packet = bytearray(
            b'\x1a\x00\x00\x02\x03\x64\x65\x66\x00\x00\x00\x04'
            b'\x53\x70\x61\x6d\x00\x0c\x21\x00\x09\x00\x00\x00'
            b'\xfd\x01\x00\x1f\x00\x00')
        exp = ('Spam', 253, None, None, None, None, 0, 1)
        res = self._protocol.parse_column(column_packet)
        self.assertEqual(exp, res)

    def test_parse_eof(self):
        """Parse EOF-packet sent by MySQL"""
        res = self._protocol.parse_eof(EOF_PACKET)
        self.assertEqual(EOF_PACKET_RESULT, res)

    def test_read_text_result(self):
        # Tested by MySQLConnectionTests.test_get_rows() and .test_get_row()
        pass

    def test_parse_binary_prepare_ok(self):
        """Parse Prepare OK packet"""
        cases = [
            # SELECT CONCAT(?, ?) AS c1
            (bytearray(b'\x0c\x00\x00\x01'
                       b'\x00\x01\x00\x00\x00\x01\x00\x02\x00\x00\x00\x00'),
             {'num_params': 2,
              'statement_id': 1,
              'warning_count': 0,
              'num_columns': 1
             }
            ),
            # DO 1
            (bytearray(b'\x0c\x00\x00\x01'
                       b'\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'),
             {'num_params': 0,
              'statement_id': 1,
              'warning_count': 0,
              'num_columns': 0
             }
            ),
        ]
        for packet, exp in cases:
            self.assertEqual(exp,
                             self._protocol.parse_binary_prepare_ok(packet))

    def test__parse_binary_integer(self):
        """Parse an integer from a binary packet"""
        # Case = Expected value; pack format; field type; field flag
        cases = [
            (-128, '<b', FieldType.TINY, 0),
            (-32768, '<h', FieldType.SHORT, 0),
            (-2147483648, '<i', FieldType.LONG, 0),
            (-9999999999, '<q', FieldType.LONGLONG, 0),
            (255, '<B', FieldType.TINY, FieldFlag.UNSIGNED),
            (65535, '<H', FieldType.SHORT, FieldFlag.UNSIGNED),
            (4294967295, '<I', FieldType.LONG, FieldFlag.UNSIGNED),
            (9999999999, '<Q', FieldType.LONGLONG, FieldFlag.UNSIGNED),
        ]
        field_info = [None] * 8
        field_info[0] = 'c1'
        for exp, fmt, field_type, flag in cases:
            field_info[1] = field_type
            field_info[7] = flag
            data = struct.pack(fmt, exp) + b'\x00\x00'
            res = self._protocol._parse_binary_integer(data, field_info)
            self.assertEqual((b'\x00\x00', exp), res,
                             "Failed parsing binary integer '{0}'".format(exp))

    def test__parse_binary_float(self):
        """Parse a float/double from a binary packet"""
        # Case = Expected value; data; field type
        cases = [
            (-3.14159, bytearray(b'\x6e\x86\x1b\xf0\xf9\x21\x09\xc0'),
             FieldType.DOUBLE),
            (3.14159, bytearray(b'\x6e\x86\x1b\xf0\xf9\x21\x09\x40'),
             FieldType.DOUBLE),
            (-3.14, bytearray(b'\xc3\xf5\x48\xc0'), FieldType.FLOAT),
            (3.14, bytearray(b'\xc3\xf5\x48\x40'), FieldType.FLOAT),
        ]
        field_info = [None] * 8
        field_info[0] = 'c1'
        for exp, data, field_type in cases:
            field_info[1] = field_type
            res = self._protocol._parse_binary_float(data + b'\x00\x00',
                                                     field_info)
            self.assertEqual(bytearray(b'\x00\x00'), res[0],
                             "Failed parsing binary float '{0}'".format(exp))
            self.assertAlmostEqual(
                exp, res[1], places=5,
                msg="Failed parsing binary float '{0}'".format(exp))

    def test__parse_binary_timestamp(self):
        """Parse a timestamp from a binary packet"""
        # Case = Expected value; data
        cases = [
            (datetime.date(1977, 6, 14), bytearray(b'\x04\xb9\x07\x06\x0e')),
            (datetime.datetime(1977, 6, 14, 21, 33, 14),
             bytearray(b'\x07\xb9\x07\x06\x0e\x15\x21\x0e')),
            (datetime.datetime(1977, 6, 14, 21, 33, 14, 345),
             bytearray(b'\x0b\xb9\x07\x06\x0e\x15\x21\x0e\x59\x01\x00\x00'))
        ]
        for exp, data in cases:
            res = self._protocol._parse_binary_timestamp(data + b'\x00\x00',
                                                         None)
            self.assertEqual((b'\x00\x00', exp), res,
                             "Failed parsing timestamp '{0}'".format(exp))

    def test__parse_binary_time(self):
        """Parse a time value from a binary packet"""
        cases = [
            (datetime.timedelta(0, 44130),
             bytearray(b'\x08\x00\x00\x00\x00\x00\x0c\x0f\x1e')),
            (datetime.timedelta(14, 15330),
             bytearray(b'\x08\x00\x0e\x00\x00\x00\x04\x0f\x1e')),
            (datetime.timedelta(-14, 15330),
             bytearray(b'\x08\x01\x0e\x00\x00\x00\x04\x0f\x1e')),
            (datetime.timedelta(10, 58530, 230000),
             bytearray(b'\x0c\x00\x0a\x00\x00\x00'
                       b'\x10\x0f\x1e\x70\x82\x03\x00')),
        ]
        for exp, data in cases:
            res = self._protocol._parse_binary_time(data + b'\x00\x00', None)
            self.assertEqual((bytearray(b'\x00\x00'), exp), res,
                             "Failed parsing time '{0}'".format(exp))

    def test__parse_binary_values(self):
        """Parse values from a binary result packet"""
        # The packet in this test is result of the following query:
        #       SELECT 'abc' AS aStr,"
        #        "3.14 AS aFloat,"
        #        "-3.14159 AS aDouble, "
        #        "MAKEDATE(2003, 31) AS aDate, "
        #        "TIMESTAMP('1977-06-14', '21:33:14') AS aDateTime, "
        #        "MAKETIME(256,15,30.23) AS aTime, "
        #        "NULL AS aNull"
        #

        fields = [('aStr', 253, None, None, None, None, 0, 1),
                  ('aFloat', 246, None, None, None, None, 0, 129),
                  ('aDouble', 246, None, None, None, None, 0, 129),
                  ('aDate', 10, None, None, None, None, 1, 128),
                  ('aDateTime', 12, None, None, None, None, 1, 128),
                  ('aTime', 11, None, None, None, None, 1, 128),
                  ('aNull', 6, None, None, None, None, 1, 128)]

        packet = bytearray(b'\x00\x01\x03\x61\x62\x63\x04\x33\x2e\x31\x34\x08'
                           b'\x2d\x33\x2e\x31\x34\x31\x35\x39\x04\xd3\x07'
                           b'\x01\x1f\x07\xb9\x07\x06\x0e\x15\x21\x0e\x0c'
                           b'\x00\x0a\x00\x00\x00\x10\x0f\x1e\x70\x82\x03\x00')

        # float/double are returned as DECIMAL by MySQL
        exp = ('abc',
               '3.14',
               '-3.14159',
               datetime.date(2003, 1, 31),
               datetime.datetime(1977, 6, 14, 21, 33, 14),
               datetime.timedelta(10, 58530, 230000),
               None)
        res = self._protocol._parse_binary_values(fields, packet)
        self.assertEqual(exp, res)

    def test_read_binary_result(self):
        """Read MySQL binary protocol result"""

    def test__prepare_binary_integer(self):
        """Prepare an integer for the MySQL binary protocol"""
        # Case = Data; expected value
        cases = [
            (-128, (struct.pack('<b', -128), FieldType.TINY, 0)),
            (-32768, (struct.pack('<h', -32768), FieldType.SHORT, 0)),
            (-2147483648,
             (struct.pack('<i', -2147483648), FieldType.LONG, 0)),
            (-9999999999,
             (struct.pack('<q', -9999999999), FieldType.LONGLONG, 0)),

            (255, (struct.pack('<B', 255), FieldType.TINY, 128)),
            (65535, (struct.pack('<H', 65535), FieldType.SHORT, 128)),
            (4294967295,
             (struct.pack('<I', 4294967295), FieldType.LONG, 128)),
            (9999999999,
             (struct.pack('<Q', 9999999999), FieldType.LONGLONG, 128)),
        ]
        for data, exp in cases:
            res = self._protocol._prepare_binary_integer(data)
            self.assertEqual(exp, res,
                             "Failed preparing value '{0}'".format(data))

    def test__prepare_binary_timestamp(self):
        """Prepare a timestamp object for the MySQL binary protocol"""
        cases = [
            (datetime.date(1977, 6, 14),
             (bytearray(b'\x04\xb9\x07\x06\x0e'), 10)),
            (datetime.datetime(1977, 6, 14),
             (bytearray(b'\x07\xb9\x07\x06\x0e\x00\x00\x00'), 12)),
            (datetime.datetime(1977, 6, 14, 21, 33, 14),
             (bytearray(b'\x07\xb9\x07\x06\x0e\x15\x21\x0e'), 12)),
            (datetime.datetime(1977, 6, 14, 21, 33, 14, 345),
             (bytearray(b'\x0b\xb9\x07\x06\x0e\x15'
                        b'\x21\x0e\x59\x01\x00\x00'), 12)),
        ]
        for data, exp in cases:
            self.assertEqual(exp,
                             self._protocol._prepare_binary_timestamp(data),
                             "Failed preparing value '{0}'".format(data))

        # Raise an error
        self.assertRaises(ValueError,
                          self._protocol._prepare_binary_timestamp, 'spam')

    def test__prepare_binary_time(self):
        """Prepare a time object for the MySQL binary protocol"""
        cases = [
            (datetime.timedelta(hours=123, minutes=45, seconds=16),
             (bytearray(b'\x08\x00\x05\x00\x00\x00\x03\x2d\x10'), 11)),
            (datetime.timedelta(hours=-123, minutes=45, seconds=16),
             (bytearray(b'\x08\x01\x06\x00\x00\x00\x15\x2d\x10'), 11)),
            (datetime.timedelta(hours=123, minutes=45, seconds=16,
                                microseconds=345),
             (bytearray(b'\x0c\x00\x05\x00\x00\x00\x03'
                        b'\x2d\x10\x59\x01\x00\x00'), 11)),
            (datetime.timedelta(days=123, minutes=45, seconds=16),
             (bytearray(b'\x08\x00\x7b\x00\x00\x00\x00\x2d\x10'), 11)),
            (datetime.time(14, 53, 36),
             (bytearray(b'\x08\x00\x00\x00\x00\x00\x0e\x35\x24'), 11)),
            (datetime.time(14, 53, 36, 345),
             (bytearray(b'\x0c\x00\x00\x00\x00\x00\x0e'
                        b'\x35\x24\x59\x01\x00\x00'), 11))
        ]
        for data, exp in cases:
            self.assertEqual(exp,
                             self._protocol._prepare_binary_time(data),
                             "Failed preparing value '{0}'".format(data))

        # Raise an error
        self.assertRaises(ValueError,
                          self._protocol._prepare_binary_time, 'spam')

    def test_make_stmt_execute(self):
        """Make a MySQL packet with the STMT_EXECUTE command"""
        statement_id = 1
        self.assertRaises(errors.InterfaceError,
                          self._protocol.make_stmt_execute, statement_id,
                          ('ham', 'spam'), (1, 2, 3))

        data = ('ham', 'spam')
        exp = bytearray(
            b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x00\x01\x0f'
            b'\x00\x0f\x00\x03\x68\x61\x6d\x04\x73\x70\x61\x6d'
        )
        res = self._protocol.make_stmt_execute(statement_id, data, (1, 2))
        self.assertEqual(exp, res)

        # Testing types
        cases = [
            ('ham',
             bytearray(b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x00'
                       b'\x01\x0f\x00\x03\x68\x61\x6d')),
            (decimal.Decimal('3.14'),
             bytearray(b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x00'
                       b'\x01\x00\x00\x04\x33\x2e\x31\x34')),
            (255,
             bytearray(b'\x01\x00\x00\x00\x80\x01\x00'
                       b'\x00\x00\x00\x01\x01\x80\xff')),
            (-128,
             bytearray(b'\x01\x00\x00\x00\x00\x01\x00'
                       b'\x00\x00\x00\x01\x01\x00\x80')),
            (datetime.datetime(1977, 6, 14, 21, 20, 30),
             bytearray(b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x00'
                       b'\x01\x0c\x00\x07\xb9\x07\x06\x0e\x15\x14\x1e')),
            (datetime.time(14, 53, 36, 345),
             bytearray(b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x00\x01\x0b\x00'
                       b'\x0c\x00\x00\x00\x00\x00\x0e\x35\x24\x59\x01\x00\x00')),
            (3.14,
             bytearray(b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x00\x01\x05\x00'
                       b'\x1f\x85\xeb\x51\xb8\x1e\x09\x40')),
        ]
        for data, exp in cases:
            res = self._protocol.make_stmt_execute(statement_id, (data,), (1,))
            self.assertEqual(
                exp, res, "Failed preparing statement with '{0}'".format(data))

        # Testing null bitmap
        data = (None, None)
        exp = bytearray(b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x03\x01\x06'
                        b'\x00\x06\x00')
        res = self._protocol.make_stmt_execute(statement_id, data, (1, 2))
        self.assertEqual(exp, res)

        data = (None, 'Ham')
        exp = bytearray(
            b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x01\x01\x06\x00\x0f\x00'
            b'\x03\x48\x61\x6d'
        )
        res = self._protocol.make_stmt_execute(statement_id, data, (1, 2))
        self.assertEqual(exp, res)

        data = ('a',) * 11
        exp = bytearray(
            b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x01'
            b'\x0f\x00\x0f\x00\x0f\x00\x0f\x00\x0f\x00\x0f\x00\x0f\x00'
            b'\x0f\x00\x0f\x00\x0f\x00\x0f\x00\x01\x61\x01\x61\x01\x61'
            b'\x01\x61\x01\x61\x01\x61\x01\x61\x01\x61\x01\x61\x01\x61'
            b'\x01\x61'
        )
        res = self._protocol.make_stmt_execute(statement_id, data, (1,) * 11)
        self.assertEqual(exp, res)

        # Raise an error passing an unsupported object as parameter value
        class UnSupportedObject(object):
            pass

        data = (UnSupportedObject(), UnSupportedObject())
        self.assertRaises(errors.ProgrammingError,
                          self._protocol.make_stmt_execute,
                          statement_id, data, (1, 2))
