# coding: utf-8

import datetime

from sqlalchemy import bindparam
from sqlalchemy import Column
from sqlalchemy import DateTime
from sqlalchemy import exc
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy.dialects import mysql
from sqlalchemy.engine.url import make_url
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import mock
from ...engine import test_deprecations


class BackendDialectTest(fixtures.TestBase):
    __backend__ = True
    __only_on__ = "mysql", "mariadb"

    def test_no_show_variables(self):
        from sqlalchemy.testing import mock

        engine = engines.testing_engine()

        def my_execute(self, statement, *args, **kw):
            if statement.startswith("SHOW VARIABLES"):
                statement = "SELECT 1 FROM DUAL WHERE 1=0"
            return real_exec(self, statement, *args, **kw)

        real_exec = engine._connection_cls.exec_driver_sql
        with mock.patch.object(
            engine._connection_cls, "exec_driver_sql", my_execute
        ):
            with expect_warnings(
                "Could not retrieve SQL_MODE; please ensure the "
                "MySQL user has permissions to SHOW VARIABLES"
            ):
                engine.connect()

    def test_no_default_isolation_level(self):
        from sqlalchemy.testing import mock

        engine = engines.testing_engine()

        real_isolation_level = testing.db.dialect.get_isolation_level

        def fake_isolation_level(connection):
            connection = mock.Mock(
                cursor=mock.Mock(
                    return_value=mock.Mock(
                        fetchone=mock.Mock(return_value=None)
                    )
                )
            )
            return real_isolation_level(connection)

        with mock.patch.object(
            engine.dialect, "get_isolation_level", fake_isolation_level
        ):
            with expect_warnings(
                "Could not retrieve transaction isolation level for MySQL "
                "connection."
            ):
                engine.connect()

    def test_autocommit_isolation_level(self):
        c = testing.db.connect().execution_options(
            isolation_level="AUTOCOMMIT"
        )
        assert c.exec_driver_sql("SELECT @@autocommit;").scalar()

        c = c.execution_options(isolation_level="READ COMMITTED")
        assert not c.exec_driver_sql("SELECT @@autocommit;").scalar()

    def test_isolation_level(self):
        values = [
            "READ UNCOMMITTED",
            "READ COMMITTED",
            "REPEATABLE READ",
            "SERIALIZABLE",
        ]
        for value in values:
            c = testing.db.connect().execution_options(isolation_level=value)
            eq_(testing.db.dialect.get_isolation_level(c.connection), value)


