首页 > 解决方案 > 由于数据库上的元数据锁定,SQL Alchemy 不会在测试结束时删除表

问题描述

我正在测试我的烧瓶应用程序模型。我正在使用 mysql 5.7、sqlalchemy 和 pytest。

在我的模型中,我有一个 CRUD mixin,用于管理创建、更新和删除。每当我在将对象返回给测试函数之前尝试访问 Mixin 中的对象时,SQLAlchemy 在我的拆解中挂在 db.drop_all 处。当我在 PROCESSLIST 中查看 mysql 时,它显示 1 个睡眠查询和 1 个等待表元数据锁定的查询。

我可以通过在返回对象之前在 mixin 的 create 方法中调用 db.session.commit 来解决这个问题。但是,如果我在测试拆解时(或在主测试函数中)调用它,它就不起作用。我不想仅仅为了让我的测试工作而添加额外的提交,因为它感觉不正确。有谁知道为什么会发生这种情况或有任何建议的修复?

模型.py

class CRUDMixin(object):

    @classmethod
    def create(cls, **kwargs):
        instance = cls(**kwargs)
        saved_instance = instance.save()
        # do stuff with saved_instance (i.e. add to full text search engine)
        # db.drop_all in teardown works if add db.session.commit() here
        return saved_instance

    def save(self, commit=True):
        db.session.add(self)
        if commit:
            try:
                db.session.commit()
            except Exception:
                db.session.rollback()
                raise
        return self

class User(CRUDMixin, db.model):

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))

conftest.py

@pytest.fixture(scope='session')
def app():
    app = create_app(TestConfig)
    ctx = app.app_context()
    ctx.push()
    yield app
    ctx.pop()

@pytest.fixture(scope='session')
def test_db(app):
    db.drop_all()
    db.create_all()
    # add test db information
    yield db
    db.session.remove()
    db.drop_all()  # test hangs within drop all

@pytest.fixture(scope='function')
def db_session(test_db):
    connection = db.engine.connect()
    transaction = connection.begin()
    options = dict(bind=connection, binds={})
    session = db.create_scoped_session(options)
    db.session = session
    yield db
    db.session.remove() # tables won't drop if I put db.session.commit before the remove call
    transaction.rollback()
    connection.close() # even though connection closes, mysql still shows process

test_models.py

class TestUser(object):
 
    def test_add_new(self, db_session):
        u = User.create(name='test_name')
        assert u.name == 'test_name'
        # if I put db.session.commit() here, tables won't drop

标签: pythonmysqlsqlalchemyflask-sqlalchemypytest

解决方案


推荐阅读