Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Open
blcdb opened this issue Aug 13, 2024 · 1 comment
Open

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

blcdb opened this issue Aug 13, 2024 · 1 comment

Comments

@blcdb
Copy link

blcdb commented Aug 13, 2024

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
Copy link
Owner

Thank you for reporting. I have never used selectors, so I didn't incorporate them into a plugin. Feel free to submit a PR to fix this issue. Otherwise, I'll try to work on it in the future, but I can't tell you when it will be.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

2 participants