capsys not implementing fileno, patches stream object causing io error

[tonybaloney]

Pytest 7.1.0
plugins: asyncio-0.19.0, repeat-0.9.1

When using the capsys fixture, starting an async StreamWriter using the stdout or stderr pipes fails under test and works in regular Python.
It looks like this is because the capsys fixture sets different behaviours to the regular pipe object

import pytest
import asyncio
import sys


@pytest.mark.asyncio
async def test_stream_handler_write(event_loop, capsys):
    w_transport, w_protocol = await event_loop.connect_write_pipe(
        asyncio.streams.FlowControlMixin, sys.stdout
    )
    stdout_writer = asyncio.StreamWriter(w_transport, w_protocol, None, event_loop)
    assert stdout_writer
    stdout_writer.write(b"test")
    await stdout_writer.drain()
    stdout_writer.close()    


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_stream_handler_write(loop, None))

python test_repro.py
/Users/anthonyshaw/projects/picologging/test_repro.py:19: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()
test%                                                            

python -m pytest test_repro.py
================================================================================= test session starts ==================================================================================
platform darwin -- Python 3.10.4, pytest-7.1.2, pluggy-1.0.0
rootdir: /Users/anthonyshaw/projects/picologging
plugins: asyncio-0.19.0, repeat-0.9.1
asyncio: mode=strict
collected 1 item                                                                                                                                                                       

test_repro.py F                                                                                                                                                                  [100%]

======================================================================================= FAILURES =======================================================================================
______________________________________________________________________________ test_stream_handler_write _______________________________________________________________________________

event_loop = <_UnixSelectorEventLoop running=False closed=False debug=False>, capsys = <_pytest.capture.CaptureFixture object at 0x10382ab20>

    @pytest.mark.asyncio
    async def test_stream_handler_write(event_loop, capsys):
>       w_transport, w_protocol = await event_loop.connect_write_pipe(
            asyncio.streams.FlowControlMixin, sys.stdout
        )

test_repro.py:8: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../.pyenv/versions/3.10.4-debug/lib/python3.10/asyncio/base_events.py:1578: in connect_write_pipe
    transport = self._make_write_pipe_transport(pipe, protocol, waiter)
../../.pyenv/versions/3.10.4-debug/lib/python3.10/asyncio/unix_events.py:193: in _make_write_pipe_transport
    return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <[AttributeError("'_UnixWritePipeTransport' object has no attribute '_closing'") raised in repr()] _UnixWritePipeTransport object at 0x1037fe580>
loop = <_UnixSelectorEventLoop running=False closed=False debug=False>, pipe = <_io.TextIOWrapper encoding='UTF-8'>, protocol = <asyncio.streams.FlowControlMixin object at 0x1038294a0>
waiter = <Future pending>, extra = None

    def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
        super().__init__(extra, loop)
        self._extra['pipe'] = pipe
        self._pipe = pipe
>       self._fileno = pipe.fileno()
E       io.UnsupportedOperation: fileno

../../.pyenv/versions/3.10.4-debug/lib/python3.10/asyncio/unix_events.py:589: UnsupportedOperation
=============================================================================== short test summary info ================================================================================
FAILED test_repro.py::test_stream_handler_write - io.UnsupportedOperation: fileno
================================================================================== 1 failed in 1.01s ===================================================================================
Exception ignored in: <function _UnixWritePipeTransport.__del__ at 0x103716e60>
Traceback (most recent call last):
  File "/Users/anthonyshaw/.pyenv/versions/3.10.4-debug/lib/python3.10/asyncio/unix_events.py", line 749, in __del__
  File "/Users/anthonyshaw/.pyenv/versions/3.10.4-debug/lib/python3.10/asyncio/unix_events.py", line 626, in __repr__
AttributeError: '_UnixWritePipeTransport' object has no attribute '_closing'

[kmille]

I have a smiliar issue. Anything tips how to deal with it? Would be very important for my project to catch stdout...

      def test_run(self, capsys):
          import subprocess
          import sys 
          p = subprocess.Popen(["echo", "123"], stdout=sys.stdout)
          p.wait()
          stdout = capsys.readouterr().out

raises

/usr/lib/python3.11/subprocess.py:992: in __init__
    errread, errwrite) = self._get_handles(stdin, stdout, stderr)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Popen: returncode: None args: ['echo', '123']>, stdin = None, stdout = <_io.TextIOWrapper encoding='UTF-8'>, stderr = None

    def _get_handles(self, stdin, stdout, stderr):
        """Construct and return tuple with IO objects:
        p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite
...
            elif stdout == DEVNULL:
                c2pwrite = self._get_devnull()
            elif isinstance(stdout, int):
                c2pwrite = stdout
            else:
                # Assuming file-like object
>               c2pwrite = stdout.fileno()
E               io.UnsupportedOperation: fileno

/usr/lib/python3.11/subprocess.py:1723: UnsupportedOperation

[RonnyPfannschmidt]

@kmille sys capture will never support subprocesses, please use a supported mechanism

[RonnyPfannschmidt]

Closing this as not planned, sys capture uses bytes/stringio and will never support file no

Use fdcapture if a file number is needed

[kmille]

Can you give an example or docs about fdcapture? Sorry it's not clear to me.
Update: this worked for me.

[RonnyPfannschmidt]

@kmille what its completely and utterly surreal to me is that you want to pass a capture to a sub-process call instead of just using something like subprocess.run(..., capture_output=True) which manages pipes, contention control, buffering and more returning a CompletedProcess with all data correct and intact