Use importlib_metadata; add namespace for mariadb

The ``importlib_metadata`` library is used to scan for setuptools
entrypoints rather than pkg_resources.   as importlib_metadata is a small
library that is included as of Python 3.8, the compatibility library is
installed as a dependency for Python versions older than 3.8.

Unfortunately setuptools "attr:" is broken because it tries to import
the module; seems like this is fixed as part of
https://github.com/pypa/setuptools/pull/1753 however this is too recent
to rely upon for now.

Added a new dialect token "mariadb" that may be used in place of "mysql" in
the :func:`_sa.create_engine` URL.  This will deliver a MariaDB dialect
subclass of the MySQLDialect in use that forces the "is_mariadb" flag to
True.  The dialect will raise an error if a server version string that does
not indicate MariaDB in use is received.   This is useful for
MariaDB-specific testing scenarios as well as to support applications that
are hardcoding to MariaDB-only concepts.  As MariaDB and MySQL featuresets
and usage patterns continue to diverge, this pattern may become more
prominent.

Fixes: #5400
Fixes: #5496
Change-Id: I330815ebe572b6a9818377da56621397335fa702

2 files changed, 91 insertions, 67 deletions

diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py
index 09bdd80be..2053318b6 100644
--- a/test/dialect/mysql/test_compiler.py
+++ b/test/dialect/mysql/test_compiler.py
@@ -150,13 +150,11 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         constraint_name = "constraint"
         constraint = CheckConstraint("data IS NOT NULL", name=constraint_name)
         Table(table_name, m, Column("data", String(255)), constraint)
-        dialect = mysql.dialect()
-        dialect.server_version_info = (10, 1, 1, "MariaDB")
         self.assert_compile(
             schema.DropConstraint(constraint),
             "ALTER TABLE %s DROP CONSTRAINT `%s`"
             % (table_name, constraint_name),
-            dialect=dialect,
+            dialect="mariadb",
         )
 
     def test_create_index_with_length_quoted(self):
@@ -354,8 +352,6 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         self.assert_compile(expr, "concat('x', 'y')", literal_binds=True)
 
     def test_mariadb_for_update(self):
-        dialect = mysql.dialect()
-        dialect.server_version_info = (10, 1, 1, "MariaDB")
 
         table1 = table(
             "mytable", column("myid"), column("name"), column("description")
@@ -366,7 +362,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "SELECT mytable.myid, mytable.name, mytable.description "
             "FROM mytable WHERE mytable.myid = %s "
             "FOR UPDATE",
-            dialect=dialect,
+            dialect="mariadb",
         )
 
         self.assert_compile(
@@ -376,7 +372,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "SELECT mytable.myid, mytable.name, mytable.description "
             "FROM mytable WHERE mytable.myid = %s "
             "FOR UPDATE",
-            dialect=dialect,
+            dialect="mariadb",
         )
 
     def test_delete_extra_froms(self):
diff --git a/test/dialect/mysql/test_dialect.py b/test/dialect/mysql/test_dialect.py
index d968d994c..41a4af639 100644
--- a/test/dialect/mysql/test_dialect.py
+++ b/test/dialect/mysql/test_dialect.py
@@ -5,6 +5,7 @@ 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
@@ -12,18 +13,88 @@ 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_execute
 
 
-class DialectTest(fixtures.TestBase):
+class BackendDialectTest(fixtures.TestBase):
     __backend__ = True
     __only_on__ = "mysql"
 
+    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),
@@ -176,74 +247,31 @@ class DialectTest(fixtures.TestBase):
             conn = eng.connect()
             eq_(conn.dialect._connection_charset, enc)
 
-    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)
+class ParseVersionTest(fixtures.TestBase):
+    def test_mariadb_madness(self):
+        mysql_dialect = make_url("mysql://").get_dialect()()
 
-        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()
+        is_(mysql_dialect.is_mariadb, False)
 
-    def test_no_default_isolation_level(self):
-        from sqlalchemy.testing import mock
+        mysql_dialect = make_url("mysql+pymysql://").get_dialect()()
+        is_(mysql_dialect.is_mariadb, False)
 
-        engine = engines.testing_engine()
+        mariadb_dialect = make_url("mariadb://").get_dialect()()
 
-        real_isolation_level = testing.db.dialect.get_isolation_level
+        is_(mariadb_dialect.is_mariadb, True)
 
-        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)
+        mariadb_dialect = make_url("mariadb+pymysql://").get_dialect()()
 
-        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()
+        is_(mariadb_dialect.is_mariadb, True)
 
-    def test_autocommit_isolation_level(self):
-        c = testing.db.connect().execution_options(
-            isolation_level="AUTOCOMMIT"
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "MySQL version 5.7.20 is not a MariaDB variant.",
+            mariadb_dialect._parse_server_version,
+            "5.7.20",
         )
-        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 ParseVersionTest(fixtures.TestBase):
     @testing.combinations(
         ((10, 2, 7), "10.2.7-MariaDB", (10, 2, 7, "MariaDB"), True),
         (
@@ -286,7 +314,7 @@ class ParseVersionTest(fixtures.TestBase):
         (True, (10, 2, 6, "MariaDB", 10, 2, "6+maria~stretch", "log")),
     )
     def test_mariadb_check_warning(self, expect_, version):
-        dialect = mysql.dialect()
+        dialect = mysql.dialect(is_mariadb="MariaDB" in version)
         dialect.server_version_info = version
         if expect_:
             with expect_warnings(