Merge lp:~bac/zope.testing/1012171 into lp:~launchpad/zope.testing/3.9.4-fork

[Brad Crittenden]

Include data written to stdout or stderr by tests into the subunit output stream.

Include data written to stdout or stderr by tests into the subunit output stream.

Test:

bin/test -vvt test_subunit

[Benji York (benji)]

This looks great.

I'm disappointed in the hoops you had to jump through to build the content.Content objects.

review: Approve (code)

[Robert Collins (lifeless)]

This:
+ if msg:
+ details['STDOUT:'] = content.Content(
+ self.PLAIN_TEXT, partial(lambda x: x, msg))

Would be simpler as:
if msg:
    details['STDOUT:'] = content.text_content(msg)

(assuming msg is a unicode object, not a bytestring). If msg is a
bytestring, then:
if msg:
    details['STDOUT:'] = content.Content(content_type.UTF8_TEXT, lambda:[msg])

would be appropriate.

As it is, you're generating content objects that iterate bytes of
length 1, which will be pathologically slow, so you should change this
one way or another. Happy to chat on IRC or wherever if you need more
pointers.

-Rob

[Gavin Panella (allenap)]

On 19 June 2012 21:46, Robert Collins <email address hidden> wrote:
...
> (assuming msg is a unicode object, not a bytestring). If msg is a
> bytestring, then:
> if msg:
>    details['STDOUT:'] = content.Content(content_type.UTF8_TEXT, lambda:[msg])

This will keep a reference to msg in the enclosing scope. A quick look
at the code shows that msg is changed later in the function, so this
will break, hence the need for the partial() I think.

[Robert Collins (lifeless)]

On Wed, Jun 20, 2012 at 10:28 AM, Gavin Panella
<email address hidden> wrote:
> On 19 June 2012 21:46, Robert Collins <email address hidden> wrote:
> ...
>> (assuming msg is a unicode object, not a bytestring). If msg is a
>> bytestring, then:
>> if msg:
>>    details['STDOUT:'] = content.Content(content_type.UTF8_TEXT, lambda:[msg])
>
> This will keep a reference to msg in the enclosing scope. A quick look
> at the code shows that msg is changed later in the function, so this
> will break, hence the need for the partial() I think.

Oh! So, I'd wrap the two lines in a helper, that will break the
scoping problem and be shorter overall:

def attach(label, msg):
    if msg:
        details[label] = content.Content(content_type.UTF8_TEXT, lambda:[msg])
