python - 在集成测试 (Pytest) 和 Flask 应用程序之间创建不同的 SQLAlchemy session

标签 python flask sqlalchemy flask-sqlalchemy pytest

我有使用flask-sqlalchemy lib的flask应用程序(RESTful API后端)。集成测试使用带有 fixture 的 Pytest,一些创建用于测试目的的记录。
问题在于,当测试预期会在数据库级别引发唯一约束失败的场景时,通过 fixture 为测试创建的记录全部回滚,导致其他测试失败并显示错误 sqlalchemy.exc.InvalidRequestError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (sqlite3.IntegrityError) UNIQUE constraint failed: my_table.field1 .

我如何着手创建不同的 SQLAlchemy session 以进行测试,以便测试记录可以提交到数据库并且不受 Flask 请求生命周期内发生的错误的影响?

### globals.py ###

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()
### app.py ###

from sqlalchemy.exc import IntegrityError
from .globals import db

def handle_unique_constraint(error):
    return {"msg": "Duplicate"}, 409

def create_app(connection_string):
    app = Flask(__name__)
    app.register_error_handler(IntegrityError, handle_unique_constraint)
    app.config['SQLALCHEMY_DATABASE_URI'] = connection_string

    # register API blueprint ...
    # e.g. create new user record, done like this:
    #
    # db.session.add(User(**{'email': path_param}))
    # db.session.commit()

    db.init_app(app)
    return app
### conftest.py ###

from my_package.app import create_app
from my_package.globals import db as app_db
from my_package.models.user import User

@fixture(scope='session')
def app(request):
    app = create_app('https://localhost')
    app.debug = True

    with app.app_context():
        yield app

@fixture(scope='session')
def db(app):
    app_db.create_all()
    return app_db

@fixture(scope='session')
def client(app):

    with app.test_client() as client:
        yield client

@fixture(scope='function')
def test_user(db):
    user = User(**{'email': generate_random()})
    db.session.add(user)
    db.session.commit()
    db.session.refresh(user)
### test_user.py ###



# This test passses

def test_repeat_email(client, test_user):
    resp = client.post('/users/emails/{}'.format(test_user.email))
    assert resp.status_code == 409


# This test errors out during setting up test_user fixture
# with aforementioned error message

def test_good_email(client, test_user): # <- this 
    resp = client.post('/users/emails/{}'.format('unique@example.com'))
    assert resp.status_code == 201

最佳答案

你必须实现一个 setUp和一个 tearDown .
运行测试时 setUp将在每次测试开始时运行。 tearDown将在每个结束时运行。
setUp您将: 初始化数据库
tearnDown您将:关闭数据库连接,删除创建的项目...

我不熟悉 pytest .根据thisthis .
之前yield这是setUp部分,然后是 tearDown .
你应该有这样的东西:

@fixture(scope='session')
def app(request):
    app = create_app('https://localhost')
    app.debug = True
    
    ctx = app.app_context()
    ctx.push()

    yield app

    ctx.pop()


@fixture(scope='session')
def db(app):
    app_db.create_all()
    yield app_db
    app_db.drop_all()

关于python - 在集成测试 (Pytest) 和 Flask 应用程序之间创建不同的 SQLAlchemy session ,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/59678307/

相关文章:

python - 如何删除 Pandas Dataframe 中特定列仅包含数字的行?

python - 使用 Python 和 Flask 进行音频直播

python - 如何使用http协议(protocol)实现社交账号认证

python - 为什么 SQLAlchemy 将 hstore 字段初始化为空?

python - 如何使用 python 将值或文本添加到元素上方的其他级别

python - pyspark 聚合,同时找到组的第一个值

python - 在没有代码重复的情况下将 Flask 页面嵌入另一个页面?

python - SqlAlchemy:检查一个对象是否处于任何关系中(或_(object.relationship1.contains(otherObject),object.relationship2.contains(otherObject))

python - Flask-SQLAlchemy .like() 方法引发 NotImplementedError

python - 我应该如何在 Python 中加密 API token ?