Last active
July 27, 2022 00:06
-
-
Save alexmic/7857543 to your computer and use it in GitHub Desktop.
This file contains hidden or 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 hidden or 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() |
hoping to make things more clear. i am not using a factory. here's my app's __init__.py
:
from flask import Flask, request
app = Flask(__name__)
from . import db
here is what i have in my db.py
:
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 conftest.py
:
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
server
is the name of the app; that is, the __init__.py
is stored at server/__init__.py
and i can do from server import db
because the root of my repo is part of PYTHONPATH
for my test runs. i make that happen via tox (in tox.ini
):
[testenv]
# we install everything in 'requirements.txt' and also 'pytest'
deps = -Ur{toxinidir}/requirements.txt
pytest
# this allows pytest to import local modules like `server`
setenv =
PYTHONPATH={toxinidir}
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
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.