class DialectTest(fixtures.TestBase):
    @testing.combinations(
        (None, "cONnection was kILLEd", "InternalError", "pymysql", True),
        (None, "cONnection aLREady closed", "InternalError", "pymysql", True),
        (None, "something broke", "InternalError", "pymysql", False),
        (1927, "Connection was killed", "OperationalError", "pymysql", True),
        (1927, "Connection was killed", "OperationalError", "mysqldb", True),
        (2006, "foo", "OperationalError", "mysqldb", True),
        (2006, "foo", "OperationalError", "pymysql", True),
        (2007, "foo", "OperationalError", "mysqldb", False),
        (2007, "foo", "OperationalError", "pymysql", False),
    )
    def test_is_disconnect(
        self, arg0, message, exc_cls_name, dialect_name, is_disconnect
    ):
        class Error(Exception):
            pass

        dbapi = mock.Mock()
        dbapi.Error = Error
        dbapi.ProgrammingError = type("ProgrammingError", (Error,), {})
        dbapi.OperationalError = type("OperationalError", (Error,), {})
        dbapi.InterfaceError = type("InterfaceError", (Error,), {})
        dbapi.InternalError = type("InternalError", (Error,), {})

        dialect = getattr(mysql, dialect_name).dialect(dbapi=dbapi)

        error = getattr(dbapi, exc_cls_name)(arg0, message)
        eq_(dialect.is_disconnect(error, None, None), is_disconnect)

    @testing.combinations(
        ("mysqldb"),
        ("pymysql"),
        ("oursql"),
        id_="s",
        argnames="driver_name",
    )
    def test_ssl_arguments(self, driver_name):
        url = (
            "mysql+%s://scott:tiger@localhost:3306/test"
            "?ssl_ca=/ca.pem&ssl_cert=/cert.pem&ssl_key=/key.pem" % driver_name
        )
        url_obj = make_url(url)
        dialect = url_obj.get_dialect()()

        expected = {
            "{}".format(
                "password" if driver_name == "pymysql" else "passwd"
            ): "tiger",
            "{}".format(
                "database" if driver_name == "pymysql" else "db"
            ): "test",
            "ssl": {"ca": "/ca.pem", "cert": "/cert.pem", "key": "/key.pem"},
            "host": "localhost",
            "user": "scott",
            "port": 3306,
        }
        # add check_hostname check for mysqldb and pymysql
        if driver_name in ["mysqldb", "pymysql"]:
            url = url + "&ssl_check_hostname=false"
            expected["ssl"]["check_hostname"] = False

        kwarg = dialect.create_connect_args(make_url(url))[1]
        # args that differ between oursql and others
        for k in ("use_unicode", "found_rows", "client_flag"):
            kwarg.pop(k, None)
        eq_(kwarg, expected)

    @testing.combinations(
        ("compress", True),
        ("connect_timeout", 30),
        ("read_timeout", 30),
        ("write_timeout", 30),
        ("client_flag", 1234),
        ("local_infile", 1234),
        ("use_unicode", False),
        ("charset", "hello"),
    )
    def test_normal_arguments_mysqldb(self, kwarg, value):
        from sqlalchemy.dialects.mysql import mysqldb

        dialect = mysqldb.dialect()
        connect_args = dialect.create_connect_args(
            make_url(
                "mysql://scott:tiger@localhost:3306/test"
                "?%s=%s" % (kwarg, value)
            )
        )

        eq_(connect_args[1][kwarg], value)

    def test_mysqlconnector_buffered_arg(self):
        from sqlalchemy.dialects.mysql import mysqlconnector

        dialect = mysqlconnector.dialect()
        kw = dialect.create_connect_args(
            make_url("mysql+mysqlconnector://u:p@host/db?buffered=true")
        )[1]
        eq_(kw["buffered"], True)

        kw = dialect.create_connect_args(
            make_url("mysql+mysqlconnector://u:p@host/db?buffered=false")
        )[1]
        eq_(kw["buffered"], False)

        kw = dialect.create_connect_args(
            make_url("mysql+mysqlconnector://u:p@host/db")
        )[1]
        eq_(kw["buffered"], True)

    def test_mysqlconnector_raise_on_warnings_arg(self):
        from sqlalchemy.dialects.mysql import mysqlconnector

        dialect = mysqlconnector.dialect()
        kw = dialect.create_connect_args(
            make_url(
                "mysql+mysqlconnector://u:p@host/db?raise_on_warnings=true"
            )
        )[1]
        eq_(kw["raise_on_warnings"], True)

        kw = dialect.create_connect_args(
            make_url(
                "mysql+mysqlconnector://u:p@host/db?raise_on_warnings=false"
            )
        )[1]
        eq_(kw["raise_on_warnings"], False)

        kw = dialect.create_connect_args(
            make_url("mysql+mysqlconnector://u:p@host/db")
        )[1]
        assert "raise_on_warnings" not in kw

    @testing.only_on(
        [
            "mysql+mysqldb",
            "mysql+pymysql",
            "mariadb+mysqldb",
            "mariadb+pymysql",
        ]
    )
    def test_random_arg(self):
        dialect = testing.db.dialect
        kw = dialect.create_connect_args(
            make_url("mysql://u:p@host/db?foo=true")
        )[1]
        eq_(kw["foo"], "true")

    @testing.only_on(
        [
            "mysql+mysqldb",
            "mysql+pymysql",
            "mariadb+mysqldb",
            "mariadb+pymysql",
        ]
    )
    def test_special_encodings(self):

        for enc in ["utf8mb4", "utf8"]:
            eng = engines.testing_engine(
                options={"connect_args": {"charset": enc, "use_unicode": 0}}
            )
            conn = eng.connect()
            eq_(conn.dialect._connection_charset, enc)


