title: Writing pytest testcases for asyncio with streams
body: <p>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 <code>asyncio.gather</code>) is as shown below:</p>
  <pre><code>import asyncio
  
  async def watch(stream):
  
      while True:
          lines = await stream.read(2**16)
          if not lines or lines == &quot;&quot;:
              break
  
          lines = lines.strip().split(&quot;\n&quot;)
          for line in lines:
              print(f'myPrefix-{line}')
  </code></pre>
  <p>The pytest testcase I wrote is as follows:</p>
  <pre><code>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 = &quot;myPrefix-This is stdout&quot;
  
      def write_something():
          print(&quot;This is stdout&quot;)
  
      with patch(&quot;sys.stdout&quot;, new=StringIO()) as mock_stdout:
          write_something()
          output = await watch(mock_stdout.getvalue())
          assert output.return_value == expected_outcome
  </code></pre>
  <p>However, when I execute this pytest I encounter <code>AttributeError: 'str' object has no attribute 'read'</code> . How to test asyncio coroutines while dealing with stdout/stderr streams?</p>