Writing pytest testcases for asyncio with streams

Question

I am trying to write a pytest testcase for a asyncio function which does read the output streams (stderr/stdout) and modifies the lines. The function I want to test (which again gets called inside asyncio.gather) is as shown below:

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}')

The pytest testcase I wrote is as follows:

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

However, when I execute this pytest I encounter AttributeError: 'str' object has no attribute 'read' . How to test asyncio coroutines while dealing with stdout/stderr streams?

score 3 · Accepted Answer · answered Sep 09 '21 at 01:52

StringIO does not have coroutine methods for read, so you can't mock this and have it work with your watch coroutine function (calling getvalue on the StringIO instance also just passes in the string written to stdout, which explains the error you get). Assuming that the stream in your watch function is an instance of StreamReader, you can just create an asyncio StreamReader instance in your test and use the feed_data method to write something to the stream. Then you can pass this in to watch. You can then use the capsys fixture included with Pytest to capture what watch writes to stdout.

Below is an updated version of your code that passes as a standalone:

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

Writing pytest testcases for asyncio with streams

1 Answers1