# -*- coding: utf-8 -*-

import copy
import platform
import sys
if sys.version_info < (2, 7):
    import unittest2 as unittest
else:
    import unittest

from decimal import Decimal

from pymysqlreplication.tests import base
from pymysqlreplication.constants.BINLOG import *
from pymysqlreplication.row_event import *
from pymysqlreplication.event import *

__all__ = ["TestDataType"]


class TestDataType(base.PyMySQLReplicationTestCase):
    def ignoredEvents(self):
        return [GtidEvent]

    def create_and_insert_value(self, create_query, insert_query):
        self.execute(create_query)
        self.execute(insert_query)
        self.execute("COMMIT")

        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
        #QueryEvent for the Create Table
        self.assertIsInstance(self.stream.fetchone(), QueryEvent)

        #QueryEvent for the BEGIN
        self.assertIsInstance(self.stream.fetchone(), QueryEvent)

        self.assertIsInstance(self.stream.fetchone(), TableMapEvent)

        event = self.stream.fetchone()
        if self.isMySQL56AndMore():
            self.assertEqual(event.event_type, WRITE_ROWS_EVENT_V2)
        else:
            self.assertEqual(event.event_type, WRITE_ROWS_EVENT_V1)
        self.assertIsInstance(event, WriteRowsEvent)
        return event

    def test_decimal(self):
        create_query = "CREATE TABLE test (test DECIMAL(2,1))"
        insert_query = "INSERT INTO test VALUES(4.2)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.columns[0].precision, 2)
        self.assertEqual(event.columns[0].decimals, 1)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("4.2"))

    def test_decimal_long_values(self):
        create_query = "CREATE TABLE test (\
            test DECIMAL(20,10) \
        )"
        insert_query = "INSERT INTO test VALUES(42000.123456)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("42000.123456"))

    def test_decimal_long_values_1(self):
        create_query = "CREATE TABLE test (\
            test DECIMAL(20,10) \
        )"
        insert_query = "INSERT INTO test VALUES(9000000123.123456)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("9000000123.123456"))

    def test_decimal_long_values_2(self):
        create_query = "CREATE TABLE test (\
            test DECIMAL(20,10) \
        )"
        insert_query = "INSERT INTO test VALUES(9000000123.0000012345)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"],
                Decimal("9000000123.0000012345"))

    def test_decimal_negative_values(self):
        create_query = "CREATE TABLE test (\
            test DECIMAL(20,10) \
        )"
        insert_query = "INSERT INTO test VALUES(-42000.123456)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("-42000.123456"))

    def test_decimal_two_values(self):
        create_query = "CREATE TABLE test (\
            test DECIMAL(2,1), \
            test2 DECIMAL(20,10) \
        )"
        insert_query = "INSERT INTO test VALUES(4.2, 42000.123456)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("4.2"))
        self.assertEqual(event.rows[0]["values"]["test2"], Decimal("42000.123456"))

    def test_decimal_with_zero_scale_1(self):
        create_query = "CREATE TABLE test (test DECIMAL(23,0))"
        insert_query = "INSERT INTO test VALUES(10)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("10"))

    def test_decimal_with_zero_scale_2(self):
        create_query = "CREATE TABLE test (test DECIMAL(23,0))"
        insert_query = "INSERT INTO test VALUES(12345678912345678912345)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("12345678912345678912345"))

    def test_decimal_with_zero_scale_3(self):
        create_query = "CREATE TABLE test (test DECIMAL(23,0))"
        insert_query = "INSERT INTO test VALUES(100000.0)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("100000"))

    def test_decimal_with_zero_scale_4(self):
        create_query = "CREATE TABLE test (test DECIMAL(23,0))"
        insert_query = "INSERT INTO test VALUES(-100000.0)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("-100000"))

    def test_decimal_with_zero_scale_6(self):
        create_query = "CREATE TABLE test (test DECIMAL(23,0))"
        insert_query = "INSERT INTO test VALUES(-1234567891234567891234)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], Decimal("-1234567891234567891234"))

    def test_tiny(self):
        create_query = "CREATE TABLE test (id TINYINT UNSIGNED NOT NULL, test TINYINT)"
        insert_query = "INSERT INTO test VALUES(255, -128)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["id"], 255)
        self.assertEqual(event.rows[0]["values"]["test"], -128)

    def test_tiny_maps_to_boolean_true(self):
        create_query = "CREATE TABLE test (id TINYINT UNSIGNED NOT NULL, test BOOLEAN)"
        insert_query = "INSERT INTO test VALUES(1, TRUE)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["id"], 1)
        self.assertEqual(type(event.rows[0]["values"]["test"]), type(1))
        self.assertEqual(event.rows[0]["values"]["test"], 1)

    def test_tiny_maps_to_boolean_false(self):
        create_query = "CREATE TABLE test (id TINYINT UNSIGNED NOT NULL, test BOOLEAN)"
        insert_query = "INSERT INTO test VALUES(1, FALSE)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["id"], 1)
        self.assertEqual(type(event.rows[0]["values"]["test"]), type(0))
        self.assertEqual(event.rows[0]["values"]["test"], 0)

    def test_tiny_maps_to_none(self):
        create_query = "CREATE TABLE test (id TINYINT UNSIGNED NOT NULL, test BOOLEAN)"
        insert_query = "INSERT INTO test VALUES(1, NULL)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["id"], 1)
        self.assertEqual(type(event.rows[0]["values"]["test"]), type(None))
        self.assertEqual(event.rows[0]["values"]["test"], None)

    def test_short(self):
        create_query = "CREATE TABLE test (id SMALLINT UNSIGNED NOT NULL, test SMALLINT)"
        insert_query = "INSERT INTO test VALUES(65535, -32768)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["id"], 65535)
        self.assertEqual(event.rows[0]["values"]["test"], -32768)

    def test_long(self):
        create_query = "CREATE TABLE test (id INT UNSIGNED NOT NULL, test INT)"
        insert_query = "INSERT INTO test VALUES(4294967295, -2147483648)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["id"], 4294967295)
        self.assertEqual(event.rows[0]["values"]["test"], -2147483648)

    def test_float(self):
        create_query = "CREATE TABLE test (id FLOAT NOT NULL, test FLOAT)"
        insert_query = "INSERT INTO test VALUES(42.42, -84.84)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(round(event.rows[0]["values"]["id"], 2), 42.42)
        self.assertEqual(round(event.rows[0]["values"]["test"],2 ), -84.84)

    def test_double(self):
        create_query = "CREATE TABLE test (id DOUBLE NOT NULL, test DOUBLE)"
        insert_query = "INSERT INTO test VALUES(42.42, -84.84)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(round(event.rows[0]["values"]["id"], 2), 42.42)
        self.assertEqual(round(event.rows[0]["values"]["test"],2 ), -84.84)

    def test_timestamp(self):
        create_query = "CREATE TABLE test (test TIMESTAMP);"
        insert_query = "INSERT INTO test VALUES('1984-12-03 12:33:07')"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], datetime.datetime(1984, 12, 3, 12, 33, 7))

    def test_timestamp_mysql56(self):
        if not self.isMySQL56AndMore():
            self.skipTest("Not supported in this version of MySQL")
        self.set_sql_mode()
        create_query = '''CREATE TABLE test (test0 TIMESTAMP(0),
            test1 TIMESTAMP(1),
            test2 TIMESTAMP(2),
            test3 TIMESTAMP(3),
            test4 TIMESTAMP(4),
            test5 TIMESTAMP(5),
            test6 TIMESTAMP(6));'''
        insert_query = '''INSERT INTO test VALUES('1984-12-03 12:33:07',
            '1984-12-03 12:33:07.1',
            '1984-12-03 12:33:07.12',
            '1984-12-03 12:33:07.123',
            '1984-12-03 12:33:07.1234',
            '1984-12-03 12:33:07.12345',
            '1984-12-03 12:33:07.123456')'''
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test0"], datetime.datetime(1984, 12, 3, 12, 33, 7))
        self.assertEqual(event.rows[0]["values"]["test1"], datetime.datetime(1984, 12, 3, 12, 33, 7, 1))
        self.assertEqual(event.rows[0]["values"]["test2"], datetime.datetime(1984, 12, 3, 12, 33, 7, 12))
        self.assertEqual(event.rows[0]["values"]["test3"], datetime.datetime(1984, 12, 3, 12, 33, 7, 123))
        self.assertEqual(event.rows[0]["values"]["test4"], datetime.datetime(1984, 12, 3, 12, 33, 7, 1234))
        self.assertEqual(event.rows[0]["values"]["test5"], datetime.datetime(1984, 12, 3, 12, 33, 7, 12345))
        self.assertEqual(event.rows[0]["values"]["test6"], datetime.datetime(1984, 12, 3, 12, 33, 7, 123456))

    def test_longlong(self):
        create_query = "CREATE TABLE test (id BIGINT UNSIGNED NOT NULL, test BIGINT)"
        insert_query = "INSERT INTO test VALUES(18446744073709551615, -9223372036854775808)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["id"], 18446744073709551615)
        self.assertEqual(event.rows[0]["values"]["test"], -9223372036854775808)

    def test_int24(self):
        create_query = "CREATE TABLE test (id MEDIUMINT UNSIGNED NOT NULL, test MEDIUMINT, test2 MEDIUMINT, test3 MEDIUMINT, test4 MEDIUMINT, test5 MEDIUMINT)"
        insert_query = "INSERT INTO test VALUES(16777215, 8388607, -8388608, 8, -8, 0)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["id"], 16777215)
        self.assertEqual(event.rows[0]["values"]["test"], 8388607)
        self.assertEqual(event.rows[0]["values"]["test2"], -8388608)
        self.assertEqual(event.rows[0]["values"]["test3"], 8)
        self.assertEqual(event.rows[0]["values"]["test4"], -8)
        self.assertEqual(event.rows[0]["values"]["test5"], 0)

    def test_date(self):
        create_query = "CREATE TABLE test (test DATE);"
        insert_query = "INSERT INTO test VALUES('1984-12-03')"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], datetime.date(1984, 12, 3))

    def test_zero_date(self):
        create_query = "CREATE TABLE test (id INTEGER, test DATE, test2 DATE);"
        insert_query = "INSERT INTO test (id, test2) VALUES(1, '0000-01-21')"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], None)
        self.assertEqual(event.rows[0]["values"]["test2"], None)

    def test_zero_month(self):
        self.set_sql_mode()
        create_query = "CREATE TABLE test (id INTEGER, test DATE, test2 DATE);"
        insert_query = "INSERT INTO test (id, test2) VALUES(1, '2015-00-21')"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], None)
        self.assertEqual(event.rows[0]["values"]["test2"], None)

    def test_zero_day(self):
        self.set_sql_mode()
        create_query = "CREATE TABLE test (id INTEGER, test DATE, test2 DATE);"
        insert_query = "INSERT INTO test (id, test2) VALUES(1, '2015-05-00')"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], None)
        self.assertEqual(event.rows[0]["values"]["test2"], None)

    def test_time(self):
        create_query = "CREATE TABLE test (test1 TIME, test2 TIME);"
        insert_query = "INSERT INTO test VALUES('838:59:59', '-838:59:59')"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test1"], datetime.timedelta(
            microseconds=(((838*60) + 59)*60 + 59)*1000000
        ))
        self.assertEqual(event.rows[0]["values"]["test2"], datetime.timedelta(
            microseconds=(((-838*60) + 59)*60 + 59)*1000000
        ))

    def test_time2(self):
        if not self.isMySQL56AndMore():
            self.skipTest("Not supported in this version of MySQL")
        create_query = "CREATE TABLE test (test1 TIME(6), test2 TIME(6));"
        insert_query = """
            INSERT INTO test VALUES('838:59:59.000000', '-838:59:59.000000');
        """
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test1"], datetime.timedelta(
            microseconds=(((838*60) + 59)*60 + 59)*1000000 + 0
        ))
        self.assertEqual(event.rows[0]["values"]["test2"], datetime.timedelta(
            microseconds=(((-838*60) + 59)*60 + 59)*1000000 + 0
        ))

    def test_zero_time(self):
        create_query = "CREATE TABLE test (id INTEGER, test TIME NOT NULL DEFAULT 0);"
        insert_query = "INSERT INTO test (id) VALUES(1)"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], datetime.timedelta(seconds=0))

    def test_datetime(self):
        create_query = "CREATE TABLE test (test DATETIME);"
        insert_query = "INSERT INTO test VALUES('1984-12-03 12:33:07')"
        event = self.create_and_insert_value(create_query, insert_query)
        self.assertEqual(event.rows[0]["values"]["test"], datetime.datetime(1984, 12, 3, 12, 33, 7))

    def test_zero_datetime(self):
        self.set_sql_mode()
        create_query = "CREATE TABLE test (id INTEGER, test DATETIME NOT NULL DEFAULT 0);"
        insert_query = "INSERT INTO test (id) VALUES(1)"