class ParseVersionTest(fixtures.TestBase):
    def test_mariadb_madness(self):
        mysql_dialect = make_url("mysql://").get_dialect()()

        is_(mysql_dialect.is_mariadb, False)

        mysql_dialect = make_url("mysql+pymysql://").get_dialect()()
        is_(mysql_dialect.is_mariadb, False)

        mariadb_dialect = make_url("mariadb://").get_dialect()()

        is_(mariadb_dialect.is_mariadb, True)

        mariadb_dialect = make_url("mariadb+pymysql://").get_dialect()()

        is_(mariadb_dialect.is_mariadb, True)

        assert_raises_message(
            exc.InvalidRequestError,
            "MySQL version 5.7.20 is not a MariaDB variant.",
            mariadb_dialect._parse_server_version,
            "5.7.20",
        )

    def test_502_minimum(self):
        dialect = mysql.dialect()
        assert_raises_message(
            NotImplementedError,
            "the MySQL/MariaDB dialect supports server "
            "version info 5.0.2 and above.",
            dialect._parse_server_version,
            "5.0.1",
        )

    @testing.combinations(
        ((10, 2, 7), "10.2.7-MariaDB", (10, 2, 7), True),
        (
            (10, 2, 7),
            "5.6.15.10.2.7-MariaDB",
            (5, 6, 15, 10, 2, 7),
            True,
        ),
        ((5, 0, 51, 24), "5.0.51a.24+lenny5", (5, 0, 51, 24), False),
        ((10, 2, 10), "10.2.10-MariaDB", (10, 2, 10), True),
        ((5, 7, 20), "5.7.20", (5, 7, 20), False),
        ((5, 6, 15), "5.6.15", (5, 6, 15), False),
        (
            (10, 2, 6),
            "10.2.6.MariaDB.10.2.6+maria~stretch-log",
            (10, 2, 6, 10, 2, 6),
            True,
        ),
        (
            (10, 1, 9),
            "10.1.9-MariaDBV1.0R050D002-20170809-1522",
            (10, 1, 9, 20170809, 1522),
            True,
        ),
    )
    def test_mariadb_normalized_version(
        self, expected, raw_version, version, is_mariadb
    ):
        dialect = mysql.dialect()
        eq_(dialect._parse_server_version(raw_version), version)
        dialect.server_version_info = version
        eq_(dialect._mariadb_normalized_version_info, expected)
        assert dialect._is_mariadb is is_mariadb

    @testing.combinations(
        (True, "10.2.7-MariaDB"),
        (True, "5.6.15-10.2.7-MariaDB"),
        (False, "10.2.10-MariaDB"),
        (False, "5.7.20"),
        (False, "5.6.15"),
        (True, "10.2.6-MariaDB-10.2.6+maria~stretch.log"),
    )
    def test_mariadb_check_warning(self, expect_, version):
        dialect = mysql.dialect(is_mariadb="MariaDB" in version)
        dialect._parse_server_version(version)
        if expect_:
            with expect_warnings(
                ".*before 10.2.9 has known issues regarding "
                "CHECK constraints"
            ):
                dialect._warn_for_known_db_issues()
        else:
            dialect._warn_for_known_db_issues()


