Last active
July 27, 2022 00:06
-
-
Save alexmic/7857543 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 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!
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
hoping to make things more clear. i am not using a factory. here's my app's
__init__.py
:here is what i have in my
db.py
:finally, here is how i set up my session in my
conftest.py
:server
is the name of the app; that is, the__init__.py
is stored atserver/__init__.py
and i can dofrom server import db
because the root of my repo is part ofPYTHONPATH
for my test runs. i make that happen via tox (intox.ini
):