使用流为 asyncio 编写 pytest 测试用例

Question

我正在尝试为 asyncio 函数编写一个 pytest 测试用例，它确实读取输出流 (stderr/stdout) 并修改行。我要测试的函数（再次在 asyncio.gather 中调用）如下所示：

import asyncio

async def watch(stream):

    while True:
        lines = await stream.read(2**16)
        if not lines or lines == "":
            break

        lines = lines.strip().split("\n")
        for line in lines:
            print(f'myPrefix-{line}')

我写的pytest测试用例如下：

import asyncio
from io import StringIO
import pytest

@pytest.fixture(autouse=True)
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

@pytest.mark.asyncio
async def test_watch(event_loop):
    expected_outcome = "myPrefix-This is stdout"

    def write_something():
        print("This is stdout")

    with patch("sys.stdout", new=StringIO()) as mock_stdout:
        write_something()
        output = await watch(mock_stdout.getvalue())
        assert output.return_value == expected_outcome

但是，当我执行这个 pytest 时遇到 AttributeError: 'str' object has no attribute 'read' 。如何在处理 stdout/stderr 流时测试异步协程？

Answer 1

StringIO 没有 read 的协程方法，所以你不能模拟它并让它与你的 watch 协程函数一起工作（在 [=11 上调用 getvalue =] 实例也只是传递写入标准输出的字符串，这解释了你得到的错误）。假设您的 watch 函数中的流是 StreamReader 的实例，您可以在测试中创建一个 asyncio StreamReader 实例并使用 feed_data 方法向流中写入内容。然后你可以将它传递给 watch。然后，您可以使用 Pytest 附带的 capsys 夹具来捕获 watch 写入标准输出的内容。

以下是作为独立代码通过的更新版本：

import asyncio
import pytest


async def watch(stream):
    while True:
        lines = await stream.read(2 ** 16)
        if not lines or lines == "":
            break

        lines = lines.decode().strip().split("\n") #note the use of decode()
        for line in lines:
            print(f'myPrefix-{line}')


@pytest.fixture(autouse=True)
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()


@pytest.mark.asyncio
async def test_watch(capsys):
    expected_outcome = "myPrefix-This is stdout\n"

    stream = asyncio.StreamReader()
    stream.feed_data(b'This is stdout\n')
    stream.feed_eof()

    await watch(stream)
    captured = capsys.readouterr()
    assert captured.out == expected_outcome

使用流为 asyncio 编写 pytest 测试用例

Writing pytest testcases for asyncio with streams

python-asyncio

stdout

pytest-asyncio

nest-asyncio