attach('STDOUT:, _get_new_stream_output(sys.stdout))
attach('STDERR:, _get_new_stream_output(sys.stderr))

5 lines vs 6, and less partial gymnastics ;)

-Rob

=== modified file '.bzrignore'
--- .bzrignore	2012-06-11 18:30:09 +0000
+++ .bzrignore	2012-06-19 19:31:21 +0000
@@ -7,3 +7,4 @@
 Session.vim
 dist
 tags
+.emacs.desktop

=== modified file 'setup.py'
--- setup.py	2012-06-14 16:56:56 +0000
+++ setup.py	2012-06-19 19:31:21 +0000
@@ -85,7 +85,7 @@
 
 setup(
     name='zope.testing',
-    version = '3.9.4-p13',
+    version = '3.9.4-p14',
     url='http://pypi.python.org/pypi/zope.testing',
     license='ZPL 2.1',
     description='Zope testing framework, including the testrunner script.',

=== modified file 'src/zope/testing/testrunner/formatter.py'
--- src/zope/testing/testrunner/formatter.py	2012-06-12 19:23:43 +0000
+++ src/zope/testing/testrunner/formatter.py	2012-06-19 19:31:21 +0000
@@ -24,6 +24,7 @@
 import traceback
 
 from datetime import datetime, timedelta
+from functools import partial
 
 from zope.testing.exceptions import DocTestFailureException
 
@@ -713,6 +714,7 @@
     TAG_THREADS = 'zope:threads'
     TAG_REFCOUNTS = 'zope:refcounts'
 
+
     def __init__(self, options, stream=None):
         if subunit is None:
             raise Exception("Requires subunit 0.0.5 or better")
@@ -907,7 +909,9 @@
     def test_success(self, test, seconds):
         if self._time_tests:
             self._emit_timestamp()
-        self._subunit.addSuccess(test)
+        details = {}
+        self._add_test_output(details)
+        self._subunit.addSuccess(test, details=details)
 
     def import_errors(self, import_errors):
         """Report test-module import errors (if any)."""
@@ -943,7 +947,29 @@
 
         return {
             'traceback': content.Content(
-                self.TRACEBACK_CONTENT_TYPE, lambda: [unicode_tb.encode('utf8')])}
+                self.TRACEBACK_CONTENT_TYPE,
+                lambda: [unicode_tb.encode('utf8')])}
+
+    def _get_new_stream_output(self, stream):
+        """Get the stream output written since the last time."""
+        try:
+            stream.truncate()
+            msg = stream.getvalue()
+            stream.seek(0)
+        except (AttributeError, IOError):
+            msg = None
+        return msg
+
+    def _add_test_output(self, details):
+        """If tests write data to stdout or stderr, add it to the details."""
+        msg = self._get_new_stream_output(sys.stdout)
+        if msg:
+            details['STDOUT:'] = content.Content(
+                self.PLAIN_TEXT, partial(lambda x: x, msg))
+        msg = self._get_new_stream_output(sys.stderr)
+        if msg:
+            details['STDERR:'] = content.Content(
+                self.PLAIN_TEXT, partial(lambda x: x, msg))
 
     def test_error(self, test, seconds, exc_info):
         """Report that an error occurred while running a test.
@@ -955,6 +981,7 @@
         if self._time_tests:
             self._emit_timestamp()
         details = self._exc_info_to_details(exc_info)
+        self._add_test_output(details)
         self._subunit.addError(test, details=details)
 
     def test_failure(self, test, seconds, exc_info):
@@ -967,6 +994,7 @@
         if self._time_tests:
             self._emit_timestamp()
         details = self._exc_info_to_details(exc_info)
+        self._add_test_output(details)
         self._subunit.addFailure(test, details=details)
 
     def profiler_stats(self, stats):

=== modified file 'src/zope/testing/testrunner/options.py'
--- src/zope/testing/testrunner/options.py	2012-06-14 17:27:43 +0000
+++ src/zope/testing/testrunner/options.py	2012-06-19 19:31:21 +0000
@@ -19,6 +19,7 @@
 import optparse
 import re
 import os
+from StringIO import StringIO
 import sys
 
 import pkg_resources
@@ -511,6 +512,7 @@
         return (lambda s: not pattern(s))
     return re.compile(pattern).search
 
+
 def merge_options(options, defaults):
     odict = options.__dict__
     for name, value in defaults.__dict__.items():
@@ -601,12 +603,12 @@
                 streams_munged = True
                 # Replace stdout (and possibly stderr) with a file-like black hole
                 # so the subunit stream does not get corrupted by test output.
-                sys.stdout = NullStream()
+                sys.stdout = StringIO()
                 # If we are running in a subprocess then the test runner code will
                 # use stderr as a communication channel back to the spawning
                 # process so we shouldn't touch it.
                 if '--resume-layer' not in sys.argv:
-                    sys.__stderr__ = sys.stderr = NullStream()
+                    sys.__stderr__ = sys.stderr = StringIO()
             # Send subunit output to the real stdout.
             options.output = SubunitOutputFormatter(options, sys.__stdout__)
 

=== modified file 'src/zope/testing/testrunner/test_subunit.py'
--- src/zope/testing/testrunner/test_subunit.py	2012-05-30 11:59:41 +0000
+++ src/zope/testing/testrunner/test_subunit.py	2012-06-19 19:31:21 +0000
@@ -18,10 +18,14 @@
 import sys
 import unittest
 import formatter
-import subunit
 from StringIO import StringIO
 
 
+class FormatterOptions:
+    """Simple options class for use with formatter."""
+    verbose=False
+
+
 class TestSubunitTracebackPrinting(unittest.TestCase):
 
     def makeByteStringFailure(self, text, encoding):
@@ -35,8 +39,6 @@
             return sys.exc_info()
 
     def setUp(self):
-        class FormatterOptions:
-            verbose=False
         options = FormatterOptions()
 
         self.output = StringIO()
@@ -56,4 +58,91 @@
     def test_print_failure_containing_latin1_bytestrings(self):
         exc_info = self.makeByteStringFailure(unichr(241), 'latin1')
         self.subunit_formatter.test_failure(self, 0, exc_info)
-        assert "AssertionError: \xef\xbf\xbd0" in self.output.getvalue()
\ No newline at end of file
+        self.assertIn("\xef\xbf\xbd", self.output.getvalue())
+
+
+class TestSubunitStreamReporting(unittest.TestCase):
+    """Test capture of stdout and stderr.
+
+    The testrunner sets up fake I/O streams for the tests to use for
+    sys.stdout and sys.stderr.  Anything written to those streams is
+    captured and added as part of the subunit multi-part details output.
+    """
+    def setFakeStreams(self):
+        sys.stdout = StringIO()
+        sys.stderr = StringIO()
+
+    def restoreStreams(self):
+        sys.stdout = sys.__stdout__
+        sys.stderr = sys.__stderr__
+
+    def makeExcInfo(self):
+        try:
+            assert False
+        except:
+            return sys.exc_info()
+
+    def setUp(self):
+        class FormatterOptions:
+            verbose = False
+        options = FormatterOptions()
+
+        self.output = StringIO()
+        self.setFakeStreams()
+        self.subunit_formatter = formatter.SubunitOutputFormatter(
+            options, stream=self.output)
+        #self.subunit_formatter._set_stream_positions()
+
+    def tearDown(self):
+        self.restoreStreams()
+
+    def test_stream_success(self):
+        sys.stdout.write("Output written to stdout\n")
+        sys.stderr.write("Output written to stderr\n")
+        fake_test = formatter.FakeTest('stdout_test')
+        self.subunit_formatter.test_success(fake_test, 10)
+        self.restoreStreams()
+        self.assertIn('STDOUT:', self.output.getvalue())
+        self.assertIn('STDERR:', self.output.getvalue())
+
+
+    def test_stream_error(self):
+        sys.stdout.write("Output written to stdout\n")
+        sys.stderr.write("Output written to stderr\n")
+        fake_test = formatter.FakeTest('error_test')
+        exc_info = self.makeExcInfo()
+        self.subunit_formatter.test_error(fake_test, 10, exc_info)
+        self.restoreStreams()
+        self.assertIn('STDOUT:', self.output.getvalue())
+        self.assertIn('STDERR:', self.output.getvalue())
+
+    def test_stream_failure(self):
+        sys.stdout.write("Output written to stdout\n")
+        sys.stderr.write("Output written to stderr\n")
+        fake_test = formatter.FakeTest('error_test')
+        exc_info = self.makeExcInfo()
+        self.subunit_formatter.test_failure(fake_test, 10, exc_info)
+        self.restoreStreams()
+        self.assertIn('STDOUT:', self.output.getvalue())
+        self.assertIn('STDERR:', self.output.getvalue())
+
+    def test_multiple_messages(self):
+        # Only never-reported messages should be seen.
+        fake_test = formatter.FakeTest('stdout_test')
+        sys.stdout.write("First message written to stdout\n")
+        sys.stderr.write("First message written to stderr\n")
+        self.subunit_formatter.test_success(fake_test, 10)
+        self.assertIn(
+            "First message written to stdout", self.output.getvalue())
+        self.assertIn(
+            "First message written to stderr", self.output.getvalue())
+        self.output.seek(0)
+        self.output.truncate()
+        sys.stdout.write("Second message written to stdout\n")
+        sys.stderr.write("Second message written to stderr\n")
+        self.subunit_formatter.test_success(fake_test, 10)
+        self.restoreStreams()
+        self.assertNotIn(
+            "First message written to stdout", self.output.getvalue())
+        self.assertNotIn(
+            "First message written to stderr", self.output.getvalue())