python - 在 fixture 中包含代码后，带有异步的 pytest 中的 AttributeError

Question

我需要测试我的电报机器人。为此，我需要创建客户端用户来询问我的机器人。我找到了 telethon可以做到的图书馆。首先，我编写了一个代码示例，以确保授权和连接正常工作，并向自己发送测试消息(省略导入):

api_id = int(os.getenv("TELEGRAM_APP_ID"))
api_hash = os.getenv("TELEGRAM_APP_HASH")
session_str = os.getenv("TELETHON_SESSION")

async def main():
    client = TelegramClient(
        StringSession(session_str), api_id, api_hash,
        sequential_updates=True
    )
    await client.connect()
    async with client.conversation("@someuser") as conv:
        await conv.send_message('Hey, what is your name?')


if __name__ == "__main__":
    asyncio.run(main())

@someuser (me) 成功接收消息。好的，现在我根据上面的代码创建一个带有 fixture 的测试:

api_id = int(os.getenv("TELEGRAM_APP_ID"))
api_hash = os.getenv("TELEGRAM_APP_HASH")
session_str = os.getenv("TELETHON_SESSION")

@pytest.fixture(scope="session")
async def client():
    client = TelegramClient(
        StringSession(session_str), api_id, api_hash,
        sequential_updates=True
    )
    await client.connect()
    yield client
    await client.disconnect()


@pytest.mark.asyncio
async def test_start(client: TelegramClient):
    async with client.conversation("@someuser") as conv:
        await conv.send_message("Hey, what is your name?")

运行pytest后收到错误:

AttributeError: 'async_generator' 对象没有属性 'conversation'

似乎 client 对象从 client fixture 返回“错误”条件。这是 print(dir(client)):

['__aiter__', '__anext__', '__class__', '__class_getitem__', '__del__', '__delattr__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__name__', '__ne__', '__new__', '__qualname__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', 'aclose', 'ag_await', 'ag_code', 'ag_frame', 'ag_running', 'asend', 'athrow']

我在哪里从 fixture 中的生成器中丢失了“正确的”客户端对象？

Answer 1

根据文档 https://pypi.org/project/pytest-asyncio/#async-fixtures 在异步 fixture 中使用 @pytest_asyncio.fixture 装饰器.

像这样:

import pytest_asyncio

@pytest_asyncio.fixture(scope="session")
async def client():
    ...