'Transaction rollback and dropping database in pytest fixtures

I'm writing functional tests to test API endpoints with pytest. I want to achieve:

  • that at the end of each test DB transaction rolls back so that each test starts with the same data.
  • at the end of the test session, I want to delete the testing DB.

Bellow is my conftest.py file, and what is currently happening is that transactions aren't rolled back, and when it comes to the end of the session, it keeps hanging at drop_database on teardown, while it doesn't hang at the beginning when checking if DB exists (marked also in comments). What do I need to do to make those rollbacks? And is there anything else I need to close, in order for that drop_database to take place? Thank you!

import pytest
import sqlalchemy
from sqlalchemy_utils import database_exists, create_database, drop_database
from alembic.command import upgrade
from alembic.config import Config
from flask import Flask
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy
from api.views import dtm_api, diagrams_api, labels_api
from .data.dummy_data import import_dummy



ALEMBIC_CONFIG = 'migrations/alembic.ini'


@pytest.fixture(scope='session')
def app(request):
    _app = Flask(__name__)
    _app.config.from_object('api.config.TestingConfig')

    _app.register_blueprint(diagrams_api.bp)
    _app.register_blueprint(dtm_api.bp)
    _app.register_blueprint(labels_api.bp)

    ctx = _app.app_context()
    ctx.push()

    def teardown():
        ctx.pop()
    request.addfinalizer(teardown)

    yield _app


@pytest.fixture(scope='session')
def test_client(app, request):
    yield app.test_client()


@pytest.yield_fixture(scope='session')
def db(app, request):
    engine = sqlalchemy.create_engine(app.config.get("SQLALCHEMY_DATABASE_URI"))

    if database_exists(engine.url):
        drop_database(engine.url) #here it doesn't hang
    create_database(engine.url)

    _db = SQLAlchemy()
    _db.init_app(app)
    _db.app = app

    #Make migrations and add dummy data
    Migrate(app, _db)
    config = Config(ALEMBIC_CONFIG)
    config.set_main_option("script_location", "migrations")
    with app.app_context():
        upgrade(config, 'head')
    import_dummy(_db)

    def teardown():
        drop_database(engine.url) #here it hangs
        engine.dispose()

    request.addfinalizer(teardown)

    yield _db


@pytest.fixture(scope='function', autouse=True)
def session(db, request):
    connection = db.engine.connect()
    options = dict(bind=connection, binds={}, autoflush=False, autocommit=False)
    db.session = db.create_scoped_session(options=options)

    def teardown():
        db.session.rollback()
        db.session.close()
        connection.close()
    request.addfinalizer(teardown)

    yield db.session



Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source