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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
This one is gold! Cleared up so much for me, thanks!