Блог

Я пытаюсь написать тестовый набор pytest для функции asyncio, которая считывает выходные потоки (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
 

Однако, когда я выполняю этот тест, я сталкиваюсь AttributeError: 'str' object has no attribute 'read' с этим . Как протестировать сопрограммы asyncio при работе с потоками stdout/stderr?

StringIO не имеет методов сопрограммы для read , поэтому вы не можете издеваться над этим и заставить его работать с функцией сопрограммы watch (вызов getvalue StringIO экземпляра также просто передает строку, записанную в stdout, что объясняет полученную ошибку). Предполагая , что поток в вашей функции наблюдения является экземпляром StreamReader , вы можете просто создать экземпляр asyncio StreamReader в своем тесте и использовать feed_data метод для записи чего-либо в поток. Затем вы можете передать это watch кому-нибудь . Затем вы можете использовать capsys приспособление, входящее в состав Pytest, для записи того, что 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 stdoutn"

    stream = asyncio.StreamReader()
    stream.feed_data(b'This is stdoutn')
    stream.feed_eof()

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