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 😸

[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.

[gconklin]

Resurrecting this stale issue due to hair loss.

I am trying to apply the "even newer" method of doing this with the latest sqlalchemy noted here that uses a "join_transaction_mode".

Even though I'm manually creating a connection and binding the session maker to the connection, somewhere down the line my created connection with the open transaction is lost. So join_transaction_mode never takes effect which should create a savepoint and not a new transaction. Taking the below code outside of Flask, it works and I can even see in the sqlalchemy logs where it starts a savepoint and rolls it back.

I feel like I'm missing some fundamental piece that I cannot figure out. Or maybe this pattern just doesn't work because of the scoped_session. I'm asking for either help in finding my error, or confirmation that it just won't work at all. I would be greatly appreciative as I've spent too many hours sticking print statements in various places in both flask-sqlalchemy and sqlalchemy and not discovered a way to even hack around it.

Here is what I've been using to test with. One test that creates a record, and another test that should not find the record (because it was supposed to be rolled back, but was not- and so fails)

(Edit to add versions)
Flask==3.1.1
Flask-SQLAlchemy==3.1.1
pytest-flask==1.3.0
SQLAlchemy==2.0.41

app.py

from flask import Flask
from flask import abort
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import select

class Base(DeclarativeBase):
    pass

app = Flask(__name__)
db = SQLAlchemy(model_class=Base)

app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///project.db"
app.config['SQLALCHEMY_ECHO'] = True

db.init_app(app)

class Widget(db.Model):
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    value = db.Column(db.String(100), unique=True, nullable=False)

with app.app_context():
    db.create_all()

@app.route("/getv/<value>")
def getv(value):
    try:
        value = db.session.query(Widget).filter_by(value = value).one()
        return value.value
    except Exception:
        abort(404)

@app.route("/create/<value>", methods=["POST"])
def create(value):
    with db.session.begin():
        rec = Widget(value=value)
        db.session.add(rec)
    return "OK"

tests/test_app.py

import pytest

def test_create(client, session):
    r = client.post("/create/a-widget")
    assert 200 == r.status_code
    assert "OK" == r.text

def test_get_404(client):
    # this should be 404 if rollback was working
    r = client.get("/getv/a-widget")
    assert 404 == r.status_code

tests/conftest.py

import pytest
from app import app as flask_app
from app import db as _db
from app import Widget
from pprint import pprint
from sqlalchemy.orm import sessionmaker, scoped_session, close_all_sessions

@pytest.fixture(scope="session")
def app():
    yield flask_app

@pytest.fixture(scope="session")
def db(app):
    with app.app_context():
        _db.session.execute(_db.delete(Widget))
        _db.session.commit()

        yield _db

# https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites
@pytest.fixture(scope="function", autouse=True)
def session(app, db):
    with app.app_context():
        conn = db.engine.connect()
        txn = conn.begin()
        db.session = db._make_scoped_session(
            options={
                "bind": conn,
                "join_transaction_mode": "create_savepoint"
            }
        )

        yield db.session

        db.session.close()
        txn.rollback()
        conn.close()

@pytest.fixture(scope="function")
def client(app, session):
    with app.test_client() as client:
        yield client