Use PyMySQL in MySQL related functional/fullstack tests

mysql-python driver has been replaced by PyMySQL driver[1] in neutron
code but MySQL related functional/fullstack tests try to use
mysql-python driver because of MySQLOpportunisticTestCase[2] and tests
are skipped because mysql-python driver is no more available.

This change provides a backend implementation for mysql+pymysql, a base
base testcase MySQLTestCase[2] using mysql+pymysql implementation
(currently oslo.db provides none of them but will in the future) and
replaces MySQLOpportunisticTestCase with MySQLTestCase.

[1] I73e0fdb6eca70e7d029a40a2f6f17a7c0797a21d
[2] neutron.tests.common.base

Closes-Bug: #1463980
Change-Id: Ic5c1d12ab75443e1cc290a7447eeb4b452b4a9dd

neutron/tests/common/base.py

@@ -14,6 +14,8 @@
 import functools
 import unittest.case
 
+from oslo_db.sqlalchemy import provision
+from oslo_db.sqlalchemy import test_base
 import testtools.testcase
 
 from neutron.common import constants as n_const
@@ -67,3 +69,46 @@ def no_skip_on_missing_deps(wrapped):
                     'is enabled, skip reason: %s' % (wrapped.__name__, e))
             raise
     return wrapper
+
+
+# NOTE(cbrandily): Define mysql+pymysql backend implementation
+@provision.BackendImpl.impl.dispatch_for("mysql+pymysql")
+class PyMySQLBackendImpl(provision.BackendImpl):
+
+    default_engine_kwargs = {'mysql_sql_mode': 'TRADITIONAL'}
+
+    def create_opportunistic_driver_url(self):
+        return "mysql+pymysql://openstack_citest:openstack_citest@localhost/"
+
+    def create_named_database(self, engine, ident, conditional=False):
+        with engine.connect() as conn:
+            if not conditional or not self.database_exists(conn, ident):
+                conn.execute("CREATE DATABASE %s" % ident)
+
+    def drop_named_database(self, engine, ident, conditional=False):
+        with engine.connect() as conn:
+            if not conditional or self.database_exists(conn, ident):
+                conn.execute("DROP DATABASE %s" % ident)
+
+    def database_exists(self, engine, ident):
+        return bool(engine.scalar("SHOW DATABASES LIKE '%s'" % ident))
+
+
+impl = provision.BackendImpl.impl("mysql+pymysql")
+url = impl.create_opportunistic_driver_url()
+# NOTE(cbrandily): Declare mysql+pymysql backend implementation
+provision.Backend("mysql+pymysql", url)
+
+
+# NOTE(cbrandily): Define mysql+pymysql db fixture
+class PyMySQLFixture(test_base.DbFixture):
+    DRIVER = 'mysql+pymysql'
+
+
+# NOTE(cbrandily): Define mysql+pymysql base testcase
+class MySQLTestCase(test_base.DbTestCase):
+    """Base test class for MySQL tests.
+
+    Enforce the supported driver, which is PyMySQL.
+    """
+    FIXTURE = PyMySQLFixture

neutron/tests/fullstack/base.py

@@ -17,9 +17,10 @@ from oslo_db.sqlalchemy import test_base
 
 from neutron.db.migration.models import head  # noqa
 from neutron.db import model_base
+from neutron.tests.common import base
 
 
-class BaseFullStackTestCase(test_base.MySQLOpportunisticTestCase):
+class BaseFullStackTestCase(base.MySQLTestCase):
     """Base test class for full-stack tests."""
 
     def __init__(self, environment, *args, **kwargs):
@@ -42,10 +43,10 @@ class BaseFullStackTestCase(test_base.MySQLOpportunisticTestCase):
     def create_db_tables(self):
         """Populate the new database.
 
-        MySQLOpportunisticTestCase creates a new database for each test, but
-        these need to be populated with the appropriate tables. Before we can
-        do that, we must change the 'connection' option which the Neutron code
-        knows to look at.
+        MySQLTestCase creates a new database for each test, but these need to
+        be populated with the appropriate tables. Before we can do that, we
+        must change the 'connection' option which the Neutron code knows to
+        look at.
 
         Currently, the username and password options are hard-coded by
         oslo.db and neutron/tests/functional/contrib/gate_hook.sh. Also,

neutron/tests/functional/db/test_ipam.py

@@ -26,6 +26,7 @@ from neutron.db import db_base_plugin_v2 as base_plugin
 from neutron.db import model_base
 from neutron.db import models_v2
 from neutron.tests import base
+from neutron.tests.common import base as common_base
 
 
 def get_admin_test_context(db_url):
@@ -205,7 +206,7 @@ class IpamTestCase(object):
             ip_avail_ranges_expected)
 
 
-class TestIpamMySql(test_base.MySQLOpportunisticTestCase, base.BaseTestCase,
+class TestIpamMySql(common_base.MySQLTestCase, base.BaseTestCase,
                     IpamTestCase):
 
     def setUp(self):

neutron/tests/functional/db/test_migrations.py

@@ -30,6 +30,7 @@ import sqlalchemy
 from neutron.db.migration.alembic_migrations import external
 from neutron.db.migration import cli as migration
 from neutron.db.migration.models import head as head_models
+from neutron.tests.common import base
 
 LOG = logging.getLogger(__name__)
 
@@ -208,7 +209,7 @@ class _TestModelsMigrations(test_migrations.ModelsMigrationsSync):
 
 
 class TestModelsMigrationsMysql(_TestModelsMigrations,
-                                test_base.MySQLOpportunisticTestCase):
+                                base.MySQLTestCase):
     pass