Code for http://alexmic.net/flask-sqlalchemy-pytest/

conftest.py

import os
import pytest

from alembic.command import upgrade
from alembic.config import Config

from project.factory import create_app
from project.database import db as _db


TESTDB = 'test_project.db'
TESTDB_PATH = "/opt/project/data/{}".format(TESTDB)
TEST_DATABASE_URI = 'sqlite:///' + TESTDB_PATH


ALEMBIC_CONFIG = '/opt/project/alembic.ini'


@pytest.fixture(scope='session')
def app(request):
    """Session-wide test `Flask` application."""
    settings_override = {
        'TESTING': True,
        'SQLALCHEMY_DATABASE_URI': TEST_DATABASE_URI
    }
    app = create_app(__name__, settings_override)

    # Establish an application context before running the tests.
    ctx = app.app_context()
    ctx.push()

    def teardown():
        ctx.pop()

    request.addfinalizer(teardown)
    return app


def apply_migrations():
    """Applies all alembic migrations."""
    config = Config(ALEMBIC_CONFIG)
    upgrade(config, 'head')


@pytest.fixture(scope='session')
def db(app, request):
    """Session-wide test database."""
    if os.path.exists(TESTDB_PATH):
        os.unlink(TESTDB_PATH)

    def teardown():
        _db.drop_all()
        os.unlink(TESTDB_PATH)

    _db.app = app
    apply_migrations()

    request.addfinalizer(teardown)
    return _db


@pytest.fixture(scope='function')
def session(db, request):
    """Creates a new database session for a test."""
    connection = db.engine.connect()
    transaction = connection.begin()

    options = dict(bind=connection, binds={})
    session = db.create_scoped_session(options=options)
    
    db.session = session

    def teardown():
        transaction.rollback()
        connection.close()
        session.remove()

    request.addfinalizer(teardown)
    return session

database.py

from flask.ext.sqlalchemy import SQLAlchemy, SignallingSession, SessionBase


class _SignallingSession(SignallingSession):
    """A subclass of `SignallingSession` that allows for `binds` to be specified
    in the `options` keyword arguments.

    """
    def __init__(self, db, autocommit=False, autoflush=True, **options):
        self.app = db.get_app()
        self._model_changes = {}
        self.emit_modification_signals = \
            self.app.config['SQLALCHEMY_TRACK_MODIFICATIONS']

        bind = options.pop('bind', None)
        if bind is None:
            bind = db.engine

        binds = options.pop('binds', None)
        if binds is None:
            binds = db.get_binds(self.app)

        SessionBase.__init__(self,
                             autocommit=autocommit,
                             autoflush=autoflush,
                             bind=bind,
                             binds=binds,
                             **options)


class _SQLAlchemy(SQLAlchemy):
    """A subclass of `SQLAlchemy` that uses `_SignallingSession`."""
    def create_session(self, options):
        return _SignallingSession(self, **options)


db = _SQLAlchemy()

@stefanprobst, it seems like he is doing that to replace the live app instance with the test app instance which he creates with the app fixture and which has specific test related config values set in settings_override (namely TESTING and SQLALCHEMY_DATABASE_URI).
So without that line the _SQLAlchemy instance wouldn't have the proper config values set for the test environment.

I was using Flask 0.11 and Line 76: session.remove() wasnt working, updated to Flask==0.12(latest as for now) and worked.
(Probably this Flask PR has to do with the fix.)
(spent some hours until i realize why it didn't work in case anybody else wastes time on that)

I have a similar use case. However, I don't need to use transaction for each test, I'm fine with new DB each time. Did anyone succeed at getting it work without subclassing SignallingSession? Sample fixtures which fail with in-memory SQLite (I did not test it with other DB yet).

@pytest.fixture()
def app():
    """An application for the tests."""
    _app = create_app(TestConfig)
    ctx = _app.test_request_context()
    ctx.push()

    yield _app

    ctx.pop()


@pytest.fixture
def db(request, app):
    _db.app = app
    with app.app_context():
        from flask_migrate import upgrade
        upgrade(directory=TestConfig.ALEMBIC_DIR)

    yield _db

    # Explicitly close DB connection
    _db.session.close()
    _db.drop_all()

I can confirm that Alembic migration gets executed, yet tests still fail with "no such table" exceptions.

Would be nice if someone could post a full working example so I/we can observe:

• The flask directory structure
• Where and how the models are defined (what Base do they use, etc)
• Where and how factory.py is made
• How to run the flask application without using pytest

Given the current snippets of code, it is hard to 'guess' how the rest of the application should look like. Instead of a gist with 2 files, a full working Flask application would be best.

One example, snippet database.py has db = _SQLAlchemy(), however, SQLAlchemy should be called with an app parameter. How is this suppose to work?

@alexmic what's the point of writing a tutorial if you do not give the full code? :)

I second that, I came here expecting to see a fully working example and am instead leaving with a bunch of question marks... this snippet requires too much guessing and imagination.

hoping to make things more clear. i am not using a factory. here's my app's __init__.py:

from flask import Flask, request
app = Flask(__name__)

from . import db

here is what i have in my db.py:

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

from . import app
db = SQLAlchemy(app, session_options={'autocommit': True})

finally, here is how i set up my session in my conftest.py:

import pytest

from server.db import db

@pytest.fixture
def session(request):
  """Creates a session that's bound to a connection. See:
  http://alexmic.net/flask-sqlalchemy-pytest/
  """
  # first, set up our connection-scoped session
  connection = db.engine.connect()
  transaction = connection.begin()

  options = dict(bind=connection, binds={})
  session = db.create_scoped_session(options=options)

  # this is how we're going to clean up
  def teardown():
    transaction.rollback()
    connection.close()
    session.remove()

  request.addfinalizer(teardown)

  # finally, use the session we made
  db.session = session
  return db.session

server is the name of the app; that is, the __init__.py is stored at server/__init__.py and i can do from server import db because the root of my repo is part of PYTHONPATH for my test runs. i make that happen via tox (in tox.ini):

[testenv]
# we install everything in 'requirements.txt' and also 'pytest'
deps = -Ur{toxinidir}/requirements.txt
       pytest

# this allows pytest to import local modules like `server`
setenv =
    PYTHONPATH={toxinidir}

This gist and the related blog post saved me so much time and helped me understand the flask/pytest setup/ecosystem better.

Thanks! Beautifully written. 🕊

This one is gold! Cleared up so much for me, thanks!