Monitoring The Output

Next we want to test the following behaviour: we register a callback with our LineMonitor object using its .register_callback() method, and it calls our callback with each line of output it reads from the pseudo-terminal.

Python streams have a useful .readline() method, so let’s wrap the read file-descriptor of the pseudo-terminal with a stream. It turns out that you can wrap a file descriptor with a simple call to the built-on open() function, so we’ll use that.

Note that we add a new test, leaving the previous one intact. This means that we keep everything we already have working, while we add a test for this new behaviour.

Let’s start by describing a scenario where we read several lines from the pseudo-terminal and demand that they are transferred to our callback.

def test_receive_output_lines_via_callback(override_imports):
    tested = line_monitor.LineMonitor()
    with Scenario() as s:
        s.pty.openpty() >> ('write_to_fd', 'read_from_fd')
        s.open('read_from_fd', encoding='latin-1') >> Fake('reader') # wrapping a binary file descriptor with a text-oriented stream requires an encoding
        s.subprocess.Popen(['my', 'command', 'line'], stdout='write_to_fd', close_fds=True)

        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'

What’s going on here?

1. We add a demand that our code create a Python stream from the pseudo-terminal’s read-descriptor before launching the subprocess.

2. We then call .launch_subprocess() to meet those demands.

3. We describe the “read-from-pseudo-terminal-forwared-to-callback” data flow for 3 consecutive lines.

4. We register a Fake('my_callback') object as our callback - this way, when the code calls the callback, it will be meeting our demands in this test. It’s important that 'my_callback' is used as this Fake’s name, since we refer to it in the Scenario.

5. We then call the .monitor() method - this method should do all the reading and forwarding.

We must also remember to mock the built-in open:

@pytest.fixture
def override_imports(patch_module):
    patch_module(line_monitor, 'subprocess') # this replaces the subprocess object inside line_monitor with a Fake("subprocess") object
    patch_module(line_monitor, 'pty') # does the same for the pty module
    patch_module(line_monitor, 'open')

We can already see a problem: the scenario is actually built out of two parts - the part which tests .launch_subprocess(), and the part which tests .monitor().

Furthermore, since we have our previous test in test_lauch_subprocess_with_pseudoterminal, which doesn’t expect the call to open(), the two tests are in contradiction.

The way to handle this is to refactor our test a bit:

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')

        tested.register_callback(Fake('my_callback'))
        tested.monitor()

By convention, helper functions that help us modify scenarios end with _scenario.

OK this seems reasonable, let’s get some RED! Running this both our tests fail:

E       Failed:
E       testix: ExpectationException
E       testix details:
E       === Scenario (no title) ===
E        expected: open('read_from_fd', encoding = 'latin-1')
E        actual  : subprocess.Popen(['my', 'command', 'line'], stdout = 'write_to_fd', close_fds = True)

We changed our expectations from .launch_subprocess() to call open(), but we did not change the implementation yet, so Testix is surprised to find that we actually call subprocess.Popen - and makes our test fail.

Good, let’s fix it and get to GREEN. We introduce the following to our code:

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):
        for _ in range(3):
            line = self._reader.readline()
            self._callback(line)

This passes the test, but that’s not really what we meant - right? Obviously we would like a while True to replace the for _ in range(3) here.

However, if we write a while True, then Testix will fail us for the 4th call to .readline(), since it only expects 3 calls.

Testing infinite, while True loops is a problem, but we can get around it by injecting an exception that will terminate the loop. Just as we can determine what calls to Fake objects return, we can make them raise exceptions.

Testix even comes with an exception class just for this use case, LoopBreaker. Let’s introduce another .readline() expectation into our test, using Testix’s Throwing construct:

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()

NOTE - we once more change the test first. Also note that we can use Throwing to raise any type of exception we want, not just LoopBreaker.

This gets us back into the RED.

E           Failed: DID NOT RAISE <class 'testix.loop_breaker.LoopBreaker'>

Since our code calls .readline() 3 times exactly, the fourth call, which would have resulted in LoopBreaker being raised, did not happen.

Let’s fix our code:

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

And we’re back in GREEN.

Edge Case Test: When There is no Callback

What happens if .monitor() is called, but no callback has been registered? We can of course implement all kinds of behaviour, for example, we can make it “illegal”, and raise an Exception from .monitor() in such a case.

However, let’s do something else. Let’s just define things such that output collected from the subprocess when no callback has been registered is discarded.

def test_monitoring_with_no_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.reader.readline() >> 'line 2'
        s.reader.readline() >> 'line 3'
        s.reader.readline() >> Throwing(loop_breaker.LoopBreaker) # this tells the Fake('reader') to raise an instance of TestixLoopBreaker()

        with pytest.raises(loop_breaker.LoopBreaker):
            tested.monitor()

Notice there’s no .register_callback() here. We demand that .readline() be called, but we don’t demand anything else.

Running this fails with a RED

    def monitor(self):
        while True:
            line = self._reader.readline()
>           self._callback(line)
E           TypeError: 'NoneType' object is not callable

Which reveals that we in fact, did not handle this edge case very well.

Let’s add code that fixes this.

    def monitor(self):
        while True:
            line = self._reader.readline()
            if self._callback is None:
                continue
            self._callback(line)

Our test passes - back to GREEN.

Edge Case Test: Asynchronous Callback Registration

What happens if we start monitoring without a callback, wait a while, and only then register a callback?

This allows a use case where we call the .monitor() (which blocks) in one thread, and register a callback in another thread.

Let’s decide that in this case, the callback will receive output which is captured only after the callback has been registered.

Our test will be similar to test_receive_output_lines_via_callback(), however, we need to somehow make tested.register_callback() happen somewhere between one .readline() and the next. This is not so easy to do because of the same while True that gave us some trouble before.

Testix allows us to simulate asynchronous events like this using its Hook construct. Essentially Hook(function, *args, **kawrgs) can be injected into the middle of a Scenario, and it will call function(*args, **kwargs) at the point in which it’s injected.

Here’s how to write such a test:

def test_callback_registered_mid_monitoring(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.reader.readline() >> 'line 2'
        s.reader.readline() >> 'line 3'
        s << Hook(tested.register_callback, Fake('my_callback')) # the hook will execute right after the 'line 3' readline finishes
        s.my_callback('line 3') # callback is now registered, so it should be called
        s.reader.readline() >> Throwing(loop_breaker.LoopBreaker)

        with pytest.raises(loop_breaker.LoopBreaker):
            tested.monitor()

When we run it, we discover it’s already GREEN! Oh no!

Turns out our previous change already solved this problem. This happens sometimes in TDD, so to deal with it, we revert our previous change and make sure this test becomes RED - and carefully check that it failed properly. Happily, this is the case for this particular test.

Let’s Recap

We now have our first implementation of the LineMonitor. It essentially works, but it’s still has its problems. We’ll tackle these problems later in this tutorial, but first, let’s do a short recap.