Recap

Let’s recap a bit on what we’ve been doing in this tutorial.

We started with a test for the basic subprocess-launch behaviour, got to RED, implemented the code, and got to GREEN.

Next, when moving to implement the actual output monitoring behaviour, we kept the first test, and added a new one. This is very important in TDD - the old test keeps the old behaviour intact - if, when implementing the new behaviour we break the old one - we will know.

When working with Testix, you are encourage to track all your mocks (Fake objects) very precisely. This effectively made us refactor the launch-process test scenario into a launch_scenario() helper function, since you must launch a subprocess before monitoring it.

We also saw that adding a call to open made the original launch-process test fail as well as the new monitor test. This makes sense, since the launching behaviour now includes a call to open that it didn’t before - and the code doesn’t support that yet, so the test fails.

Another thing we ran into is that sometimes we get GREEN even when we wanted RED. This should make you uneasy - it usually means that the test is not really testing what you think it is. In our case, however, it was just because an edge case which we added a test for was already covered by our existing code. When that happens, strict TDD isn’t really possible - and you need to revert to making sure that if you break the code on purpose, it breaks the test in the proper manner.

YAGNI

Another thing to notice, is that the call to Popen is simply

subprocess.Popen(*popen_args, **popen_kwargs)

And not, for example,

self._process = subprocess.Popen(*popen_args, **popen_kwargs)

Why didn’t we save the subprocess in an instance variable? Working TDD makes us want to get to GREEN - no more, no less. Since we don’t need to store the subprocess to pass the test, we don’t do it.

Let me repeat that for you: if we don’t need it to pass the test, we don’t do it.

You might say “but we need to hold on to the subprocess to control it, see if it’s still alive, or kill it”.

Well, maybe we do. If that’s what we really think, we should express this need in a test - make sure it’s RED, and then write the code to make it GREEN.

This is one particular way of implementing the YAGNI principle - if you’re not familiar with it, you should take the time to read about it.

Upholding the YAGNI requires a special kind of discipline, and TDD and Testix in particular, helps us achieve it.

Code Recap

Before we continue, here is the current state of our unit test and code.

The test:

from testix import *
import pytest
import line_monitor

@pytest.fixture
def override_imports(patch_module):
    patch_module(line_monitor, 'subprocess')
    patch_module(line_monitor, 'pty')
    patch_module(line_monitor, 'open')

def launch_scenario(s):
    s.pty.openpty() >> ('write_to_fd', 'read_from_fd')
    s.open('read_from_fd', encoding='latin-1') >> Fake('reader')
    s.subprocess.Popen(['my', 'command', 'line'], stdout='write_to_fd', close_fds=True)

def test_lauch_subprocess_with_pseudoterminal(override_imports):
    tested = line_monitor.LineMonitor()
    with Scenario() as s:
        launch_scenario(s)
        tested.launch_subprocess(['my', 'command', 'line'])

def test_receive_output_lines_via_callback(override_imports):
    tested = line_monitor.LineMonitor()
    with Scenario() as s:
        launch_scenario(s)
        tested.launch_subprocess(['my', 'command', 'line'])

        s.reader.readline() >> 'line 1'
        s.my_callback('line 1')
        s.reader.readline() >> 'line 2'
        s.my_callback('line 2')
        s.reader.readline() >> 'line 3'
        s.my_callback('line 3')
        s.reader.readline() >> Throwing(loop_breaker.LoopBreaker) # this tells the Fake('reader') to raise an instance of TestixLoopBreaker()

        tested.register_callback(Fake('my_callback'))
        with pytest.raises(loop_breaker.LoopBreaker):
            tested.monitor()

and the code that passes it

import subprocess
import pty

class LineMonitor:
    def __init__(self):
        self._callback = None

    def register_callback(self, callback):
        self._callback = callback

    def launch_subprocess(self, *popen_args, **popen_kwargs):
        write_to, read_from = pty.openpty()
        popen_kwargs['stdout'] = write_to
        popen_kwargs['close_fds'] = True
        self._reader = open(read_from, encoding='latin-1')
        subprocess.Popen(*popen_args, **popen_kwargs)

    def monitor(self):
        while True:
            line = self._reader.readline()
            self._callback(line)