我正在使用pytest为大型应用程序重写测试套件，并希望在每个测试函数之间实现隔离。我注意到，在SAVEPOINT中对commit的多次调用导致记录被输入到DB中。我已经为以下示例提取了尽可能多的代码：
init.py
# Create the SQLAlchemy db instance
db: SQLAlchemy = SQLAlchemy(
engine_options={"connect_args": {"options": "-c timezone=utc"}}
)
# Initialize Marshmallow
ma: Marshmallow = Marshmallow()
unleash = Unleash()
def create_app(config=None):
# Create the Flask app
app = Flask(__name__)
# Flask is not autoloading FLASK_ENV anymore
app.config.from_object("app.config.Testing")
if not database_exists(app.config["SQLALCHEMY_DATABASE_URI"]):
create_database(app.config["SQLALCHEMY_DATABASE_URI"])
# initialize DB
db.init_app(app)
# Initialize Marshmallow
ma.init_app(app)
with app.app_context():
# import models and setup blueprints
...
conftest.py
import os
import pytest
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import event
from sqlalchemy.orm import Session
import db_data
from app import create_app
from app import db as _db
@pytest.fixture(scope="session")
def app():
"""
Returns session-wide application.
"""
os.environ["FLASK_ENV"] = "testing"
return create_app()
@pytest.fixture(scope="session")
def db(app: Flask, request):
"""
Returns session-wide initialised database.
"""
with app.app_context():
_db.drop_all()
_db.create_all()
db_data.initialize_common_data(_db, app.logger)
db_data.create_test_users(_db, app.logger)
db_data.initialize_functional_test_data(_db, app.logger)
yield _db
@pytest.fixture(scope="function", autouse=True)
def session(app: Flask, db: SQLAlchemy):
"""
Returns function-scoped session.
"""
# from https://docs.sqlalchemy.org/en/13/orm/session_transaction.html
with app.app_context():
connection = db.engine.connect()
# begin a non-ORM transaction
trans = connection.begin()
# bind an individual Session to the connection
sess = Session(bind=connection)
# start the session in a SAVEPOINT...
sess.begin_nested()
# then each time that SAVEPOINT ends, reopen it
@event.listens_for(sess, "after_transaction_end")
def restart_savepoint(s, t):
if t.nested and (t._parent is None or not t._parent.nested):
s.expire_all()
s.begin_nested()
yield sess
### Cleanup ##
# rollback - everything that happened with the Session above
# (including calls to commit()) is rolled back.
sess.close()
trans.rollback()
# return connection to the Engine
connection.close()
conftestv2.py
@pytest.fixture(scope="function", autouse=True)
def session(app: Flask, db: SQLAlchemy):
"""
Returns function-scoped session.
"""
# from https://docs.sqlalchemy.org/en/13/orm/session_transaction.html
with app.app_context():
connection = db.engine.connect()
# begin a non-ORM transaction
trans = connection.begin()
# start the session in a SAVEPOINT...
db.session.begin_nested()
# then each time that SAVEPOINT ends, reopen it
@event.listens_for(db.session, "after_transaction_end")
def restart_savepoint(s, t):
if t.nested and (t._parent is None or not t._parent.nested):
s.expire_all()
s.begin_nested()
# yield sess
yield db.session
### Cleanup ##
# rollback - everything that happened with the Session above
# (including calls to commit()) is rolled back.
db.session.close()
trans.rollback()
# return connection to the Engine
connection.close()
test_user.py
from app.user.models import User
from app import db as _db
def test_list_trackers():
print("SESSSS:", _db.session)
_db.session.add(User(email="aaa@aaa.aaa"))
_db.session.commit()
_db.session.add(User(email="aaa@aaa.aaab"))
_db.session.commit()
我试着用这个作为指导：https://docs.sqlalchemy.org/en/13/orm/session_transaction.html#joining-a-session-in-and-external-transaction-and-such as-for-test-suites
并研究了以下资源(以及许多其他资源(：
- 在Flask中的测试之间回滚许多事务
- https://docs.sqlalchemy.org/en/13/orm/session_api.html#sqlalchemy.orm.session.Session
- https://docs.sqlalchemy.org/en/14/orm/events.html#sqlalchemy.orm.SessionEvents.after_transaction_create
- https://github.com/jeancochrane/pytest-flask-sqlalchemy/issues/28
我安装的相关软件包和版本：
$ pip list
Package Version
--------------------------------- ---------
coverage 6.3.1
Faker 12.1.0
Flask 1.1.2
Flask-Cors 3.0.10
Flask-Environments 0.1
Flask-HTTPAuth 4.2.0
flask-marshmallow 0.11.0
Flask-Migrate 2.7.0
Flask-SQLAlchemy 2.5.1
psycopg2-binary 2.8.6
pytest 6.2.3
pytest-cov 3.0.0
SQLAlchemy 1.3.18
SQLAlchemy-Paginator 0.2
sqlalchemy-stubs 0.3
SQLAlchemy-Utils 0.38.2
timezonefinder 5.2
数据库正在使用postgres 12.2-alpine映像在docker-compose中运行。
在SQLAlchemy的Gitter社区的帮助下，我能够解决这个问题。有两个问题需要解决：
- 为每个单独的测试注册
after_transaction_end事件，但在测试结束后没有删除。因此，在每个测试之间都会调用多个事件 - 从
dbfixture中生成的_db位于应用程序上下文中，而它本不应该位于该上下文中
更新的conftest.py:
@pytest.fixture(scope="session")
def db(app: Flask, request):
"""
Returns session-wide initialised database.
"""
with app.app_context():
_db.drop_all()
_db.create_all()
db_data.initialize_common_data(_db, app.logger)
db_data.create_test_users(_db, app.logger)
db_data.initialize_functional_test_data(_db, app.logger)
_db.session.close_all()
yield _db
@pytest.fixture(scope="function", autouse=True)
def session(app: Flask, db: SQLAlchemy):
"""
Returns function-scoped session.
"""
# from https://docs.sqlalchemy.org/en/13/orm/session_transaction.html
with app.app_context():
connection = db.engine.connect()
# begin a non-ORM transaction
trans = connection.begin()
# start the session in a SAVEPOINT...
db.session.begin_nested()
# then each time that SAVEPOINT ends, reopen it
@event.listens_for(db.session, "after_transaction_end")
def restart_savepoint(s, t):
if t.nested and (t._parent is not None and not t._parent.nested):
s.expire_all()
s.begin_nested()
# yield sess
yield db.session
### Cleanup ##
# rollback - everything that happened with the Session above
# (including calls to commit()) is rolled back.
event.remove(db.session, "after_transaction_end", restart_savepoint)
db.session.close_all()
trans.rollback()
# return connection to the Engine
connection.close()