python - 当 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 使其失败，它 < strong>将无限期地挂起在最后一行(模式应该被删除的地方)并且甚至不能被 [Ctrl+c] 中止。我必须kill py.test 进程(或python 进程，具体取决于我调用测试运行程序的方式)以终止它。 (如果我追加

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

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

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

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

Answer 1

长话短说

打电话

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?

而不是deleting ResultProxy result，显式close()它在你的 finally 子句中:

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 持有的所有行和表锁。

要避免嵌套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()它。如果结果的行数(可能出乎意料地)多于您要获取的行数，则结果将保持打开状态并继续持有可能阻止执行以下清理的锁。

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