如何在 SQLAlchemy 中测试回滚

问题描述 投票：0回答：1

我在测试逻辑时遇到测试隔离问题，涉及 SQLAlchemy 中的事务回滚。

型号：

class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    company = db.Column(db.Text)
    subtype = db.Column(db.Text)

    __table_args__ = (db.UniqueConstraint(company, subtype),)

查看：

def create():
    instance = Product(**request.json)
    db.session.add(instance)
    try:
        db.session.commit()
    except IntegrityError:
        db.session.rollback()
        return {"detail": "Product object already exists", "status": 406, "title": "Duplicate object"}, 406
    return {"uri": f"/products/{instance.id}"}, 201

测试：

DEFAULT_DATA = {"company": "Test", "subtype": "Sub"}


def test_create(client):
    response = client.post("/products", json=DEFAULT_DATA)
    assert response.status_code == 201
    instance = Product.query.one()
    assert response.json == {"uri": f"/products/{instance.id}"}


def test_create_duplicate(client):
    response = client.post("/products", json=DEFAULT_DATA)
    assert response.status_code == 201

    instance = Product.query.one()
    assert response.json == {"uri": f"/products/{instance.id}"}
    response = client.post("/products", json=DEFAULT_DATA)
    assert response.status_code == 406
    assert response.json == {"detail": "Product object already exists", "status": 406, "title": "Duplicate object"}

conftest.py：

import flask_migrate
import pytest
from sqlalchemy import event

from project.app import create_connexion_app
from project.models import db


@pytest.fixture(scope="session")
def connexion_app():
    return create_connexion_app("project.settings.TestSettings")


@pytest.fixture(scope="session")
def app(connexion_app):
    app = connexion_app.app
    with app.app_context():
        yield app


@pytest.fixture(scope="session", name="db")
def db_setup(app):
    flask_migrate.upgrade()

    yield db

    flask_migrate.downgrade()
    db.engine.execute("DROP TABLE IF EXISTS alembic_version")


@pytest.fixture(autouse=True)
def session(db):
    with db.engine.connect() as connection:

        @event.listens_for(db.session, "after_transaction_end")
        def restart_savepoint(session, transaction):
            if transaction.nested and not transaction._parent.nested:
                # ensure that state is expired the way
                # session.commit() at the top level normally does
                # (optional step)
                session.expire_all()
                session.begin_nested()

        transaction = connection.begin()

        db.session.begin_nested()

        yield db.session

        db.session.rollback()
        db.session.close()

        if transaction.is_active:
            transaction.rollback()


SQLALCHEMY_COMMIT_ON_TEARDOWN
设置为 
False

第二个测试失败，输出如下：

    def test_create_duplicate(client):
        response = client.post("/products", json=DEFAULT_DATA)
>       assert response.status_code == 201
E       AssertionError: assert 406 == 201
E        +  where 406 = <<class 'pytest_flask.plugin.JSONResponse'> streamed [406 NOT ACCEPTABLE]>.status_code

相关PG日志：

LOG:  statement: BEGIN
LOG:  statement: INSERT INTO product (company, subtype) VALUES ('Test', 'Sub') RETURNING product.id
LOG:  statement: COMMIT
LOG:  statement: BEGIN
LOG:  statement: SELECT product.id AS product_id, product.company AS product_company, product.subtype AS product_subtype
    FROM product
    WHERE product.id = 1
LOG:  statement: SELECT product.id AS product_id, product.company AS product_company, product.subtype AS product_subtype
    FROM product
LOG:  statement: ROLLBACK
LOG:  statement: BEGIN
LOG:  statement: INSERT INTO product (company, subtype) VALUES ('Test', 'Sub') RETURNING product.id
ERROR:  duplicate key value violates unique constraint "product_company_subtype_key"
DETAIL:  Key (company, subtype)=(Test, Sub) already exists.
STATEMENT:  INSERT INTO product (company, subtype) VALUES ('Test', 'Sub') RETURNING product.id
LOG:  statement: ROLLBACK

因此，第一个测试将一行提交到数据库中，并且在测试之间不会回滚，因此在运行之间不会恢复数据库状态。

其他测试，不涉及显式回滚，工作正常。尝试将

SQLALCHEMY_COMMIT_ON_TEARDOWN
更改为
True
并使用
flush
代替
commit
，但在这种情况下，
test_create_duplicate
之后的测试会受到影响。

如何设置测试套件来测试此类代码，其中涉及手动提交/回滚？

套餐：

  • 烧瓶==1.0.2
  • Flask-迁移==2.2.1
  • Flask-SQLAlchemy==2.3.2
  • SQLAlchemy==1.2.9
  • 字典炼金==0.1.2.7
  • 连接==1.4.2
  • pytest==3.6.2
  • pytest-flask==0.10.0

Python版本：3.6.6

RDBMS：PostgreSQL 10.4

python postgresql sqlalchemy transactions pytest
1个回答
0
投票

注意：问题提到“RDBMS：PostgreSQL 10.4”，但这无法使用 PostgreSQL 重现。

看来您实际上正在使用 SQLite 和默认的 pysqlite 驱动程序进行测试。

在conftest.py中添加以下内容：

@event.listens_for(db.engine, "connect")
def do_connect(dbapi_connection, connection_record):
    # disable pysqlite's emitting of the BEGIN statement entirely.
    # also stops it from emitting COMMIT before any DDL.
    dbapi_connection.isolation_level = None

@event.listens_for(db.engine, "begin")
def do_begin(conn):
    # emit our own BEGIN
    conn.exec_driver_sql("BEGIN")

来自 https://docs.sqlalchemy.org/en/20/dialects/sqlite.html#pysqlite-serialized：

pysqlite DBAPI 驱动程序有几个长期存在的错误，影响其事务行为的正确性。在其默认操作模式下，SQLite 功能（例如 SERIALIZABLE 隔离、事务性 DDL 和SAVEPOINT 支持）是不起作用的，为了使用这些功能，必须采取解决方法。

...

好消息是，通过一些事件，我们可以通过完全禁用 pysqlite 的功能并自己发出 BEGIN 来完全实现事务支持。这是使用两个事件侦听器实现的：

（如上所述；以及强调我的）

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.