在 pytest 中，如何编写返回另一个函数的 AsyncGenerator 的装置？

Question

以下作品；这是一个成功使用异步生成器作为固定装置的测试。

from collections.abc import AsyncGenerator

import pytest

@pytest.fixture()
async def fixture() -> AsyncGenerator[str, None]:
    yield "a"

@pytest.mark.asyncio
async def test(fixture: str):
    assert fixture[0] == "a"

但是，假设我想fixture()返回AsyncGenerator由其他函数生成的结果。在这种情况下，我收到错误：

from collections.abc import AsyncGenerator

import pytest

async def _fixture() -> AsyncGenerator[str, None]:
    yield "a"

@pytest.fixture()
async def fixture() -> AsyncGenerator[str, None]:
    return _fixture()

@pytest.mark.asyncio
async def test(fixture: str):
    assert fixture[0] == "a"

错误是：

>     assert fixture[0] == "a"
E     TypeError: 'async_generator' object is not subscriptable

我缺少什么？

Answer 1

首先，示例代码应该可以工作，但不应该（并且不适合我）。它产生与异步生成器相同的TypeError结果。fixture根据（相当糟糕的）文档pytest-asyncio：

异步装置的定义就像普通的 pytest 装置一样，除了它们应该用@pytest_asyncio.fixture.

因此，除非您专门配置asyncio_mode = auto，否则要使第一个测试正常工作，您需要将代码更改为：

from collections.abc import AsyncGenerator

import pytest
import pytest_asyncio


@pytest_asyncio.fixture
async def fixture() -> AsyncGenerator[str, None]:
    yield "abcdef"


@pytest.mark.asyncio
async def test(fixture: str):
    assert fixture[0] == "a"

fixture第二个例子也是如此。

其次，_fixture只是一个常规的旧异步生成器函数。如果没有奇怪和晦涩的 pytest 魔法，调用_fixture()只会返回一个异步生成器对象（如返回类型注释所正确指示的那样AsyncGenerator）。

您只是从fixture函数中返回它，而不是执行组合。对于普通（非async）发电机，您会yield from在这种情况下这样做。但这对于异步生成器不起作用（PEP 525），因此您需要执行循环async for和yield元素：

from collections.abc import AsyncGenerator

import pytest
import pytest_asyncio


async def _fixture() -> AsyncGenerator[str, None]:
    yield "a"


@pytest_asyncio.fixture
async def fixture() -> AsyncGenerator[str, None]:
    async for item in _fixture():
        yield item


@pytest.mark.asyncio
async def test(fixture: str):
    assert fixture[0] == "a"

旁白：这是我不太喜欢的原因之一pytest。这种奇怪的坚持有时会掩盖逻辑。

常规固定装置只是运行，它们的返回值被传递给请求该固定装置的测试函数。但生成器显然不够好，因此我们没有将生成器本身传递给测试函数，而是从中获取第一项，然后剩下的就是“终结器”，我猜。（参见产量固定装置）

我想，如果我想要在测试函数中使用实际的生成器，我必须完全按照您（错误地）在上面所做的操作，即单独定义它，然后从固定功能返回它。

test因此，如果您更改函数以使用它所传递的异步生成器，您也可以使原始代码正常工作。由于它只产生一次，因此将内容放入列表中会将您"a"作为其第一个元素：

...

@pytest_asyncio.fixture
async def fixture() -> AsyncGenerator[str, None]:
    return _fixture()

@pytest.mark.asyncio
async def test(fixture: AsyncGenerator[str, None]):
    items = [item async for item in fixture]
    assert items[0] == "a"