FakeProcess stdout and stderr doesn't work with selectors

Hello,

In order to have a the logs of a time lasting command printed at the user while the command is still running i did this code :

def exec_command(
    command: str, log_stdout: bool = True
) -> int:
    try:
        process = subprocess.Popen(
            shlex.split(command),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        sel = selectors.DefaultSelector()
        sel.register(process.stdout, selectors.EVENT_READ)
        sel.register(process.stderr, selectors.EVENT_READ)
        while True:
            should_break = False
            for key, _ in sel.select():
                data = key.fileobj.read1().decode()
                if not data:
                    should_break = True
                    break
                if key.fileobj is process.stdout:
                    if log_stdout:
                        logger.info(data)
                else:
                    logger.error(data)
            if should_break:
                break
        process.poll()
        return process.returncode

    except Exception as e:
        logger.error("An error occured during the run of the command : " + command)
        raise e

It's a solution that avoid to have thread to be able to read both stdout and stderr while the command is still running.

I tried to test my code whith pytest-subprocess with a test similar to this :

def test_exec_comande(fp: FakeProcess):
    fp.register(
        ["test"], stdout="mocked", stderr="mocked err", returncode=1
    )
    exec_command("test")

but the test failed because selectors.register need an object like an io.BufferedReader (that is returned by the subprocess.Popen() upward.) However fp.register is unable to take an io.BufferedReader for stdout and stderr.

Do you think it is possible to allow this stdout and stderr mocking ?

aklajnert / pytest-subprocess

FakeProcess stdout and stderr doesn't work with selectors #166