Skip to content

Instantly share code, notes, and snippets.

@minhoryang
Forked from alexmic/conftest.py
Last active Mar 7, 2016
Embed
What would you like to do?
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment