使用pytest编写异步流（asyncio with streams）的测试用例

Question

我试图编写一个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流的同时测试asyncio协程？

- SKP

Answer 1

StringIO没有协程方法可以用于read，因此您无法模拟它并使其与您的监视协程函数一起工作（在StringIO实例上调用getvalue也只是将写入stdout的字符串传递给该函数，这解释了您收到的错误）。假设您的watch函数中的流是StreamReader的实例，则可以在测试中创建一个asyncio StreamReader实例，并使用feed_data方法向流中写入内容。然后，您可以将其传递给watch。然后，您可以使用Pytest中包含的capsys装置来捕获watch写入stdout的内容。

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

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

- Matt Fowler