Testing use of the subprocess package

When using the subprocess package there are two approaches to testing:

  • Have your tests exercise the real processes being instantiated and used.

  • Mock out use of the subprocess package and provide expected output while recording interactions with the package to make sure they are as expected.

While the first of these should be preferred, it means that you need to have all the external software available everywhere you wish to run tests. Your tests will also need to make sure any dependencies of that software on an external environment are met. If that external software takes a long time to run, your tests will also take a long time to run.

These challenges can often make the second approach more practical and can be the more pragmatic approach when coupled with a mock that accurately simulates the behaviour of a subprocess. MockPopen is an attempt to provide just such a mock.

Note

To use MockPopen, you must have the mock package installed or be using Python 3.3 or later.

Warning

Previous versions of this mock made use of mock_calls. These are deceptively incapable of recording some information important in the use of this mock, so please switch to making assertions about all_calls and calls instead.

Example usage

As an example, suppose you have code such as the following that you need to test:

from subprocess import Popen, PIPE


def my_func():
    process = Popen(['svn', 'ls', '-R', 'foo'], stdout=PIPE, stderr=PIPE)
    out, err = process.communicate()
    if process.returncode:
        raise RuntimeError('something bad happened')
    return out

Tests that exercise this code using MockPopen could be written as follows:

from unittest import TestCase

from testfixtures.mock import call
from testfixtures import Replacer, ShouldRaise, compare
from testfixtures.popen import MockPopen, PopenBehaviour


class TestMyFunc(TestCase):

    def setUp(self):
        self.Popen = MockPopen()
        self.r = Replacer()
        self.r.replace(dotted_path, self.Popen)
        self.addCleanup(self.r.restore)

    def test_example(self):
        # set up
        self.Popen.set_command('svn ls -R foo', stdout=b'o', stderr=b'e')

        # testing of results
        compare(my_func(), b'o')

        # testing calls were in the right order and with the correct parameters:
        process = call.Popen(['svn', 'ls', '-R', 'foo'], stderr=PIPE, stdout=PIPE)
        compare(Popen.all_calls, expected=[
            process,
            process.communicate()
        ])

    def test_example_bad_returncode(self):
        # set up
        Popen.set_command('svn ls -R foo', stdout=b'o', stderr=b'e',
                          returncode=1)

        # testing of error
        with ShouldRaise(RuntimeError('something bad happened')):
            my_func()

Passing input to processes

If your testing requires passing input to the subprocess, you can do so by checking for the input passed to communicate() method when you check the calls on the mock as shown in this example:

def test_communicate_with_input(self):
    # setup
    Popen = MockPopen()
    Popen.set_command('a command')
    # usage
    process = Popen('a command', stdout=PIPE, stderr=PIPE, shell=True)
    out, err = process.communicate('foo')
    # test call list
    compare(Popen.all_calls, expected=[
            process.root_call,
            process.root_call.communicate('foo'),
    ])

Note

Accessing .stdin isn’t current supported by this mock.

Reading from stdout and stderr

The stdout and stderr attributes of the mock returned by MockPopen will be file-like objects as with the real Popen and can be read as shown in this example:

def test_read_from_stdout_and_stderr(self):
    # setup
    Popen = MockPopen()
    Popen.set_command('a command', stdout=b'foo', stderr=b'bar')
    # usage
    process = Popen('a command', stdout=PIPE, stderr=PIPE, shell=True)
    compare(process.stdout.read(), b'foo')
    compare(process.stderr.read(), b'bar')

Warning

While these streams behave a lot like the streams of a real Popen object, they do not exhibit the deadlocking behaviour that can occur when the two streams are read as in the example above. Be very careful when reading stdout and stderr and consider using communicate() instead.

Writing to stdin

If you set stdin=PIPE in your call to Popen then the stdin attribute of the mock returned by MockPopen will be a mock and you can then examine the write calls to it as shown in this example:

def test_write_to_stdin(self):
    # setup
    Popen = MockPopen()
    Popen.set_command('a command')
    # usage
    process = Popen('a command', stdin=PIPE, shell=True)
    process.stdin.write('some text')
    process.stdin.close()
    # test call list
    compare(Popen.all_calls, expected=[
        process.root_call,
        process.root_call.stdin.write('some text'),
        process.root_call.stdin.close(),
    ])

Specifying the return code

Often code will need to behave differently depending on the return code of the launched process. Specifying a simulated response code, along with testing for the correct usage of wait(), can be seen in the following example:

def test_wait_and_return_code(self):
    # setup
    Popen = MockPopen()
    Popen.set_command('a command', returncode=3)
    # usage
    process = Popen('a command')
    compare(process.returncode, None)
    # result checking
    compare(process.wait(), 3)
    compare(process.returncode, 3)
    # test call list
    compare(Popen.all_calls, expected=[
        call.Popen('a command'),
        call.Popen('a command').wait(),
    ])

