test_connection.py 文件源码

def test_close(engine):
    conn = await engine.connect()
    assert not conn.closed

    result = await conn.execute(select([1]))
    assert await result.scalar() == 1

    await conn.close()
    assert conn.closed

    with pytest.raises(StatementError) as exc:
        await conn.execute(select([1]))
    assert "This Connection is closed" in str(exc)