class RemoveUTCTimestampTest(fixtures.TablesTest):
    """This test exists because we removed the MySQL dialect's
    override of the UTC_TIMESTAMP() function, where the commit message
    for this feature stated that "it caused problems with executemany()".
    Since no example was provided, we are trying lots of combinations
    here.

    [ticket:3966]

    """

    __only_on__ = "mysql", "mariadb"
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        Table(
            "t",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("x", Integer),
            Column("data", DateTime),
        )

        Table(
            "t_default",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("x", Integer),
            Column("idata", DateTime, default=func.utc_timestamp()),
            Column("udata", DateTime, onupdate=func.utc_timestamp()),
        )

    def test_insert_executemany(self, connection):
        conn = connection
        conn.execute(
            self.tables.t.insert().values(data=func.utc_timestamp()),
            [{"x": 5}, {"x": 6}, {"x": 7}],
        )

    def test_update_executemany(self, connection):
        conn = connection
        timestamp = datetime.datetime(2015, 4, 17, 18, 5, 2)
        conn.execute(
            self.tables.t.insert(),
            [
                {"x": 5, "data": timestamp},
                {"x": 6, "data": timestamp},
                {"x": 7, "data": timestamp},
            ],
        )

        conn.execute(
            self.tables.t.update()
            .values(data=func.utc_timestamp())
            .where(self.tables.t.c.x == bindparam("xval")),
            [{"xval": 5}, {"xval": 6}, {"xval": 7}],
        )

    def test_insert_executemany_w_default(self, connection):
        conn = connection
        conn.execute(
            self.tables.t_default.insert(), [{"x": 5}, {"x": 6}, {"x": 7}]
        )

    def test_update_executemany_w_default(self, connection):
        conn = connection
        timestamp = datetime.datetime(2015, 4, 17, 18, 5, 2)
        conn.execute(
            self.tables.t_default.insert(),
            [
                {"x": 5, "idata": timestamp},
                {"x": 6, "idata": timestamp},
                {"x": 7, "idata": timestamp},
            ],
        )

        conn.execute(
            self.tables.t_default.update()
            .values(idata=func.utc_timestamp())
            .where(self.tables.t_default.c.x == bindparam("xval")),
            [{"xval": 5}, {"xval": 6}, {"xval": 7}],
        )


class SQLModeDetectionTest(fixtures.TestBase):
    __only_on__ = "mysql", "mariadb"
    __backend__ = True

    def _options(self, modes):
        def connect(con, record):
            cursor = con.cursor()
            cursor.execute("set sql_mode='%s'" % (",".join(modes)))

        e = engines.testing_engine(
            options={
                "pool_events": [
                    (connect, "first_connect"),
                    (connect, "connect"),
                ]
            }
        )
        return e

    def test_backslash_escapes(self):
        engine = self._options(["NO_BACKSLASH_ESCAPES"])
        c = engine.connect()
        assert not engine.dialect._backslash_escapes
        c.close()
        engine.dispose()

        engine = self._options([])
        c = engine.connect()
        assert engine.dialect._backslash_escapes
        c.close()
        engine.dispose()

    def test_ansi_quotes(self):
        engine = self._options(["ANSI_QUOTES"])
        c = engine.connect()
        assert engine.dialect._server_ansiquotes
        c.close()
        engine.dispose()

    def test_combination(self):
        engine = self._options(["ANSI_QUOTES,NO_BACKSLASH_ESCAPES"])
        c = engine.connect()
        assert engine.dialect._server_ansiquotes
        assert not engine.dialect._backslash_escapes
        c.close()
        engine.dispose()


class ExecutionTest(fixtures.TestBase):
    """Various MySQL execution special cases."""

    __only_on__ = "mysql", "mariadb"
    __backend__ = True

    def test_charset_caching(self):
        engine = engines.testing_engine()

        cx = engine.connect()
        meta = MetaData()
        charset = engine.dialect._detect_charset(cx)

        meta.reflect(cx)
        eq_(cx.dialect._connection_charset, charset)
        cx.close()

    def test_sysdate(self, connection):
        d = connection.execute(func.sysdate()).scalar()
        assert isinstance(d, datetime.datetime)


class AutocommitTextTest(
    test_deprecations.AutocommitKeywordFixture, fixtures.TestBase
):
    __only_on__ = "mysql", "mariadb"

    def test_load_data(self):
        self._test_keyword("LOAD DATA STUFF")

    def test_replace(self):
        self._test_keyword("REPLACE THING")