pytest-mysql with sqlalchemy client

conftest.py

"""Tests main conftest file."""
import sys
import warnings

import pytest

sys.dont_write_bytecode = True

if not sys.version_info >= (3, 6):
    warnings.simplefilter("error", category=DeprecationWarning)


def pytest_addoption(parser):
    parser.addoption(
        "--run-database", action="store_true", default=False, help="run database tests"
    )


def pytest_configure(config):
    config.addinivalue_line("markers", "database: mark test as needing a real database")


def pytest_collection_modifyitems(config, items):
    if config.getoption("--run-database"):
        # --run-database given in cli: do not skip database tests
        return
    skip_database = pytest.mark.skip(reason="need --run-database option to run")
    for item in items:
        if "database" in item.keywords:
            item.add_marker(skip_database)

test_real_mysql.py

import pytest
from pytest_mysql.executor import MySQLExecutor
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from copy import copy

# https://github.com/ClearcodeHQ/pytest-mysql

from sqlalchemy.engine.url import make_url


def create_database(url: str, character_set: str = "utf8") -> object:
    url_obj = copy(make_url(url))
    database = url_obj.database
    url_obj.database = None
    engine = create_engine(url_obj)
    sql = f"CREATE DATABASE {database} CHARACTER SET = '{character_set}'"
    engine.execute(sql)
    engine.dispose()
    return create_engine(url)


QUERY = """CREATE TABLE pet (name VARCHAR(20), owner VARCHAR(20),
    species VARCHAR(20), sex CHAR(1), birth DATE, death DATE);"""


@pytest.mark.database
def test_proc(mysql_proc: MySQLExecutor):
    """Check first, basic server fixture factory."""
    assert mysql_proc.running()


@pytest.mark.database
def test_mysql(mysql):
    """Check first, basic client fixture factory."""
    cursor = mysql.cursor()
    cursor.execute(QUERY)
    mysql.commit()
    cursor.close()


@pytest.mark.database
def test_sqlalchemy(mysql_proc: MySQLExecutor):
    url = "mysql+pymysql://{username}:{password}@{hostname}:{port}/{schema}".format(
        username="root",
        password="",
        hostname=mysql_proc.host,
        port=mysql_proc.port,
        schema="my_test_schema",
    )

    engine = create_database(url)
    db_session = sessionmaker(bind=engine)
    session = db_session()

    session.execute(QUERY)

    session.close()