当 pytest 测试断言查询结果行数失败时数据库操作挂起

Question

以下测试工作正常：

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.engine.url import URL
from sqlalchemy_utils import database_exists, create_database

@pytest.fixture()
def db_engine():
    engine = create_engine(
        URL('postgres',
            username='postgres', password='postgres', database='my_database')
    )
    if not database_exists(engine.url):
        create_database(engine.url)
    return engine


def test_new_table_is_empty(db_engine):
    try:
        db_engine.execute(text('CREATE SCHEMA test_schema;'))
        db_engine.execute(text('CREATE TABLE test_schema.new_table ();'))
        result = db_engine.execute(text('SELECT * FROM test_schema.new_table;'))
        assert result.rowcount == 0
    finally:
        try:
            del result  # The result would block the dropping of
                        # SCHEMA "test_schema" in the following cleanup.
        except NameError:
            pass
        db_engine.execute(text('DROP SCHEMA IF EXISTS test_schema CASCADE;'))

但是如果我通过将 assert result.rowcount == 0 更改为 assert result.rowcount == 1 使其失败，它将无限期挂起最后一行（模式应该被删除的地方）甚至不能被 [Ctrl+c] 中止。我必须 kill py.test 进程（或 python 进程，取决于我如何调用测试运行ner）来终止它。（如果我追加

if __name__ == "__main__":
    test_new_table_is_empty(db_engine())

和运行文件作为普通 python 脚本而不是通过 py.test，我得到了预期的 AssertionError。）

但是，如果我只是将断言替换为 assert False（并且再次将运行替换为 py.test），则测试套件会按预期以一次失败的测试结束。因此，我假设 pytest 保留对 result 的引用，前提是断言失败，这可能是因为它与堆栈跟踪一起显示的错误分析。 是这样吗？

我怎样才能并且应该避免阻塞？ 我应该只对从结果中获取的数据而不是 ResultProxy 本身的属性进行测试断言吗？

Answer 1

TL;DR

打电话

result.close()

而不是

del result

您具体问题的答案：

I assume that pytest retains a reference to result iff the assertion failed, probably for the error analysis it displays with the stack trace. Is that the case?

这仍然是我的工作假设。有知道的还请赐教

How can and should I avoid the blocking?

而不是 del设置 ResultProxy result，在 finally 子句中明确 close() 它：

def test_new_table_is_empty(db_engine):
    try:
        db_engine.execute(text('CREATE SCHEMA test_schema;'))
        db_engine.execute(text('CREATE TABLE test_schema.new_table ();'))
        result = db_engine.execute(text('SELECT * FROM test_schema.new_table;'))
        assert result.rowcount == 0
    finally:
        try:
            result.close()  # Release row and table locks.
        except NameError:
            pass
        db_engine.execute(text('DROP SCHEMA IF EXISTS test_schema CASCADE;'))

这将 release 所有行和 result 持有的 table 锁。

要~~避免嵌套 try 子句的混乱~~ 将嵌套 try 子句的混乱移动到别处，您可以使用 with contextlib.closing(...):：

from contextlib import closing

# ...

def test_new_table_is_empty(db_engine):
    try:
        db_engine.execute(text('CREATE SCHEMA test_schema;'))
        db_engine.execute(text('CREATE TABLE test_schema.new_table ();'))
        with closing(
                db_engine.execute(text('SELECT * FROM test_schema.new_table;'))
        ) as result:
            assert result.rowcount == 0
    finally:
        db_engine.execute(text('DROP SCHEMA IF EXISTS test_schema CASCADE;'))

Should I only ever make test assertions on data fetched from the result rather than on properties of the ResultProxy itself?

只有当您获取所有行时才有效，从而耗尽 ResultProxy，这将隐式地 _soft_close() 它。如果结果的行数（可能出乎意料地）多于您要获取的行数，则结果将保持打开状态并继续持有可能阻止执行以下清理的锁。

由于您只对测试中的行计数感兴趣，而不对实际结果的内容感兴趣，因此明确关闭是比获取您不会使用的结果更好的选择，除了可能对它们进行计数或计算它们的长度。

当 pytest 测试断言查询结果行数失败时数据库操作挂起

Database operation hangs when pytest test asserting query result rowcount fails

python

postgresql

deadlock

sqlalchemy

pytest

TL;DR

您具体问题的答案：