0

我正在尝试为 asyncio 函数编写一个 pytest 测试用例，该函数确实读取输出流（stderr/stdout）并修改行。我要测试的函数（再次被调用 inside 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 协程？

4

1 回答 1

0

StringIO没有协程方法read，所以你不能模拟它并让它与你的监视协程函数一起工作（调用getvalue实例StringIO也只是传入写入标准输出的字符串，这解释了你得到的错误）。假设您的 watch 函数中的流是 的一个实例StreamReader，您可以StreamReader在测试中创建一个 asyncio 实例并使用该feed_data方法向流中写入一些内容。然后你可以把它传递给watch. 然后，您可以使用capsysPytest 附带的夹具来捕获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
于 2021-09-09T01:52:58.537 回答