import datetime
from sqlalchemy import bindparam
from sqlalchemy import Column
from sqlalchemy import DateTime
from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import select
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 in_
from sqlalchemy.testing import is_
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import AssertsCompiledSQL
from .test_compiler import ReservedWordFixture
class BackendDialectTest(
ReservedWordFixture, fixtures.TestBase, AssertsCompiledSQL
):
__backend__ = True
__only_on__ = "mysql", "mariadb"
@testing.fixture
def mysql_version_dialect(self, testing_engine):
"""yield a MySQL engine that will simulate a specific version.
patches out various methods to not fail
"""
engine = testing_engine()
_server_version = [None]
with mock.patch.object(
engine.dialect,
"_get_server_version_info",
lambda conn: engine.dialect._parse_server_version(
_server_version[0]
),
), mock.patch.object(
engine.dialect, "_set_mariadb", lambda *arg: None
), mock.patch.object(
engine.dialect,
"get_isolation_level",
lambda *arg: "REPEATABLE READ",
):
def go(server_version):
_server_version[0] = server_version
return engine
yield go
def test_reserved_words_mysql_vs_mariadb(
self, mysql_mariadb_reserved_words
):
"""test #7167 - real backend level
We want to make sure that the "is mariadb" flag as well as the
correct identifier preparer are set up for dialects no matter how they
determine their "is_mariadb" flag.
"""
dialect = testing.db.dialect
expect_mariadb = testing.only_on("mariadb").enabled
table, expected_mysql, expected_mdb = mysql_mariadb_reserved_words
self.assert_compile(
select(table),
expected_mdb if expect_mariadb else expected_mysql,
dialect=dialect,
)
def test_no_show_variables(self):
engine = engines.testing_engine()
def my_execute(self, statement, *args, **kw):
if statement.startswith("SELECT @@"):
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):
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()
@testing.combinations(
"10.5.12-MariaDB", "5.6.49", "5.0.2", argnames="server_version"
)
def test_variable_fetch(self, mysql_version_dialect, server_version):
"""test #7518"""
engine = mysql_version_dialect(server_version)
fetches = []
@event.listens_for(engine, "do_execute_no_params")
@event.listens_for(engine, "do_execute")
def do_execute_no_params(cursor, statement, *arg):
if statement.startswith("SHOW VARIABLES") or statement.startswith(
"SELECT @@"
):
fetches.append(statement)
return None
engine.connect()
if server_version == "5.0.2":
eq_(
fetches,
[
"SHOW VARIABLES LIKE 'sql_mode'",
"SHOW VARIABLES LIKE 'lower_case_table_names'",
],
)
else:
eq_(
fetches,
["SELECT @@sql_mode", "SELECT @@lower_case_table_names"],
)
def test_autocommit_isolation_level(self):
c = testing.db.connect().execution_options(
isolation_level="AUTOCOMMIT"
)
assert c.exec_driver_sql("SELECT @@autocommit;").scalar()
c.rollback()
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):
__backend__ = True
@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),
(4031, "foo", "OperationalError", "mysqldb", True),
(4031, "foo", "OperationalError", "pymysql", True),
(4032, "foo", "OperationalError", "mysqldb", False),
(4032, "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"), 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,
}
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]
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+mysqldb://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+mysqldb://u:p@host/db?foo=true")
)[1]
eq_(kw["foo"], "true")
@testing.only_on(
[
"mysql+mysqldb",
"mysql+pymysql",
"mariadb+mysqldb",
"mariadb+pymysql",
]
)
@testing.combinations(
("utf8mb4",),
("utf8",),
)
def test_special_encodings(self, enc):
eng = engines.testing_engine(
options={"connect_args": {"charset": enc, "use_unicode": 0}}
)
conn = eng.connect()
detected = conn.dialect._connection_charset
if enc == "utf8mb4":
eq_(detected, enc)
else:
in_(detected, ["utf8", "utf8mb3"])
@testing.only_on("mariadb+mariadbconnector")
def test_mariadb_connector_special_encodings(self):
eng = engines.testing_engine()
conn = eng.connect()
detected = conn.dialect._connection_charset
eq_(detected, "utf8mb4")
class ParseVersionTest(fixtures.TestBase):
def test_mariadb_madness(self):
mysql_dialect = make_url("mysql+mysqldb://").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)