pytest-flask example with SQLAlchemy #70
Comments
|
+1 Confusing docs. |
|
+1 |
|
I got something working, thanks to http://alexmic.net/flask-sqlalchemy-pytest/ Here's my code: import os
import pytest
from myapp import create_app, db as _db
@pytest.fixture(scope='session')
def app():
app = create_app()
app.config.from_object('test_settings')
return app
@pytest.fixture(scope='session')
def db(app, request):
if os.path.exists(app.config['DB_PATH']):
os.unlink(app.config['DB_PATH'])
def teardown():
_db.drop_all()
os.unlink(app.config['DB_PATH'])
_db.init_app(app)
_db.create_all()
request.addfinalizer(teardown)
return _db |
|
After much searching and hair-tearing, I found that the current approach works quite nicely: # module conftest.py
import pytest
from app import create_app
from app import db as _db
from sqlalchemy import event
from sqlalchemy.orm import sessionmaker
@pytest.fixture(scope="session")
def app(request):
"""
Returns session-wide application.
"""
return create_app("testing")
@pytest.fixture(scope="session")
def db(app, request):
"""
Returns session-wide initialised database.
"""
with app.app_context():
_db.drop_all()
_db.create_all()
@pytest.fixture(scope="function", autouse=True)
def session(app, db, request):
"""
Returns function-scoped session.
"""
with app.app_context():
conn = _db.engine.connect()
txn = conn.begin()
options = dict(bind=conn, binds={})
sess = _db.create_scoped_session(options=options)
# establish a SAVEPOINT just before beginning the test
# (http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#using-savepoint)
sess.begin_nested()
@event.listens_for(sess(), 'after_transaction_end')
def restart_savepoint(sess2, trans):
# Detecting whether this is indeed the nested transaction of the test
if trans.nested and not trans._parent.nested:
# The test should have normally called session.commit(),
# but to be safe we explicitly expire the session
sess2.expire_all()
sess.begin_nested()
_db.session = sess
yield sess
# Cleanup
sess.remove()
# This instruction rollsback any commit that were executed in the tests.
txn.rollback()
conn.close()The key here is to run your tests within a nested session, and then rollback everything after the execution of each test (this also assumes there are no dependencies across your tests). |
|
@kenshiro-o it would be great if you could open a PR adding this to the docs somewhere. 😁 |
|
@nicoddemus sure! Will do so during the week 😸 |
|
Hi @kenshiro-o , I tried updating the factories during session creation and it actually works: but I was wondering if some of you guys know a more "elegant" solution. |
|
@AnderUstarroz I encountered the same obstacle in the face factories not using the session, and overcome it with a bit more flexible solution: |
|
This works with SQLAlchemy + Factoryboy: @pytest.fixture
def app():
app = make_app('test')
with app.app_context():
db.create_all()
yield app
db.session.remove()
db.drop_all()Then you may use |
|
Resurrecting this stale issue due to hair loss. I am trying to apply the "even newer" method of doing this with the latest sqlalchemy noted here that uses a "join_transaction_mode". Even though I'm manually creating a connection and binding the session maker to the connection, somewhere down the line my created connection with the open transaction is lost. So join_transaction_mode never takes effect which should create a savepoint and not a new transaction. Taking the below code outside of Flask, it works and I can even see in the sqlalchemy logs where it starts a savepoint and rolls it back. I feel like I'm missing some fundamental piece that I cannot figure out. Or maybe this pattern just doesn't work because of the scoped_session. I'm asking for either help in finding my error, or confirmation that it just won't work at all. I would be greatly appreciative as I've spent too many hours sticking print statements in various places in both flask-sqlalchemy and sqlalchemy and not discovered a way to even hack around it. Here is what I've been using to test with. One test that creates a record, and another test that should not find the record (because it was supposed to be rolled back, but was not- and so fails) (Edit to add versions) app.py from flask import Flask
from flask import abort
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import select
class Base(DeclarativeBase):
pass
app = Flask(__name__)
db = SQLAlchemy(model_class=Base)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///project.db"
app.config['SQLALCHEMY_ECHO'] = True
db.init_app(app)
class Widget(db.Model):
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
value = db.Column(db.String(100), unique=True, nullable=False)
with app.app_context():
db.create_all()
@app.route("/getv/<value>")
def getv(value):
try:
value = db.session.query(Widget).filter_by(value = value).one()
return value.value
except Exception:
abort(404)
@app.route("/create/<value>", methods=["POST"])
def create(value):
with db.session.begin():
rec = Widget(value=value)
db.session.add(rec)
return "OK"tests/test_app.py import pytest
def test_create(client, session):
r = client.post("/create/a-widget")
assert 200 == r.status_code
assert "OK" == r.text
def test_get_404(client):
# this should be 404 if rollback was working
r = client.get("/getv/a-widget")
assert 404 == r.status_codetests/conftest.py import pytest
from app import app as flask_app
from app import db as _db
from app import Widget
from pprint import pprint
from sqlalchemy.orm import sessionmaker, scoped_session, close_all_sessions
@pytest.fixture(scope="session")
def app():
yield flask_app
@pytest.fixture(scope="session")
def db(app):
with app.app_context():
_db.session.execute(_db.delete(Widget))
_db.session.commit()
yield _db
# https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites
@pytest.fixture(scope="function", autouse=True)
def session(app, db):
with app.app_context():
conn = db.engine.connect()
txn = conn.begin()
db.session = db._make_scoped_session(
options={
"bind": conn,
"join_transaction_mode": "create_savepoint"
}
)
yield db.session
db.session.close()
txn.rollback()
conn.close()
@pytest.fixture(scope="function")
def client(app, session):
with app.test_client() as client:
yield client |
Hi, there
Could you include a SQLAlchemy example in the documentation? I'm using the following code but it doesn't work:
This is the error that I get:
The text was updated successfully, but these errors were encountered: