pytest-flask example with SQLAlchemy

[andredias]

Hi, there Could you include a SQLAlchemy example in the documentation? I'm using the following code but it doesn't work:

import pytest

from projeto import create_app, db as _db


@pytest.fixture(scope='session')
def app():
    app = create_app('testing')
    return app


@pytest.fixture(scope='session')
def db(app, request):
    _db.app = app
    _db.create_all()
    yield _db
    _db.drop_all()


@pytest.yield_fixture(scope='function')
def session(db):
    connection = db.engine.connect()
    transaction = connection.begin()

    options = dict(bind=connection)
    session = db.create_scoped_session(options=options)

    db.session = session

    yield session

    transaction.rollback()
    connection.close()
    session.remove()

This is the error that I get:

app = <Flask 'projeto'>, request = <SubRequest 'db' for <Function 'test_login_page'>>

    @pytest.fixture(scope='session')
    def db(app, request):
       _db.app = app
       _db.create_all()
       yield _db
 >       _db.drop_all()

conftest.py:14: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/home/andref/.virtualenvs/projeto/lib/python3.5/site-packages/flask_sqlalchemy/__init__.py:1015: in drop_all
    self._execute_for_all_tables(app, bind, 'drop_all')
/home/andref/.virtualenvs/projeto/lib/python3.5/site-packages/flask_sqlalchemy/__init__.py:999: in _execute_for_all_tables
    op(bind=self.get_engine(app, bind), **extra)
/home/andref/.virtualenvs/projeto/lib/python3.5/site-packages/flask_sqlalchemy/__init__.py:941: in get_engine
    return connector.get_engine()
/home/andref/.virtualenvs/projeto/lib/python3.5/site-packages/flask_sqlalchemy/__init__.py:533: in get_engine
    uri = self.get_uri()
/home/andref/.virtualenvs/projeto/lib/python3.5/site-packages/flask_sqlalchemy/__init__.py:524: in get_uri
    return self._app.config['SQLALCHEMY_DATABASE_URI']
/home/andref/.virtualenvs/projeto/lib/python3.5/site-packages/werkzeug/local.py:347: in __getattr__
    return getattr(self._get_current_object(), name)
/home/andref/.virtualenvs/projeto/lib/python3.5/site-packages/werkzeug/local.py:306: in _get_current_object
    return self.__local()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    def _find_app():
        top = _app_ctx_stack.top
        if top is None:
>           raise RuntimeError(_app_ctx_err_msg)
E           RuntimeError: Working outside of application context.
E           
E           This typically means that you attempted to use functionality that needed
E           to interface with the current application object in a way.  To solve
E           this set up an application context with app.app_context().  See the
E           documentation for more information.

/home/andref/.virtualenvs/projeto/lib/python3.5/site-packages/flask/globals.py:51: RuntimeError

[sanderfoobar]

+1

Confusing docs.

[scottwilson312]

+1

[pmourlanne]

I got something working, thanks to http://alexmic.net/flask-sqlalchemy-pytest/

Here's my code:

import os

import pytest

from myapp import create_app, db as _db


@pytest.fixture(scope='session')
def app():
    app = create_app()
    app.config.from_object('test_settings')
    return app


@pytest.fixture(scope='session')
def db(app, request):
    if os.path.exists(app.config['DB_PATH']):
        os.unlink(app.config['DB_PATH'])

    def teardown():
        _db.drop_all()
        os.unlink(app.config['DB_PATH'])

    _db.init_app(app)
    _db.create_all()

    request.addfinalizer(teardown)
    return _db

[kenshiro-o]

After much searching and hair-tearing, I found that the current approach works quite nicely:

# module conftest.py
import pytest

from app import create_app
from app import db as _db
from sqlalchemy import event
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope="session")
def app(request):
    """
    Returns session-wide application.
    """
    return create_app("testing")