Checking for signal sending

Calls to send_signal(), MockPopenInstance.terminate() and MockPopenInstance.kill() are all recorded by the mock returned by MockPopen but otherwise do nothing as shown in the following example, which doesn’t make sense for a real test of sub-process usage but does show how the mock behaves:

def test_send_signal(self):
    # setup
    Popen = MockPopen()
    Popen.set_command('a command')
    # usage
    process = Popen('a command', stdout=PIPE, stderr=PIPE, shell=True)
    process.send_signal(0)
    # result checking
    compare(Popen.all_calls, expected=[
        process.root_call,
        process.root_call.send_signal(0),
    ])

Polling a process

The poll() method is often used as part of a loop in order to do other work while waiting for a sub-process to complete. The mock returned by MockPopen supports this by allowing the poll() method to be called a number of times before the returncode is set using the poll_count parameter as shown in the following example:

def test_poll_until_result(self):
    # setup
    Popen = MockPopen()
    Popen.set_command('a command', returncode=3, poll_count=2)
    # example usage
    process = Popen('a command')
    while process.poll() is None:
        # you'd probably have a sleep here, or go off and
        # do some other work.
        pass
    # result checking
    compare(process.returncode, 3)
    compare(Popen.all_calls, expected=[
        process.root_call,
        process.root_call.poll(),
        process.root_call.poll(),
        process.root_call.poll(),
    ])

Different behaviour on sequential processes

If your code needs to call the same command but have different behaviour on each call, then you can pass a callable behaviour like this:

def test_multiple_responses(self):
    # set up
    behaviours = [
        PopenBehaviour(stderr=b'e', returncode=1),
        PopenBehaviour(stdout=b'o'),
    ]

    def behaviour(command, stdin):
        return behaviours.pop(0)

    self.Popen.set_command('svn ls -R foo', behaviour=behaviour)

    # testing of error:
    with ShouldRaise(RuntimeError('something bad happened')):
        my_func()
    # testing of second call:
    compare(my_func(), b'o')

If you need to keep state across calls, such as accumulating stdin or failing for a configurable number of calls, then wrap that behaviour up into a class:

class CustomBehaviour(object):

    def __init__(self, fail_count=1):
        self.fail_count = fail_count

    def __call__(self, command, stdin):
        while self.fail_count > 0:
            self.fail_count -= 1
            return PopenBehaviour(stderr=b'e', returncode=1)
        return PopenBehaviour(stdout=b'o')

This can then be used like this:

def test_count_down(self):
    # set up
    self.Popen.set_command('svn ls -R foo', behaviour=CustomBehaviour())
    # testing of error:
    with ShouldRaise(RuntimeError('something bad happened')):
        my_func()
    # testing of second call:
    compare(my_func(), b'o')

Using default behaviour

If you’re testing something that needs to make many calls to many different commands that all behave the same, it can be tedious to specify the behaviour of each with set_command. For this case, MockPopen has the set_default method which can be used to set the behaviour of any command that has not been specified with set_command as shown in the following example:

def test_default_behaviour(self):
    # set up
    self.Popen.set_default(stdout=b'o', stderr=b'e')

    # testing of results
    compare(my_func(), b'o')

    # testing calls were in the right order and with the correct parameters:
    root_call = call.Popen(['svn', 'ls', '-R', 'foo'],
                           stderr=PIPE, stdout=PIPE)
    compare(Popen.all_calls, expected=[
        root_call,
        root_call.communicate()
    ])

Tracking multiple simultaneous processes

Conversely, if you’re testing something that spins up multiple subprocesses and manages their simultaneous execution, you will want to explicitly define the behaviour of each process using set_command and then make assertions about each process using all_calls.

For example, suppose we wanted to test this function:

def process_in_batches(n):
    processes = []
    for i in range(n):
        processes.append(Popen('process --batch='+str(i),
                               stdout=PIPE, stderr=PIPE, shell=True))
    total = 0
    for process in processes:
        out, err = process.communicate()
        total += int(out)
    return total

Then you could test it as follows:

def test_multiple_processes(self):
    # set up
    self.Popen.set_command('process --batch=0', stdout=b'42')
    self.Popen.set_command('process --batch=1', stdout=b'13')

    # testing of results
    compare(process_in_batches(2), expected=55)

    # testing of process management:
    p1 = call.Popen('process --batch=0', shell=True, stderr=PIPE, stdout=PIPE)
    p2 = call.Popen('process --batch=1', shell=True, stderr=PIPE, stdout=PIPE)
    compare(Popen.all_calls, expected=[
        p1,
        p2,
        p1.communicate(),
        p2.communicate(),
    ])

Note that the order of all calls is explicitly recorded. If the order of these calls is non-deterministic due to your method of process management, you will need to do more work and be very careful when testing.