diff --git a/.travis.yml b/.travis.yml index af3fed2..31e434e 100644 --- a/.travis.yml +++ b/.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://root@127.0.0.1/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 diff --git a/pytest_flask_sqlalchemy/fixtures.py b/pytest_flask_sqlalchemy/fixtures.py index c38951b..45c045d 100644 --- a/pytest_flask_sqlalchemy/fixtures.py +++ b/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(): ''' diff --git a/setup.py b/setup.py index aff550c..4c9c084 100644 --- a/setup.py +++ b/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', diff --git a/tests/_conftest.py b/tests/_conftest.py index e6b84f9..2f86380 100644 --- a/tests/_conftest.py +++ b/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') diff --git a/tests/test_fixtures.py b/tests/test_fixtures.py index d1e953f..61447ca 100644 --- a/tests/test_fixtures.py +++ b/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 """)