Capture output from subprocess.call that I have no control over

Question

I'm testing a piece of Python code that uses subprocess.call(), so I have no control over that function call. I need to capture the output from that system call to do assertions. I tried to set os.stdout to a StringIO object, but that doesn't capture the system call outputs. How do I solve this problem? Here's my code so far:

Code to test (I have no control over this):

def runFile(filename):
    check_call("./" + filename)

My attempt to capture system call output:

import StringIO
oldout = sys.stdout
try:
    sys.stdout = StringIO.StringIO()
    runFile("somefile")
    output = sys.stdout.getvalue()
    assert output == "expected-output"
finally:
    sys.stdout = oldout

Answer 1

You could redirect your own stdout by

rd, wr = os.pipe()
oldstdout = os.dup(1)
os.dup2(wr, 1)
os.close(wr)

and then read from rd while the external process is running.

This might not work, as the external process might write more than fits in the pipe buffer. In this case, you would have to spawn a reading thread.

Afterwards, you restore the old state with

os.dup2(oldstdout, 1)
os.close(oldstdout)
os.close(rd)

and continue normally.

Answer 2

The subprocess module directly writes to the output stream using os.write, so changing sys.stdout won't do anything.

That's why you generally need to specify the output stream used with the subprocess module using the ...([cmd, arg, ...], stdout=output_file, ...) notation, and why you can't use a StringIO as output file as it has no fileno to write to.

So the only way to achieve what you want to do ist o monkeypatch subproces.check_call before importing the module you need to test, so that it works a little more like check_output.

For example:

import subprocess

def patched_call(*popenargs, **kwargs):
    if 'stdout' in kwargs:
        raise ValueError('stdout argument not allowed, it will be overridden.')
    process = subprocess.Popen(*popenargs, stdout=subprocess.PIPE, **kwargs)
    output, unused_err = process.communicate()
    # do something with output here
    retcode = process.poll()
    if retcode:
        cmd = kwargs.get("args")
        if cmd is None:
            cmd = popenargs[0]
        raise CalledProcessError(retcode, cmd, output=output)
    return retcode

subprocess.check_call = patched_call