Launching the Subprocess

High Level Design

We will implement LineMonitor as follows:

  1. a LineMonitor sill launch the subprocess using the subprocess Python standard library.

  2. It will attach a pseudo-terminal to said subprocess (using pty). If you don’t know too much about what a pseudo-terminal is - don’t worry about it, I don’t either.

Essentially it’s attaching the subprocess’s input and output streams to the father process. Another way of doing this is using pipes, but there are some technical advantages to using a pseudo-terminal.

  1. it will monitor the terminal using poll() from the standard Python library’s select module. This call allows to you check if the pseudo-terminal has any data available to read (that is, check if the subprocess has written some output).

  2. when data is available, we will read it line by line, and send it to the registered callbacks.

Let’s start by working on the launching a subprocess with an attached pseudo-terminal.

Implementation

First step is to launch the subprocess with an attached pseudo-terminal. Let’s write a test for that. We want to enforce, using Testix, that subprocess.Popen() is called with appropriate arguments.

If the following paragraph is confusing, don’t worry - things will become clearer after you see it all working.

Since Testix’s Scenario object only tracks Testix Fake objects, we must somehow fool the LineMonitor to use a Fake('subprocess') object instead of the actual subprocess module. We need to do the same for the pty module.

There’s more than one way of doing this, but here we will use Testix’s helper fixture, patch_module.

 1from testix import *
 2import pytest
 3import line_monitor
 4
 5@pytest.fixture
 6def override_imports(patch_module):
 7    patch_module(line_monitor, 'subprocess') # this replaces the subprocess object inside line_monitor with a Fake("subprocess") object
 8    patch_module(line_monitor, 'pty') # does the same for the pty module
 9
10def test_lauch_subprocess_with_pseudoterminal(override_imports):
11    tested = line_monitor.LineMonitor()
12    with Scenario() as s:
13        s.pty.openpty() >> ('write_to_fd', 'read_from_fd')
14        s.subprocess.Popen(['my', 'command', 'line'], stdout='write_to_fd', close_fds=True)
15
16        tested.launch_subprocess(['my', 'command', 'line'])

What’s going on here?

  1. First, we use patch_module to mock imported modules subprocess and pty, as described above. Note that our test function depends on override_imports to make everything work.

  2. In our Scenario we demand two things:

    • That our code calls pty.openpty() to create a pseudo-terminal and obtain its two file descriptors.

    • That our code then launch a subprocess and point its stdout to the write file-descriptor of the pseudo-terminal (we also demand close_fds=True wince we want to fully specify our subprocess’s inputs and outputs).

  3. Finally, we call our .launch_subprocess() method to actually do the work - we can’t hope that our code meet our expectations if we never actually call it, right?

A few points on this:

  1. See how we first write our expectations and only then call the code to deliver on these expectations. This is one way Testix pushes you into a Test Driven mindset.

  2. In real life, pty.openpty() returns two file descriptors - which are integers. In our test, we made this call return two strings.

We could have, e.g. define two constants equal to some integers, e.g. WRITE_FD=20 and READ_FD=30 and used those - but it wouldn’t really matter and would make the test more cluttered. Technically, what’s important is that openpty() returns a tuple and we demand that the first item in this tuple is passed over to the right place in the call to Popen(). Some people find fault with this style. Personally I think passing strings around (recall that in Python strings are immutable) where all you’re testing is moving around objects - is a good way to make a readable test.

Failing the Test

Remember, when practicing TDD you should always fail your tests first, and make sure they fail properly.

So let’s see some failures! Let’s see some RED!

Running this test with the skeleton implementation we have for LineMonitor results in:

E       Failed:
E       testix: ScenarioException
E       testix details:
E       Scenario ended, but not all expectations were met. Pending expectations (ordered): [pty.openpty(), subprocess.Popen(['my', 'command', 'line'], stdout = 'write_to_fd', close_fds = True)]

Very good, our tests fails as it should: the test expects, e.g. openpty() to be called, but our current implementation doesn’t call anything - so the test fails in disappointment.

Now that we have our RED, let’s get to GREEN.

Passing the Test

Let’s write some code that makes the test pass:

 1import subprocess
 2import pty
 3
 4class LineMonitor:
 5    def register_callback(self, callback):
 6        pass
 7
 8    def launch_subprocess(self, *popen_args, **popen_kwargs):
 9        write_to, read_from = pty.openpty()
10        popen_kwargs['stdout'] = write_to
11        popen_kwargs['close_fds'] = True
12        subprocess.Popen(*popen_args, **popen_kwargs)
13
14    def monitor(self):
15        pass

Running our test with this code produces

test_line_monitor.py::test_lauch_subprocess_with_pseudoterminal PASSED

Finally, we see some GREEN!

Usually we will now take the time to REFACTOR our code, but we have so little code at this time that we’ll skip it for now.

OK, we have our basic subprocess with a pseudo-terminal - now’s the time to test for and implement actually monitoring the output.