Mechanism for testing long-running processes

[simonw]

The runner.invoke(...) mechanism has a limitation at the moment: it runs the command to completion before allowing you to run assertions against the returned Result object.

I'm working on a tool that streams from standard input - and I'd like to run some assertions on various pieces of state in the middle of the stream. So I'd like to be able to do this:

• Start the tool running
• Pass a line of text to standard input
• Make some assertions
• Pass another line of text to standard input
• Make some more assertions
• Send an indication that "standard input has ended" such that the tool exits
• Run some final assertions

I can't see a way to do this using CliRunner at the moment. Is there a known pattern for this already?

[simonw]

I think what I'm really looking for here is a way to start a Click command running from a test in a way that lets me send to its stdin and read from its stdout.

[simonw]

I figured out a pattern that solved my problem using subprocess.Popen - details here: https://til.simonwillison.net/pytest/test-click-app-with-streaming-input

[davidism]

invoke calls isolation(input), which calls make_input_stream, which always returns a new BytesIO object. There's probably a way to change that so it preserves an existing BytesIO value, that's where I'd start.

[davidism]

Never mind, that already does the right thing, the issue is that invoke runs the command to completion (or error), and prompt/input raises EOFError if the input stream is exhausted. How would someone test streaming stdin with plain Python?

[davidism]

I can't think of a clean way to do this. I wrote a little __getattr__ wrapper around a stream, and the last call that is looked up is read1, but overriding it reveals that it's not what raises EOFError, so I'm not sure what would need to be done to intercept that and wait instead.

Reading a stream is a blocking operation, so even if we were able to make the stream writeable, you'd need to set up a thread so that input could wait without blocking your test code from writing more to the stream.

Another possibility is implementing some API that handles I/O as a generator of events. You would call session = runner.invoke_stream(), iterate it to get output and wait events, then call session.send("value") to send more input. But that's really complex, and would require extensive changes to Click.

Unless you have a better suggestion, I'm going to close this and say your solution with subprocess is good enough for this case.

[simonw]

I'm not blocked on this since I found a good enough workaround, but I'd be delighted to see Click come up with an official solution that's as pleasant to use as the rest of the existing Click testing utilities!

It's likely not worth it if it requires major rearchitecting though.

So I won't be at all disappointed if you close this as wontfix.