Python: Printing file to stream as file changes

Question

I am running a large suite of unittests in a subprocess through another application (Autodesk Maya). Maya runs a special Python interpreter with it's own libraries that cannot be used outside of the application, thus the need to test within the application. I would like the parent process to print the results of the testing as it is happening. The subprocess is very 'noisy' though, so I do not want to simply redirect the subprocess's stdout to the parent process's stdout. Instead, I would like the test runner to somehow directly stream to the parent process's stdout

I am currently using a TextTestRunner in the subprocess with it's stdout set to an open file. The parent process knows where this file exists, and writes the contents of the tile to stdout once the subprocess is complete. Since the tests can take a long time to run though, I would prefer that the parent process can somehow 'stream' the contents of this file as it is being created by the subprocess. But I am not sure how to do this or if there is a better approach.

Here's an example of how this is currently set up.

module_path = 'my.test.module'
suite_callable = 'suite'
stream_fpath = '/tmp/the_test_results.txt'
script_fpath = '/tmp/the_test_script.py'
script = '''
import sys
if sys.version_info[0] <= 2 and sys.version_info[1] <= 6:
    import unittest2 as unittest
else:
    import unittest

import {module_path}
suite = {module_path}.{suite_callable}()
with open("{stream_path}", "w") as output:
    runner = unittest.TextTestRunner(stream=output)
    runner.run(suite)
    output.close()

'''.format(**locals())

with open(script_fpath, 'w') as f:
    f.write(script)
subprocess.call(['maya', '-command', '\'python("execfile(\\"{script_fpath}\\")")\''.format(**locals())]
with open(stream_fpath, 'r') as f:
    print f.read()

Thanks for any info!

Answer 1

Rather than writing to a file, you shoud be able to make a file-like object to replace stderr. The object's write method could do something with each input as it comes in; you could squirt it to something listening on TCP, or print stuff to a TK window, or anything else in addition to logging to a file if you still want the results.

Implementing a stream replacement is pretty simple, in this case you probably only need to implement write, writelines, open and close (unless your testrunner also uses flush).

class FakeStdErr(object):
     def __init__(self)
         self.lines = []

     def write(self, text):
         self.lines.append(text)

    def writelines(self, *args):
         for item in args: self.lines.append(item)

    def open(self):
        self.lines = []

    def close (self):
        pass

In your use case you might want to use a silencer class (which is a variant on the same trick) to replace the default stdout (to shut up your chatty process) and direct your test runner stream to this guy; after all the tests are done you could dump the contents to disk as a file or print them to the screen by restoring the default stdout (the link shows how to do that if you're not familiar).

Answer 2

(Edited - suggest using stderr, or parsing)

Alternative 1: Intercept the output rather than having it go to a file.

Have script write to sys.stderr instead of the open() of stream_fpath:

runner = unittest.TextTestRunner(stream=sys.stderr)

Replace subprocess.call with running = subprocess.Popen(<existing parameters>, stderr=PIPE). Then read running.stderr until EOF or until running.poll() returns other than None. You can do what you want with the data. For example, you can print it to the screen and also print it to stream_fpath.

This assumes that the noisy output comes from maya, which will still be dumping to stdout.

Alternative 2: parse the noisy output from stdout=PIPE. If you can differentiate the test runner's output by adding some tag to each line, you can search for that tag and only print the lines that match.

Popen documentation (Python 2)