Launching the Subprocess¶
High Level Design¶
We will implement LineMonitor
as follows:
a
LineMonitor
sill launch the subprocess using the subprocess Python standard library.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.
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).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?
First, we use
patch_module
to mock imported modulessubprocess
andpty
, as described above. Note that our test function depends onoverride_imports
to make everything work.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 demandclose_fds=True
wince we want to fully specify our subprocess’s inputs and outputs).
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:
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.
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
andREAD_FD=30
and used those - but it wouldn’t really matter and would make the test more cluttered. Technically, what’s important is thatopenpty()
returns a tuple and we demand that the first item in this tuple is passed over to the right place in the call toPopen()
. 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.