Adding MySQL support for db fixtures

.travis.yml

@@ -10,10 +10,16 @@ python:
   - '3.6'
 
 env:
-  - TEST_DATABASE_URL=postgresql://postgres@localhost:5432/pytest_test
+  - DB=pgsql TEST_DATABASE_URL=postgresql://postgres@localhost:5432/pytest_test
+  - DB=mysql TEST_DATABASE_URL=mysql+mysqldb://[email protected]/pytest_test
 
 addons:
   postgresql: '9.6'
+  mysql: '5.7'
+
+before_script:
+  - sh -c "if [ '$DB' = 'postgres' ]; then psql -c 'DROP DATABASE IF EXISTS pytest_test;' -U postgres; fi"
+  - sh -c "if [ '$DB' = 'mysql' ]; then mysql -e 'DROP DATABASE IF EXISTS pytest_test;'; fi"
 
 install:
   - pip install --upgrade pip

pytest_flask_sqlalchemy/fixtures.py

@@ -102,6 +102,9 @@ def _engine(pytestconfig, request, _transaction, mocker):
     # the Engine dialect to reflect tables)
     engine.dialect = connection.dialect
 
+    # Necessary for branching db test code
+    engine.name = connection.engine.name
+
     @contextlib.contextmanager
     def begin():
         '''

setup.py

@@ -20,7 +20,7 @@ def readme():
                       'pytest-mock>=1.6.2',
                       'SQLAlchemy>=1.2.2',
                       'Flask-SQLAlchemy>=2.3'],
-    extras_require={'tests': ['pytest-postgresql']},
+    extras_require={'tests': ['pytest-postgresql', 'mysqlclient']},
     classifiers=[
         'Development Status :: 4 - Beta',
         'Environment :: Plugins',

tests/_conftest.py

@@ -6,6 +6,7 @@
 import sqlalchemy as sa
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
+import MySQLdb
 from pytest_postgresql.factories import (init_postgresql_database,
                                          drop_postgresql_database)
 
@@ -17,26 +18,59 @@
                    'connection string to the environmental variable ' +
                    'TEST_DATABASE_URL in order to run tests.')
 else:
-    DB_OPTS = sa.engine.url.make_url(DB_CONN).translate_connect_args()
+    DB_URL = sa.engine.url.make_url(DB_CONN)
+    DB_OPTS = DB_URL.translate_connect_args()
 
 pytest_plugins = ['pytest-flask-sqlalchemy']
 
 
+def _init_mysql_database(request, db_user, db_host, db_name, db_pass=''):
+
+    mysql_conn = MySQLdb.connect(
+        host=db_host,
+        user=db_user,
+        passwd=db_pass
+    )
+
+    mysql_conn.query("CREATE DATABASE IF NOT EXISTS {name}".format(name=db_name))
+
+    mysql_conn.query("USE {name}".format(name=db_name))
+
+    @request.addfinalizer
+    def drop_database():
+        mysql_conn.query("DROP DATABASE IF EXISTS {name}".format(name=db_name))
+        mysql_conn.close()
+
+    return mysql_conn
+
+
 @pytest.fixture(scope='session')
 def database(request):
     '''
     Create a Postgres database for the tests, and drop it when the tests are done.
     '''
-    pg_host = DB_OPTS.get("host")
-    pg_port = DB_OPTS.get("port")
-    pg_user = DB_OPTS.get("username")
-    pg_db = DB_OPTS["database"]
+    db_host = DB_OPTS.get("host")
+    db_port = DB_OPTS.get("port")
+    db_user = DB_OPTS.get("username")
+    db_name = DB_OPTS["database"]
 
-    init_postgresql_database(pg_user, pg_host, pg_port, pg_db)
+    BACKEND = DB_URL.get_backend_name()
 
-    @request.addfinalizer
-    def drop_database():
-        drop_postgresql_database(pg_user, pg_host, pg_port, pg_db, 9.6)
+    if BACKEND == 'mysql':
+        _init_mysql_database(request, db_user, db_host, db_name)
+
+    elif BACKEND == 'postgresql':
+        init_postgresql_database(db_user, db_host, db_port, db_name)
+
+        @request.addfinalizer
+        def drop_database():
+            drop_postgresql_database(db_user, db_host, db_port, db_name, 9.6)
+
+    else:
+        raise ValueError(
+            'Unsupported database type ({}) requested in '
+            'TEST_DATABASE_URL: {}'.format(BACKEND, DB_URL)
+        )
 
 
 @pytest.fixture(scope='session')

tests/test_fixtures.py

@@ -40,7 +40,6 @@ def test_db_session_changes_dont_persist(person, db_engine, db_session):
             assert not db_session.query(person).first()
     """)
 
-    # Run tests
     result = db_testdir.runpytest()
     result.assert_outcomes(passed=2)
 
@@ -212,33 +211,57 @@ def test_drop_table(db_testdir):
     '''
     Make sure that we can drop tables and verify they do not exist in the context
     of a test.
+
+    NOTE: For MySQL `DROP TABLE ...` statements "cause an implicit commit after
+    executing. The intent is to handle each such statement in its own special
+    transaction because it cannot be rolled back anyway."
+    https://dev.mysql.com/doc/refman/5.7/en/implicit-commit.html
+
+    So this test is skipped for 'mysql' engines
     '''
+
     db_testdir.makepyfile("""
         def test_drop_table(person, db_engine):
+            if db_engine.name == 'mysql':
+                return
 
             # Drop the raw table
             db_engine.execute('''
-                DROP TABLE "person"
+                DROP TABLE person
             ''')
 
             # Check if the raw table exists
-            existing_tables = db_engine.execute('''
-                SELECT relname
-                FROM pg_catalog.pg_class
-                WHERE relkind in ('r', 'm')
-                AND relname = 'person'
-            ''').first()
+            if db_engine.name == 'postgresql':
+                existing_tables = db_engine.execute('''
+                    SELECT relname
+                    FROM pg_catalog.pg_class
+                    WHERE relkind in ('r', 'm')
+                    AND relname = 'person'
+                ''').first()
+
+            else:
+                raise ValueError(
+                    'unsupported database engine type: ' + db_engine.name
+                )
 
             assert not existing_tables
 
         def test_drop_table_changes_dont_persist(person, db_engine):
-
-            existing_tables = db_engine.execute('''
-                SELECT relname
-                FROM pg_catalog.pg_class
-                WHERE relkind in ('r', 'm')
-                AND relname = 'person'
-            ''').first()
+            if db_engine.name == 'mysql':
+                return
+
+            if db_engine.name == 'postgresql':
+                existing_tables = db_engine.execute('''
+                    SELECT relname
+                    FROM pg_catalog.pg_class
+                    WHERE relkind in ('r', 'm')
+                    AND relname = 'person'
+                ''').first()
+
+            else:
+                raise ValueError(
+                    'unsupported database engine type: ' + db_engine.name
+                )
 
             assert existing_tables
     """)