@pytest.fixture(scope="session")
def db(app, request):
    """
    Returns session-wide initialised database.
    """
    with app.app_context():
        _db.drop_all()
        _db.create_all()


@pytest.fixture(scope="function", autouse=True)
def session(app, db, request):
    """
    Returns function-scoped session.
    """
    with app.app_context():
        conn = _db.engine.connect()
        txn = conn.begin()

        options = dict(bind=conn, binds={})
        sess = _db.create_scoped_session(options=options)

        # establish  a SAVEPOINT just before beginning the test
        # (http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#using-savepoint)
        sess.begin_nested()

        @event.listens_for(sess(), 'after_transaction_end')
        def restart_savepoint(sess2, trans):
            # Detecting whether this is indeed the nested transaction of the test
            if trans.nested and not trans._parent.nested:
                # The test should have normally called session.commit(),
                # but to be safe we explicitly expire the session
                sess2.expire_all()
                sess.begin_nested()

        _db.session = sess
        yield sess

        # Cleanup
        sess.remove()
        # This instruction rollsback any commit that were executed in the tests.
        txn.rollback()
        conn.close()

The key here is to run your tests within a nested session, and then rollback everything after the execution of each test (this also assumes there are no dependencies across your tests).

[nicoddemus]

@kenshiro-o it would be great if you could open a PR adding this to the docs somewhere. 😁

[kenshiro-o]

@nicoddemus sure! Will do so during the week :smile_cat:

[AnderUstarroz]

Hi @kenshiro-o , Have you ever tried using SQLAlchemy fixture + Factoryboy? I was wondering how Factoryboy could obtain an instance of your session fixture. This is the default example in the documentation, but doesn't seems to work when testing:

from sqlalchemy import Column, Integer, Unicode, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine('sqlite://')
session = scoped_session(sessionmaker(bind=engine))
Base = declarative_base()


class User(Base):
    """ A SQLAlchemy simple model class who represents a user """
    __tablename__ = 'UserTable'

    id = Column(Integer(), primary_key=True)
    name = Column(Unicode(20))

Base.metadata.create_all(engine)

import factory

class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = session   # Here needs the SQLAlchemy session object!!!!!!

    id = factory.Sequence(lambda n: n)
    name = factory.Sequence(lambda n: u'User %d' % n)

I tried updating the factories during session creation and it actually works:

@pytest.fixture(scope="function", autouse=True)
def session(app, db, request):
    """
    Returns function-scoped session.
    """
    with app.app_context():
        conn = _db.engine.connect()
        txn = conn.begin()

        options = dict(bind=conn, binds={})
        sess = _db.create_scoped_session(options=options)

        # establish  a SAVEPOINT just before beginning the test
        # (http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#using-savepoint)
        sess.begin_nested()

        @event.listens_for(sess(), 'after_transaction_end')
        def restart_savepoint(sess2, trans):
            # Detecting whether this is indeed the nested transaction of the test
            if trans.nested and not trans._parent.nested:
                # The test should have normally called session.commit(),
                # but to be safe we explicitly expire the session
                sess2.expire_all()
                sess.begin_nested()

        _db.session = sess
        UserFactory._meta.sqlalchemy_session = sess  # THIS WILL DO THE MAGIC
        yield sess

        # Cleanup
        sess.remove()
        # This instruction rollsback any commit that were executed in the tests.
        txn.rollback()
        conn.close()

but I was wondering if some of you guys know a more "elegant" solution.

[NothisIm]

@AnderUstarroz I encountered the same obstacle in the face factories not using the session, and overcome it with a bit more flexible solution:

from . import factories
for name in dir(factories):
    item = getattr(factories, name)
    if isinstance(item, factories.factory.base.FactoryMetaClass):
        item._meta.sqlalchemy_session = session

[zgoda]

This works with SQLAlchemy + Factoryboy:

@pytest.fixture
def app():
    app = make_app('test')
    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()

Then you may use db.session normally in factory Meta. My tests run against in-memory SQLite.