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 |
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() |
This comment has been minimized.
This comment has been minimized.
Thank you to share this example code! This is really helpful. |
This comment has been minimized.
This comment has been minimized.
I got the following error when I was trying this example (excluding the custom DB file): |
This comment has been minimized.
This comment has been minimized.
I'm having the same issue as @krmarien - the example won't work without that factory function! |
This comment has been minimized.
This comment has been minimized.
Hi @krmarien and @flyte in case you're still searching, my solution was to substitute app = Flask(__name__)
app.config.update(settings_override) Which is what, I assume, the function must do, given it's name and arguments. And of course, the change @vlttnv mentioned is also required, since the gist were probably written with an older version of There I did it, I mentioned everyone in the thread Bai |
This comment has been minimized.
This comment has been minimized.
I have code that imports the session from a models.py directly. Overwriting that (
|
This comment has been minimized.
This comment has been minimized.
Is it possible to change the app to used the session with rollback too? It doesn't look like this will work with app.test_client() for example. |
This comment has been minimized.
This comment has been minimized.
@chekunkov, the commit you pointed to didn't actually help: empty dict |
This comment has been minimized.
This comment has been minimized.
Hm, they already fixed it in upstream, but current version from flask_sqlalchemy import SignallingSession
def wrap_signalling_session(cls):
_original_init = cls.__init__
def _fixed_init(self, *args, **kwargs):
binds = kwargs.get('binds')
if binds == {}:
def empty(*args, **kwargs):
pass
self._add_bind = empty # override class' method with a function
_original_init(self, *args, **kwargs)
if binds == {}:
del self._add_bind
cls.__init__ = _fixed_init
wrap_signalling_session(SignallingSession) |
This comment has been minimized.
This comment has been minimized.
Another catch: the approach described in the article is great, but it has a non-obvious caveat - took me several hours to debug. from werkzeug.local import LocalProxy
db = ...
def init_app():
session = LocalProxy(lambda: db.session)
admin = ...
admin.add_view(ModelView(User, session)) As a result, it always uses the correct session. |
This comment has been minimized.
This comment has been minimized.
I didn't want to have to change my application code to fix my tests, so I opted for the hack suggested here and faithfully reproduced for your convenience: class _dict(dict):
def __nonzero__(self):
return True
connection = db.engine.connect()
transaction = connection.begin()
options = dict(bind=connection, binds=_dict())
session = db.create_scoped_session(options=options) Thanks for the help y'all! |
This comment has been minimized.
This comment has been minimized.
Could somebody please explain to me why the line |
This comment has been minimized.
This comment has been minimized.
@stefanprobst, it seems like he is doing that to replace the live |
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
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).
I can confirm that Alembic migration gets executed, yet tests still fail with "no such table" exceptions. |
This comment has been minimized.
This comment has been minimized.
Would be nice if someone could post a full working example so I/we can observe:
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 @alexmic what's the point of writing a tutorial if you do not give the full code? :) |
This comment has been minimized.
This comment has been minimized.
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. |
This comment has been minimized.
This comment has been minimized.
hoping to make things more clear. i am not using a factory. here's my app's from flask import Flask, request
app = Flask(__name__)
from . import db here is what i have in my 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 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
|
This comment has been minimized.
This comment has been minimized.
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 comment has been minimized.
This comment has been minimized.
This one is gold! Cleared up so much for me, thanks! |
This comment has been minimized.
Looks like signalling session from flask-sqlalchemy should work out of the box now pallets/flask-sqlalchemy@ac135c6