Merge lp:~lifeless/subunit/streamresult into lp:~subunit/subunit/trunk

[Robert Collins]

Woo! Overlapping concurrent tests with built in enumeration and twice as fast parsing.

[lp:~lifeless/subunit/streamresult]

203. By Robert Collins on 2013-03-06

Fixes from getting testrepository running with v2.

204. By Robert Collins on 2013-03-12

* ``subunit.run`` now replaces sys.stdout to ensure that stdout is unbuffered
  - without this pdb output is not reliably visible when stdout is a pipe
  as it usually is. (Robert Collins)

205. By Robert Collins on 2013-03-31

Switch to variable length encoded integers.

=== modified file 'Makefile.am'
--- Makefile.am	2012-12-17 07:32:51 +0000
+++ Makefile.am	2013-03-31 05:51:20 +0000
@@ -47,6 +47,8 @@
 include_subunitdir = $(includedir)/subunit
 
 dist_bin_SCRIPTS = \
+	filters/subunit-1to2 \
+	filters/subunit-2to1 \
 	filters/subunit-filter \
 	filters/subunit-ls \
 	filters/subunit-notify \
 
=== modified file 'NEWS'
--- NEWS	2013-02-07 11:33:11 +0000
+++ NEWS	2013-03-31 05:51:20 +0000
@@ -5,6 +5,41 @@
 NEXT (In development)
 ---------------------
 
+v2 protocol draft included in this release. The v2 protocol trades off human
+readability for a massive improvement in robustness, the ability to represent
+concurrent tests in a single stream, cheaper parsing, and that provides
+significantly better in-line debugging support and structured forwarding
+of non-test data (such as stdout or stdin data).
+
+This change includes two new filters (subunit-1to2 and subunit-2to1). Use
+these filters to convert old streams to v2 and convert v2 streams to v1.
+
+All the other filters now only parse and emit v2 streams. V2 is still in
+draft format, so if you want to delay and wait for v2 to be finalised, you
+should use subunit-2to1 before any serialisation steps take place.
+With the ability to encapsulate multiple non-test streams, another significant
+cange is that filters which emit subunit now encapsulate any non-subunit they
+encounter, labelling it 'stdout'. This permits multiplexing such streams and
+detangling the stdout streams from each input.
+
+The subunit libraries (Python etc) have not changed their behaviour: they
+still emit v1 from their existing API calls. New API's are being added
+and applications should migrate once their language has those API's available.
+
+IMPROVEMENTS
+~~~~~~~~~~~~
+
+* ``subunit.run`` now replaces sys.stdout to ensure that stdout is unbuffered
+  - without this pdb output is not reliably visible when stdout is a pipe
+  as it usually is. (Robert Collins)
+
+* v2 protocol draft included in this release.
+  (Robert Collins)
+
+* Two new Python classes -- ``StreamResultToBytes`` and
+  ``ByteStreamToStreamResult`` handle v2 generation and parsing.
+  (Robert Collins)
+
 0.0.10
 ------
 
 
=== modified file 'README'
--- README	2013-02-07 11:33:11 +0000
+++ README	2013-03-31 05:51:20 +0000
@@ -21,9 +21,26 @@
 Subunit
 -------
 
-Subunit is a streaming protocol for test results. The protocol is human
-readable and easily generated and parsed. By design all the components of 
-the protocol conceptually fit into the xUnit TestCase->TestResult interaction.
+Subunit is a streaming protocol for test results.
+
+There are two major revisions of the protocol. Version 1 was trivially human
+readable but had significant defects as far as highly parallel testing was
+concerned - it had no room for doing discovery and execution in parallel,
+required substantial buffering when multiplexing and was fragile - a corrupt
+byte could cause an entire stream to be misparsed. Version 1.1 added
+encapsulation of binary streams which mitigated some of the issues but the
+core remained.
+
+Version 2 shares many of the good characteristics of Version 1 - it can be
+embedded into a regular text stream (e.g. from a build system) and it still
+models xUnit style test execution. It also fixes many of the issues with
+Version 1 - Version 2 can be multiplexed without excessive buffering (in
+time or space), it has a well defined recovery mechanism for dealing with
+corrupted streams (e.g. where two processes write to the same stream
+concurrently, or where the stream generator suffers a bug).
+
+More details on both protocol version s can be found in the 'Protocol' section
+of this document.
 
 Subunit comes with command line filters to process a subunit stream and
 language bindings for python, C, C++ and shell. Bindings are easy to write
@@ -32,11 +49,12 @@
 A number of useful things can be done easily with subunit:
  * Test aggregation: Tests run separately can be combined and then
    reported/displayed together. For instance, tests from different languages
-   can be shown as a seamless whole.
+   can be shown as a seamless whole, and tests running on multiple machines
+   can be aggregated into a single stream through a multiplexer.
  * Test archiving: A test run may be recorded and replayed later.
  * Test isolation: Tests that may crash or otherwise interact badly with each
    other can be run seperately and then aggregated, rather than interfering
-   with each other.
+   with each other or requiring an adhoc test->runner reporting protocol.
  * Grid testing: subunit can act as the necessary serialisation and
    deserialiation to get test runs on distributed machines to be reported in
    real time.
@@ -68,20 +86,20 @@
 in python and there are facilities for using Subunit to increase test isolation
 seamlessly within a test suite.
 
-One simple way to run an existing python test suite and have it output subunit
-is the module ``subunit.run``::
+The most common way is to run an existing python test suite and have it output
+subunit via the ``subunit.run`` module::
 
   $ python -m subunit.run mypackage.tests.test_suite
  
 For more information on the Python support Subunit offers , please see
-``pydoc subunit``, or the source in ``python/subunit/__init__.py``
+``pydoc subunit``, or the source in ``python/subunit/``
 
 C
 =
 
-Subunit has C bindings to emit the protocol, and comes with a patch for 'check'
-which has been nominally accepted by the 'check' developers. See 'c/README' for
-more details.
+Subunit has C bindings to emit the protocol. The 'check' C unit testing project
+has included subunit support in their project for some years now. See
+'c/README' for more details.
 
 C++
 ===
@@ -92,9 +110,13 @@
 shell
 =====
 
-Similar to C, the shell bindings consist of simple functions to output protocol
-elements, and a patch for adding subunit output to the 'ShUnit' shell test
-runner. See 'shell/README' for details.
+There are two sets of shell tools. There are filters, which accept a subunit
+stream on stdin and output processed data (or a transformed stream) on stdout.
+
+Then there are unittest facilities similar to those for C : shell bindings
+consisting of simple functions to output protocol elements, and a patch for
+adding subunit output to the 'ShUnit' shell test runner. See 'shell/README' for
+details.
 
 Filter recipes
 --------------
@@ -104,9 +126,225 @@
   subunit-filter --without 'AttributeError.*flavor'
 
 
+The xUnit test model
+--------------------
+
+Subunit implements a slightly modified xUnit test model. The stock standard
+model is that there are tests, which have an id(), can be run, and when run
+start, emit an outcome (like success or failure) and then finish.
+
+Subunit extends this with the idea of test enumeration (find out about tests
+a runner has without running them), tags (allow users to describe tests in
+ways the test framework doesn't apply any semantic value to), file attachments
+(allow arbitrary data to make analysing a failure easy) and timestamps.
+
 The protocol
 ------------
 
+Version 2, or v2 is new and still under development, but is intended to
+supercede version 1 in the very near future. Subunit's bundled tools accept
+only version 2 and only emit version 2, but the new filters subunit-1to2 and
+subunit-2to1 can be used to interoperate with older third party libraries.
+
+Version 2
+=========
+
+Version 2 is a binary protocol consisting of independent packets that can be
+embedded in the output from tools like make - as long as each packet has no
+other bytes mixed in with it (which 'make -j N>1' has a tendency of doing).
+Version 2 is currently in draft form, and early adopters should be willing
+to either discard stored results (if protocol changes are made), or bulk
+convert them back to v1 and then to a newer edition of v2.
+
+The protocol synchronises at the start of the stream, after a packet, or
+after any 0x0A byte. That is, a subunit v2 packet starts after a newline or
+directly after the end of the prior packet.
+
+Subunit is intended to be transported over a reliable streaming protocol such
+as TCP. As such it does not concern itself with out of order delivery of
+packets. However, because of the possibility of corruption due to either
+bugs in the sender, or due to mixed up data from concurrent writes to the same
+fd when being embedded, subunit strives to recover reasonably gracefully from
+damaged data.
+
+A key design goal for Subunit version 2 is to allow processing and multiplexing
+without forcing buffering for semantic correctness, as buffering tends to hide
+hung or otherwise misbehaving tests. That said, limited time based buffering
+for network efficiency is a good idea - this is ultimately implementator
+choice. Line buffering is also discouraged for subunit streams, as dropping
+into a debugger or other tool may require interactive traffic even if line
+buffering would not otherwise be a problem.
+
+In version two there are two conceptual events - a test status event and a file
+attachment event. Events may have timestamps, and the path of multiplexers that
+an event is routed through is recorded to permit sending actions back to the
+source (such as new tests to run or stdin for driving debuggers and other
+interactive input). Test status events are used to enumerate tests, to report
+tests and test helpers as they run. Tests may have tags, used to allow
+tunnelling extra meanings through subunit without requiring parsing of
+arbitrary file attachments. Things that are not standalone tests get marked
+as such by setting the 'Runnable' flag to false. (For instance, individual
+assertions in TAP are not runnable tests, only the top level TAP test script
+is runnable).
+
+File attachments are used to provide rich detail about the nature of a failure.
+File attachments can also be used to encapsulate stdout and stderr both during
+and outside tests.
+
+Most numbers are stored in network byte order - Most Significant Byte first
+encoded using a variation of http://www.dlugosz.com/ZIP2/VLI.html. The first
+byte's top 2 high order bits encode the total number of octets in the number.
+This encoding can encode values from 0 to 2**30-1, enough to encode a
+nanosecond. Numbers that are not variable length encoded are still stored in
+MSB order.
+
+ prefix   octets   max       max
++-------+--------+---------+------------+
+| 00    |      1 |  2**6-1 |         63 |
+| 01    |      2 | 2**14-1 |      16383 |
+| 10    |      3 | 2**22-1 |    4194303 |
+| 11    |      4 | 2**30-1 | 1073741823 |
++-------+--------+---------+------------+
+
+All variable length elements of the packet are stored with a length prefix
+number allowing them to be skipped over for consumers that don't need to
+interpret them.
+
+UTF-8 strings are with no terminating NUL and should not have any embedded NULs
+(implementations SHOULD validate any such strings that they process and take
+some remedial action (such as discarding the packet as corrupt).
+
+In short the structure of a packet is:
+PACKET := SIGNATURE FLAGS PACKET_LENGTH TIMESTAMP? TESTID? TAGS? MIME?
+          FILECONTENT? ROUTING_CODE? CRC32
+
+In more detail...
+
+Packets are identified by a single byte signature - 0xB3, which is never legal
+in a UTF-8 stream as the first byte of a character. 0xB3 starts with the first
+bit set and the second not, which is the UTF-8 signature for a continuation
+byte. 0xB3 was chosen as 0x73 ('s' in ASCII') with the top two bits replaced by
+the 1 and 0 for a continuation byte.
+
+If subunit packets are being embedded in a non-UTF-8 text stream, where 0x73 is
+a legal character, consider either recoding the text to UTF-8, or using
+subunit's 'file' packets to embed the text stream in subunit, rather than the
+other way around.
+
+Following the signature byte comes a 16-bit flags field, which includes a
+4-bit version field - if the version is not 0x2 then the packet cannot be
+read. It is recommended to signal an error at this point (e.g. by emitting
+a synthetic error packet and returning to the top level loop to look for
+new packets, or exiting with an error). If recovery is desired, treat the
+packet signature as an opaque byte and scan for a new synchronisation point.
+NB: Subunit V1 and V2 packets may legitimately included 0xB3 internally,
+as they are an 8-bit safe container format, so recovery from this situation
+may involve an arbitrary number of false positives until an actual packet
+is encountered : and even then it may still be false, failing after passing
+the version check due to coincidence.
+
+Flags are stored in network byte order too.
++-------------------------+------------------------+
+| High byte               | Low byte               |
+| 15 14 13 12 11 10  9  8 | 7  6  5  4  3  2  1  0 |
+| VERSION    |feature bits|                        |
++------------+------------+------------------------+
+
+Valid version values are:
+0x2 - version 2
+
+Feature bits:
+Bit 11 - mask 0x0800 - Test id present.
+Bit 10 - mask 0x0400 - Routing code present.
+Bit  9 - mask 0x0200 - Timestamp present.
+Bit  8 - mask 0x0100 - Test is 'runnable'.
+Bit  7 - mask 0x0080 - Tags are present.
+Bit  6 - mask 0x0040 - File content is present.
+Bit  5 - mask 0x0020 - File MIME type is present.
+Bit  4 - mask 0x0010 - EOF marker.
+Bit  3 - mask 0x0008 - Must be zero in version 2.
+
+Test status gets three bits:
+Bit 2 | Bit 1 | Bit 0 - mask 0x0007 - A test status enum lookup:
+000 - undefined / no test
+001 - Enumeration / existence
+002 - In progress
+003 - Success
+004 - Unexpected Success
+005 - Skipped
+006 - Failed
+007 - Expected failure
+
+After the flags field is a number field giving the length in bytes for the
+entire packet including the signature and the checksum. This length must
+be less than 4MiB - 4194303 bytes. The encoding can obviously record a larger
+number but one of the goals is to avoid requiring large buffers, or causing
+large latency in the packet forward/processing pipeline. Larger file
+attachments can be communicated in multiple packets, and the overhead in such a
+4MiB packet is approximately 0.2%.
+
+The rest of the packet is a series of optional features as specified by the set
+feature bits in the flags field. When absent they are entirely absent.
+
+Forwarding and multiplexing of packets can be done without interpreting the
+remainder of the packet until the routing code and checksum (which are both at
+the end of the packet). Additionally, routers can often avoid copying or moving
+the bulk of the packet, as long as the routing code size increase doesn't force
+the length encoding to take up a new byte (which will only happen to packets
+less than or equal to 16KiB in length) - large packets are very efficient to
+route.
+
+Timestamp when present is a 32 bit unsigned integer for secnods, and a variable
+length number for nanoseconds, representing UTC time since Unix Epoch in
+seconds and nanoseconds.
+
+Test id when present is a UTF-8 string. The test id should uniquely identify
+runnable tests such that they can be selected individually. For tests and other
+actions which cannot be individually run (such as test
+fixtures/layers/subtests) uniqueness is not required (though being human
+meaningful is highly recommended).
+
+Tags when present is a length prefixed vector of UTF-8 strings, one per tag.
+There are no restrictions on tag content (other than the restrictions on UTF-8
+strings in subunit in general). Tags have no ordering.
+
+When a MIME type is present, it defines the MIME type for the file across all
+packets same file (routing code + testid + name uniquely identifies a file,
+reset when EOF is flagged). If a file never has a MIME type set, it should be
+treated as application/octet-stream.
+
+File content when present is a UTF-8 string for the name followed by the length
+in bytes of the content, and then the content octets.
+
+If present routing code is a UTF-8 string. The routing code is used to
+determine which test backend a test was running on when doing data analysis,
+and to route stdin to the test process if interaction is required.
+
+Multiplexers SHOULD add a routing code if none is present, and prefix any
+existing routing code with a routing code ('/' separated) if one is already
+present. For example, a multiplexer might label each stream it is multiplexing
+with a simple ordinal ('0', '1' etc), and given an incoming packet with route
+code '3' from stream '0' would adjust the route code when forwarding the packet
+to be '0/3'.
+
+Following the end of the packet is a CRC-32 checksum of the contents of the
+packet including the signature.
+
+Example packets
+~~~~~~~~~~~~~~~
+
+Trivial test "foo" enumeration packet, with test id, runnable set,
+status=enumeration. Spaces below are to visually break up signature / flags /
+length / testid / crc32
+
+b3 2901 0c 03666f6f 08555f1b
+
+
+Version 1 (and 1.1)
+===================
+
+Version 1 (and 1.1) are mostly human readable protocols. 
+
 Sample subunit wire contents
 ----------------------------
 
 
=== added file 'filters/subunit-1to2'
--- filters/subunit-1to2	1970-01-01 00:00:00 +0000
+++ filters/subunit-1to2	2013-03-31 05:51:20 +0000
@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+#  subunit: extensions to python unittest to get test results from subprocesses.
+#  Copyright (C) 2013  Robert Collins <[e-mail]>
+#
+#  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
+#  license at the users choice. A copy of both licenses are available in the
+#  project source as Apache-2.0 and BSD. You may not use this file except in
+#  compliance with one of these two licences.
+#  
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+#  license you chose for the specific language governing permissions and
+#  limitations under that license.
+#
+
+"""Convert a version 1 subunit stream to version 2 stream."""
+
+from optparse import OptionParser
+import sys
+
+from testtools import ExtendedToStreamDecorator
+
+from subunit import StreamResultToBytes
+from subunit.filters import run_tests_from_stream
+
+
+def make_options(description):
+    parser = OptionParser(description=__doc__)
+    return parser
+
+
+def main():
+    parser = make_options(__doc__)
+    (options, args) = parser.parse_args()
+    run_tests_from_stream(sys.stdin,
+        ExtendedToStreamDecorator(StreamResultToBytes(sys.stdout)))
+    sys.exit(0)
+
+
+if __name__ == '__main__':
+    main()
 
=== added file 'filters/subunit-2to1'
--- filters/subunit-2to1	1970-01-01 00:00:00 +0000
+++ filters/subunit-2to1	2013-03-31 05:51:20 +0000
@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+#  subunit: extensions to python unittest to get test results from subprocesses.
+#  Copyright (C) 2013  Robert Collins <[e-mail]>
+#
+#  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
+#  license at the users choice. A copy of both licenses are available in the
+#  project source as Apache-2.0 and BSD. You may not use this file except in
+#  compliance with one of these two licences.
+#  
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+#  license you chose for the specific language governing permissions and
+#  limitations under that license.
+#
+
+"""Convert a version 2 subunit stream to a version 1 stream."""
+
+from optparse import OptionParser
+import sys
+
+from testtools import StreamToExtendedDecorator
+
+from subunit import ByteStreamToStreamResult, TestProtocolClient
+from subunit.filters import run_tests_from_stream
+
+
+def make_options(description):
+    parser = OptionParser(description=__doc__)
+    return parser
+
+
+def main():
+    parser = make_options(__doc__)
+    (options, args) = parser.parse_args()
+    case = ByteStreamToStreamResult(sys.stdin, non_subunit_name='stdout')
+    result = StreamToExtendedDecorator(TestProtocolClient(sys.stdout))
+    # What about stdout chunks?
+    result.startTestRun()
+    case.run(result)
+    result.stopTestRun()
+    sys.exit(0)
+
+
+if __name__ == '__main__':
+    main()
 
=== modified file 'filters/subunit-filter'
--- filters/subunit-filter	2012-05-03 08:18:01 +0000
+++ filters/subunit-filter	2013-03-31 05:51:20 +0000
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #  subunit: extensions to python unittest to get test results from subprocesses.
-#  Copyright (C) 2008  Robert Collins <[e-mail]>
+#  Copyright (C) 200-2013  Robert Collins <[e-mail]>
 #            (C) 2009  Martin Pool
 #
 #  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
@@ -30,10 +30,12 @@
 import sys
 import re
 
+from testtools import ExtendedToStreamDecorator, StreamToExtendedDecorator
+
 from subunit import (
     DiscardStream,
     ProtocolTestCase,
-    TestProtocolClient,
+    StreamResultToBytes,
     read_test_list,
     )
 from subunit.filters import filter_by_result
@@ -55,9 +57,11 @@
     parser.add_option("-f", "--no-failure", action="store_true",
         help="exclude failures", dest="failure")
     parser.add_option("--passthrough", action="store_false",
-        help="Show all non subunit input.", default=False, dest="no_passthrough")
+        help="Forward non-subunit input as 'stdout'.", default=False,
+        dest="no_passthrough")
     parser.add_option("--no-passthrough", action="store_true",
-        help="Hide all non subunit input.", default=False, dest="no_passthrough")
+        help="Discard all non subunit input.", default=False,
+        dest="no_passthrough")
     parser.add_option("-s", "--success", action="store_false",
         help="include successes", dest="success")
     parser.add_option("--no-success", action="store_true",
@@ -126,15 +130,16 @@
     fixup_expected_failures = set()
     for path in options.fixup_expected_failures or ():
         fixup_expected_failures.update(read_test_list(path))
-    return TestResultFilter(
-        TestProtocolClient(output),
+    return StreamToExtendedDecorator(TestResultFilter(
+        ExtendedToStreamDecorator(
+        StreamResultToBytes(output)),
         filter_error=options.error,
         filter_failure=options.failure,
         filter_success=options.success,
         filter_skip=options.skip,
         filter_xfail=options.xfail,
         filter_predicate=predicate,
-        fixup_expected_failures=fixup_expected_failures)
+        fixup_expected_failures=fixup_expected_failures))
 
 
 def main():
@@ -150,7 +155,8 @@
         lambda output_to: _make_result(sys.stdout, options, filter_predicate),
         output_path=None,
         passthrough=(not options.no_passthrough),
-        forward=False)
+        forward=False,
+        protocol_version=2)
     sys.exit(0)
 
 
 
=== modified file 'filters/subunit-ls'
--- filters/subunit-ls	2011-05-23 09:57:58 +0000
+++ filters/subunit-ls	2013-03-31 05:51:20 +0000
@@ -19,9 +19,14 @@
 from optparse import OptionParser
 import sys
 
-from subunit import DiscardStream, ProtocolTestCase
+from testtools import (
+    CopyStreamResult, StreamToExtendedDecorator, StreamResultRouter,
+    StreamSummary)
+
+from subunit import ByteStreamToStreamResult
+from subunit.filters import run_tests_from_stream
 from subunit.test_results import (
-    AutoTimingTestResultDecorator,
+    CatFiles,
     TestIdPrintingResult,
     )
 
@@ -30,18 +35,24 @@
 parser.add_option("--times", action="store_true",
     help="list the time each test took (requires a timestamped stream)",
         default=False)
+parser.add_option("--exists", action="store_true",
+    help="list tests that are reported as existing (as well as ran)",
+        default=False)
 parser.add_option("--no-passthrough", action="store_true",
     help="Hide all non subunit input.", default=False, dest="no_passthrough")
 (options, args) = parser.parse_args()
-result = AutoTimingTestResultDecorator(
-    TestIdPrintingResult(sys.stdout, options.times))
-if options.no_passthrough:
-    passthrough_stream = DiscardStream()
-else:
-    passthrough_stream = None
-test = ProtocolTestCase(sys.stdin, passthrough=passthrough_stream)
+test = ByteStreamToStreamResult(sys.stdin, non_subunit_name="stdout")
+result = TestIdPrintingResult(sys.stdout, options.times, options.exists)
+if not options.no_passthrough:
+    result = StreamResultRouter(result)
+    cat = CatFiles(sys.stdout)
+    result.map(cat, 'test_id', test_id=None)
+summary = StreamSummary()
+result = CopyStreamResult([result, summary])
+result.startTestRun()
 test.run(result)
-if result.wasSuccessful():
+result.stopTestRun()
+if summary.wasSuccessful():
     exit_code = 0
 else:
     exit_code = 1
 
=== modified file 'filters/subunit-notify'
--- filters/subunit-notify	2012-03-27 11:17:37 +0000
+++ filters/subunit-notify	2013-03-31 05:51:20 +0000
@@ -19,6 +19,7 @@
 import pygtk
 pygtk.require('2.0')
 import pynotify
+from testtools import StreamToExtendedDecorator
 
 from subunit import TestResultStats
 from subunit.filters import run_filter_script
@@ -28,6 +29,7 @@
 
 
 def notify_of_result(result):
+    result = result.decorated
     if result.failed_tests > 0:
         summary = "Test run failed"
     else:
@@ -41,4 +43,6 @@
     nw.show()
 
 
-run_filter_script(TestResultStats, __doc__, notify_of_result)
+run_filter_script(
+    lambda output:StreamToExtendedDecorator(TestResultStats(output)),
+    __doc__, notify_of_result, protocol_version=2)
 
=== modified file 'filters/subunit-stats'
--- filters/subunit-stats	2009-09-30 12:04:18 +0000
+++ filters/subunit-stats	2013-03-31 05:51:20 +0000
@@ -16,26 +16,17 @@
 
 """Filter a subunit stream to get aggregate statistics."""
 
-from optparse import OptionParser
 import sys
-import unittest
-
-from subunit import DiscardStream, ProtocolTestCase, TestResultStats
-
-parser = OptionParser(description=__doc__)
-parser.add_option("--no-passthrough", action="store_true",
-    help="Hide all non subunit input.", default=False, dest="no_passthrough")
-(options, args) = parser.parse_args()
+
+from testtools import StreamToExtendedDecorator
+
+from subunit import TestResultStats
+from subunit.filters import run_filter_script
+
+
 result = TestResultStats(sys.stdout)
-if options.no_passthrough:
-    passthrough_stream = DiscardStream()
-else:
-    passthrough_stream = None
-test = ProtocolTestCase(sys.stdin, passthrough=passthrough_stream)
-test.run(result)
-result.formatStats()
-if result.wasSuccessful():
-    exit_code = 0
-else:
-    exit_code = 1
-sys.exit(exit_code)
+def show_stats(r):
+    r.decorated.formatStats()
+run_filter_script(
+    lambda output:StreamToExtendedDecorator(result),
+    __doc__, show_stats, protocol_version=2)
 
=== modified file 'filters/subunit2csv'
--- filters/subunit2csv	2012-03-27 10:57:51 +0000
+++ filters/subunit2csv	2013-03-31 05:51:20 +0000
@@ -16,8 +16,11 @@
 
 """Turn a subunit stream into a CSV"""
 
+from testtools import StreamToExtendedDecorator
+
 from subunit.filters import run_filter_script
 from subunit.test_results import CsvResult
 
 
-run_filter_script(CsvResult, __doc__)
+run_filter_script(lambda output:StreamToExtendedDecorator(CsvResult(output)),
+    __doc__, protocol_version=2)
 
=== modified file 'filters/subunit2gtk'
--- filters/subunit2gtk	2010-01-25 15:45:45 +0000
+++ filters/subunit2gtk	2013-03-31 05:51:20 +0000
@@ -46,17 +46,20 @@
 """Display a subunit stream in a gtk progress window."""
 
 import sys
+import threading
 import unittest
 
 import pygtk
 pygtk.require('2.0')
 import gtk, gtk.gdk, gobject
 
+from testtools import StreamToExtendedDecorator
+
 from subunit import (
     PROGRESS_POP,
     PROGRESS_PUSH,
     PROGRESS_SET,
-    TestProtocolServer,
+    ByteStreamToStreamResult,
     )
 from subunit.progress_model import  ProgressModel
 
@@ -139,6 +142,9 @@
 
     def stopTest(self, test):
         super(GTKTestResult, self).stopTest(test)
+	gobject.idle_add(self._stopTest)
+
+    def _stopTest(self):
         self.progress_model.advance()
         if self.progress_model.width() == 0:
             self.pbar.pulse()
@@ -153,26 +159,26 @@
             super(GTKTestResult, self).stopTestRun()
         except AttributeError:
             pass
-        self.pbar.set_text('Finished')
+	gobject.idle_add(self.pbar.set_text, 'Finished')
 
     def addError(self, test, err):
         super(GTKTestResult, self).addError(test, err)
-        self.update_counts()
+	gobject.idle_add(self.update_counts)
 
     def addFailure(self, test, err):
         super(GTKTestResult, self).addFailure(test, err)
-        self.update_counts()
+	gobject.idle_add(self.update_counts)
 
     def addSuccess(self, test):
         super(GTKTestResult, self).addSuccess(test)
-        self.update_counts()
+	gobject.idle_add(self.update_counts)
 
     def addSkip(self, test, reason):
         # addSkip is new in Python 2.7/3.1
         addSkip = getattr(super(GTKTestResult, self), 'addSkip', None)
         if callable(addSkip):
             addSkip(test, reason)
-        self.update_counts()
+	gobject.idle_add(self.update_counts)
 
     def addExpectedFailure(self, test, err):
         # addExpectedFailure is new in Python 2.7/3.1
@@ -180,7 +186,7 @@
             'addExpectedFailure', None)
         if callable(addExpectedFailure):
             addExpectedFailure(test, err)
-        self.update_counts()
+	gobject.idle_add(self.update_counts)
 
     def addUnexpectedSuccess(self, test):
         # addUnexpectedSuccess is new in Python 2.7/3.1
@@ -188,7 +194,7 @@
             'addUnexpectedSuccess', None)
         if callable(addUnexpectedSuccess):
             addUnexpectedSuccess(test)
-        self.update_counts()
+	gobject.idle_add(self.update_counts)
 
     def progress(self, offset, whence):
         if whence == PROGRESS_PUSH:
@@ -212,47 +218,22 @@
         self.ok_label.set_text(str(self.testsRun - bad))
         self.not_ok_label.set_text(str(bad))
 
-
-class GIOProtocolTestCase(object):
-
-    def __init__(self, stream, result, on_finish):
-        self.stream = stream
-        self.schedule_read()
-        self.hup_id = gobject.io_add_watch(stream, gobject.IO_HUP, self.hup)
-        self.protocol = TestProtocolServer(result)
-        self.on_finish = on_finish
-
-    def read(self, source, condition, all=False):
-        #NB: \o/ actually blocks
-        line = source.readline()
-        if not line:
-            self.protocol.lostConnection()
-            self.on_finish()
-            return False
-        self.protocol.lineReceived(line)
-        # schedule more IO shortly - if we say we're willing to do it
-        # immediately we starve things.
-        if not all:
-            source_id = gobject.timeout_add(1, self.schedule_read)
-            return False
-        else:
-            return True
-
-    def schedule_read(self):
-        self.read_id = gobject.io_add_watch(self.stream, gobject.IO_IN, self.read)
-
-    def hup(self, source, condition):
-        while self.read(source, condition, all=True): pass
-        self.protocol.lostConnection()
-        gobject.source_remove(self.read_id)
-        self.on_finish()
-        return False
-
-
-result = GTKTestResult()
-test = GIOProtocolTestCase(sys.stdin, result, result.stopTestRun)
+gobject.threads_init()
+result = StreamToExtendedDecorator(GTKTestResult())
+test = ByteStreamToStreamResult(sys.stdin, non_subunit_name='stdout')
+# Get setup
+while gtk.events_pending():
+  gtk.main_iteration()
+# Start IO
+def run_and_finish():
+    test.run(result)
+    result.stopTestRun()
+t = threading.Thread(target=run_and_finish)
+t.daemon = True
+result.startTestRun()
+t.start()
 gtk.main()
-if result.wasSuccessful():
+if result.decorated.wasSuccessful():
     exit_code = 0
 else:
     exit_code = 1
 
=== modified file 'filters/subunit2junitxml'
--- filters/subunit2junitxml	2012-03-27 10:57:51 +0000
+++ filters/subunit2junitxml	2013-03-31 05:51:20 +0000
@@ -18,6 +18,9 @@
 
 
 import sys
+
+from testtools import StreamToExtendedDecorator
+
 from subunit.filters import run_filter_script
 
 try:
@@ -28,4 +31,6 @@
     raise
 
 
-run_filter_script(JUnitXmlResult, __doc__)
+run_filter_script(
+    lambda output:StreamToExtendedDecorator(JUnitXmlResult(output)), __doc__,
+    protocol_version=2)
 
=== modified file 'filters/subunit2pyunit'
--- filters/subunit2pyunit	2009-09-30 12:04:18 +0000
+++ filters/subunit2pyunit	2013-03-31 05:51:20 +0000
@@ -16,11 +16,15 @@
 
 """Display a subunit stream through python's unittest test runner."""
 
+from operator import methodcaller
 from optparse import OptionParser
 import sys
 import unittest
 
-from subunit import DiscardStream, ProtocolTestCase, TestProtocolServer
+from testtools import StreamToExtendedDecorator, DecorateTestCaseResult, StreamResultRouter
+
+from subunit import ByteStreamToStreamResult
+from subunit.test_results import CatFiles
 
 parser = OptionParser(description=__doc__)
 parser.add_option("--no-passthrough", action="store_true",
@@ -29,11 +33,16 @@
     help="Use bzrlib's test reporter (requires bzrlib)",
         default=False)
 (options, args) = parser.parse_args()
-if options.no_passthrough:
-    passthrough_stream = DiscardStream()
-else:
-    passthrough_stream = None
-test = ProtocolTestCase(sys.stdin, passthrough=passthrough_stream)
+test = ByteStreamToStreamResult(sys.stdin, non_subunit_name='stdout')
+def wrap_result(result):
+    result = StreamToExtendedDecorator(result)
+    if not options.no_passthrough:
+        result = StreamResultRouter(result)
+        result.map(CatFiles(sys.stdout), 'test_id', test_id=None)
+    return result
+test = DecorateTestCaseResult(test, wrap_result,
+    before_run=methodcaller('startTestRun'),
+    after_run=methodcaller('stopTestRun'))
 if options.progress:
     from bzrlib.tests import TextTestRunner
     from bzrlib import ui
 
=== modified file 'python/subunit/__init__.py'
--- python/subunit/__init__.py	2013-02-07 11:33:11 +0000
+++ python/subunit/__init__.py	2013-03-31 05:51:20 +0000
@@ -126,7 +126,7 @@
 except ImportError:
     _UnsupportedOperation = AttributeError
 
-
+from extras import safe_hasattr
 from testtools import content, content_type, ExtendedToOriginalDecorator
 from testtools.content import TracebackContent
 from testtools.compat import _b, _u, BytesIO, StringIO
@@ -143,9 +143,10 @@
 except ImportError:
     raise ImportError ("testtools.testresult.real does not contain "
         "_StringException, check your version.")
-from testtools import testresult
+from testtools import testresult, CopyStreamResult
 
 from subunit import chunked, details, iso8601, test_results
+from subunit.v2 import ByteStreamToStreamResult, StreamResultToBytes
 
 # same format as sys.version_info: "A tuple containing the five components of
 # the version number: major, minor, micro, releaselevel, and serial. All
@@ -992,44 +993,51 @@
     return result
 
 
-def TAP2SubUnit(tap, subunit):
+def TAP2SubUnit(tap, output_stream):
     """Filter a TAP pipe into a subunit pipe.
 
-    :param tap: A tap pipe/stream/file object.
+    This should be invoked once per TAP script, as TAP scripts get
+    mapped to a single runnable case with multiple components.
+
+    :param tap: A tap pipe/stream/file object - should emit unicode strings.
     :param subunit: A pipe/stream/file object to write subunit results to.
     :return: The exit code to exit with.
     """
+    output = StreamResultToBytes(output_stream)
+    UTF8_TEXT = 'text/plain; charset=UTF8'
     BEFORE_PLAN = 0
     AFTER_PLAN = 1
     SKIP_STREAM = 2
     state = BEFORE_PLAN
     plan_start = 1
     plan_stop = 0
-    def _skipped_test(subunit, plan_start):
-        # Some tests were skipped.
-        subunit.write('test test %d\n' % plan_start)
-        subunit.write('error test %d [\n' % plan_start)
-        subunit.write('test missing from TAP output\n')
-        subunit.write(']\n')
-        return plan_start + 1
     # Test data for the next test to emit
     test_name = None
     log = []
     result = None
+    def missing_test(plan_start):
+        output.status(test_id='test %d' % plan_start,
+            test_status='fail', runnable=False, 
+            mime_type=UTF8_TEXT, eof=True, file_name="tap meta",
+            file_bytes=b"test missing from TAP output")
     def _emit_test():
         "write out a test"
         if test_name is None:
             return
-        subunit.write("test %s\n" % test_name)
-        if not log:
-            subunit.write("%s %s\n" % (result, test_name))
-        else:
-            subunit.write("%s %s [\n" % (result, test_name))
         if log:
-            for line in log:
-                subunit.write("%s\n" % line)
-            subunit.write("]\n")
+            log_bytes = b'\n'.join(log_line.encode('utf8') for log_line in log)
+            mime_type = UTF8_TEXT
+            file_name = 'tap comment'
+            eof = True
+        else:
+            log_bytes = None
+            mime_type = None
+            file_name = None
+            eof = True
         del log[:]
+        output.status(test_id=test_name, test_status=result,
+            file_bytes=log_bytes, mime_type=mime_type, eof=eof,
+            file_name=file_name, runnable=False)
     for line in tap:
         if state == BEFORE_PLAN:
             match = re.match("(\d+)\.\.(\d+)\s*(?:\#\s+(.*))?\n", line)
@@ -1040,10 +1048,9 @@
                 if plan_start > plan_stop and plan_stop == 0:
                     # skipped file
                     state = SKIP_STREAM
-                    subunit.write("test file skip\n")
-                    subunit.write("skip file skip [\n")
-                    subunit.write("%s\n" % comment)
-                    subunit.write("]\n")
+                    output.status(test_id='file skip', test_status='skip',
+                        file_bytes=comment.encode('utf8'), eof=True,
+                        file_name='tap comment')
                 continue
         # not a plan line, or have seen one before
         match = re.match("(ok|not ok)(?:\s+(\d+)?)?(?:\s+([^#]*[^#\s]+)\s*)?(?:\s+#\s+(TODO|SKIP|skip|todo)(?:\s+(.*))?)?\n", line)
@@ -1054,7 +1061,7 @@
             if status == 'ok':
                 result = 'success'
             else:
-                result = "failure"
+                result = "fail"
             if description is None:
                 description = ''
             else:
@@ -1069,7 +1076,8 @@
             if number is not None:
                 number = int(number)
                 while plan_start < number:
-                    plan_start = _skipped_test(subunit, plan_start)
+                    missing_test(plan_start)
+                    plan_start += 1
             test_name = "test %d%s" % (plan_start, description)
             plan_start += 1
             continue
@@ -1082,18 +1090,21 @@
                 extra = ' %s' % reason
             _emit_test()
             test_name = "Bail out!%s" % extra
-            result = "error"
+            result = "fail"
             state = SKIP_STREAM
             continue
         match = re.match("\#.*\n", line)
         if match:
             log.append(line[:-1])
             continue
-        subunit.write(line)
+        # Should look at buffering status and binding this to the prior result.
+        output.status(file_bytes=line.encode('utf8'), file_name='stdout',
+            mime_type=UTF8_TEXT)
     _emit_test()
     while plan_start <= plan_stop:
         # record missed tests
-        plan_start = _skipped_test(subunit, plan_start)
+        missing_test(plan_start)
+        plan_start += 1
     return 0
 
 
@@ -1121,24 +1132,21 @@
     :return: 0
     """
     new_tags, gone_tags = tags_to_new_gone(tags)
-    def write_tags(new_tags, gone_tags):
-        if new_tags or gone_tags:
-            filtered.write("tags: " + ' '.join(new_tags))
-            if gone_tags:
-                for tag in gone_tags:
-                    filtered.write("-" + tag)
-            filtered.write("\n")
-    write_tags(new_tags, gone_tags)
-    # TODO: use the protocol parser and thus don't mangle test comments.
-    for line in original:
-        if line.startswith("tags:"):
-            line_tags = line[5:].split()
-            line_new, line_gone = tags_to_new_gone(line_tags)
-            line_new = line_new - gone_tags
-            line_gone = line_gone - new_tags
-            write_tags(line_new, line_gone)
-        else:
-            filtered.write(line)
+    source = ByteStreamToStreamResult(original, non_subunit_name='stdout')
+    class Tagger(CopyStreamResult):
+        def status(self, **kwargs):
+            tags = kwargs.get('test_tags')
+            if not tags:
+                tags = set()
+            tags.update(new_tags)
+            tags.difference_update(gone_tags)
+            if tags:
+                kwargs['test_tags'] = tags
+            else:
+                kwargs['test_tags'] = None
+            super(Tagger, self).status(**kwargs)
+    output = Tagger([StreamResultToBytes(filtered)])
+    source.run(output)
     return 0
 
 
@@ -1260,7 +1268,8 @@
     else:
         stream = sys.stdout
         if sys.version_info > (3, 0):
-            stream = stream.buffer
+            if safe_hasattr(stream, 'buffer'):
+                stream = stream.buffer
         return stream
 
 
@@ -1291,6 +1300,7 @@
         _make_binary_on_windows(fileno)
     return _unwrap_text(stream)
 
+
 def _make_binary_on_windows(fileno):
     """Win32 mangles \r\n to \n and that breaks streams. See bug lp:505078."""
     if sys.platform == "win32":
 
=== modified file 'python/subunit/filters.py'
--- python/subunit/filters.py	2012-03-27 11:17:37 +0000
+++ python/subunit/filters.py	2013-03-31 05:51:20 +0000
@@ -17,7 +17,14 @@
 from optparse import OptionParser
 import sys
 
-from subunit import DiscardStream, ProtocolTestCase
+from extras import safe_hasattr
+from testtools import CopyStreamResult, StreamResult, StreamResultRouter
+
+from subunit import (
+    DiscardStream, ProtocolTestCase, ByteStreamToStreamResult,
+    StreamResultToBytes,
+    )
+from subunit.test_results import CatFiles
 
 
 def make_options(description):
@@ -31,33 +38,75 @@
         help="Send the output to this path rather than stdout.")
     parser.add_option(
         "-f", "--forward", action="store_true", default=False,
-        help="Forward subunit stream on stdout.")
+        help="Forward subunit stream on stdout. When set, received "
+            "non-subunit output will be encapsulated in subunit.")
     return parser
 
 
 def run_tests_from_stream(input_stream, result, passthrough_stream=None,
-                          forward_stream=None):
+    forward_stream=None, protocol_version=1, passthrough_subunit=True):
     """Run tests from a subunit input stream through 'result'.
 
+    Non-test events - top level file attachments - are expected to be
+    dropped by v2 StreamResults at the present time (as all the analysis code
+    is in ExtendedTestResult API's), so to implement passthrough_stream they
+    are diverted and copied directly when that is set.
+
     :param input_stream: A stream containing subunit input.
     :param result: A TestResult that will receive the test events.
+        NB: This should be an ExtendedTestResult for v1 and a StreamResult for
+        v2.
     :param passthrough_stream: All non-subunit input received will be
         sent to this stream.  If not provided, uses the ``TestProtocolServer``
         default, which is ``sys.stdout``.
     :param forward_stream: All subunit input received will be forwarded
-        to this stream.  If not provided, uses the ``TestProtocolServer``
-        default, which is to not forward any input.
+        to this stream. If not provided, uses the ``TestProtocolServer``
+        default, which is to not forward any input. Do not set this when
+        transforming the stream - items would be double-reported.
+    :param protocol_version: What version of the subunit protocol to expect.
+    :param passthrough_subunit: If True, passthrough should be as subunit
+        otherwise unwrap it. Only has effect when forward_stream is None.
+        (when forwarding as subunit non-subunit input is always turned into
+        subunit)
     """
-    test = ProtocolTestCase(
-        input_stream, passthrough=passthrough_stream,
-        forward=forward_stream)
+    if 1==protocol_version:
+        test = ProtocolTestCase(
+            input_stream, passthrough=passthrough_stream,
+            forward=forward_stream)
+    elif 2==protocol_version:
+        # In all cases we encapsulate unknown inputs.
+        if forward_stream is not None:
+            # Send events to forward_stream as subunit.
+            forward_result = StreamResultToBytes(forward_stream)
+            # If we're passing non-subunit through, copy:
+            if passthrough_stream is None:
+                # Not passing non-test events - split them off to nothing.
+                router = StreamResultRouter(forward_result)
+                router.map(StreamResult(), 'test_id', test_id=None)
+                result = CopyStreamResult([router, result])
+            else:
+                # otherwise, copy all events to forward_result
+                result = CopyStreamResult([forward_result, result])
+        elif passthrough_stream is not None:
+            if not passthrough_subunit:
+                # Route non-test events to passthrough_stream, unwrapping them for
+                # display.
+                passthrough_result = CatFiles(passthrough_stream)
+            else:
+                passthrough_result = StreamResultToBytes(passthrough_stream)
+            result = StreamResultRouter(result)
+            result.map(passthrough_result, 'test_id', test_id=None)
+        test = ByteStreamToStreamResult(input_stream,
+            non_subunit_name='stdout')
+    else:
+        raise Exception("Unknown protocol version.")
     result.startTestRun()
     test.run(result)
     result.stopTestRun()
 
 
 def filter_by_result(result_factory, output_path, passthrough, forward,
-                     input_stream=sys.stdin):
+                     input_stream=sys.stdin, protocol_version=1):
     """Filter an input stream using a test result.
 
     :param result_factory: A callable that when passed an output stream
@@ -71,17 +120,23 @@
         ``sys.stdout`` as well as to the ``TestResult``.
     :param input_stream: The source of subunit input.  Defaults to
         ``sys.stdin``.
-    :return: A test result with the resultts of the run.
+    :param protocol_version: The subunit protocol version to expect.
+    :return: A test result with the results of the run.
     """
     if passthrough:
         passthrough_stream = sys.stdout
     else:
-        passthrough_stream = DiscardStream()
+        if 1==protocol_version:
+            passthrough_stream = DiscardStream()
+        else:
+            passthrough_stream = None
 
     if forward:
         forward_stream = sys.stdout
-    else:
+    elif 1==protocol_version:
         forward_stream = DiscardStream()
+    else:
+        forward_stream = None
 
     if output_path is None:
         output_to = sys.stdout
@@ -91,14 +146,16 @@
     try:
         result = result_factory(output_to)
         run_tests_from_stream(
-            input_stream, result, passthrough_stream, forward_stream)
+            input_stream, result, passthrough_stream, forward_stream,
+            protocol_version=protocol_version)
     finally:
         if output_path:
             output_to.close()
     return result
 
 
-def run_filter_script(result_factory, description, post_run_hook=None):
+def run_filter_script(result_factory, description, post_run_hook=None,
+    protocol_version=1):
     """Main function for simple subunit filter scripts.
 
     Many subunit filter scripts take a stream of subunit input and use a
@@ -111,14 +168,17 @@
     :param result_factory: A callable that takes an output stream and returns
         a test result that outputs to that stream.
     :param description: A description of the filter script.
+    :param protocol_version: What protocol version to consume/emit.
     """
     parser = make_options(description)
     (options, args) = parser.parse_args()
     result = filter_by_result(
         result_factory, options.output_to, not options.no_passthrough,
-        options.forward)
+        options.forward, protocol_version=protocol_version)
     if post_run_hook:
         post_run_hook(result)
+    if not safe_hasattr(result, 'wasSuccessful'):
+        result = result.decorated
     if result.wasSuccessful():
         sys.exit(0)
     else:
 
=== modified file 'python/subunit/run.py'
--- python/subunit/run.py	2012-12-17 07:24:28 +0000
+++ python/subunit/run.py	2013-03-31 05:51:20 +0000
@@ -20,9 +20,14 @@
   $ python -m subunit.run mylib.tests.test_suite
 """
 
+import io
+import os
 import sys
 
-from subunit import TestProtocolClient, get_default_formatter
+from testtools import ExtendedToStreamDecorator
+from testtools.testsuite import iterate_tests
+
+from subunit import StreamResultToBytes, get_default_formatter
 from subunit.test_results import AutoTimingTestResultDecorator
 from testtools.run import (
     BUFFEROUTPUT,
@@ -46,11 +51,34 @@
 
     def run(self, test):
         "Run the given test case or test suite."
-        result = TestProtocolClient(self.stream)
+        result = self._list(test)
+        result = ExtendedToStreamDecorator(result)
         result = AutoTimingTestResultDecorator(result)
         if self.failfast is not None:
             result.failfast = self.failfast
-        test(result)
+        result.startTestRun()
+        try:
+            test(result)
+        finally:
+            result.stopTestRun()
+        return result
+
+    def list(self, test):
+        "List the test."
+        self._list(test)
+
+    def _list(self, test):
+        try:
+            fileno = self.stream.fileno()
+        except:
+            fileno = None
+        if fileno is not None:
+            stream = os.fdopen(fileno, 'wb', 0)
+        else:
+            stream = self.stream
+        result = StreamResultToBytes(stream)
+        for case in iterate_tests(test):
+            result.status(test_id=case.id(), test_status='exists')
         return result
 
 
@@ -78,7 +106,15 @@
 
 
 if __name__ == '__main__':
+    # Disable the default buffering, for Python 2.x where pdb doesn't do it
+    # on non-ttys.
     stream = get_default_formatter()
     runner = SubunitTestRunner
+    # Patch stdout to be unbuffered, so that pdb works well on 2.6/2.7.
+    binstdout = io.open(sys.stdout.fileno(), 'wb', 0)
+    if sys.version_info[0] > 2:
+        sys.stdout = io.TextIOWrapper(binstdout, encoding=sys.stdout.encoding)
+    else:
+        sys.stdout = binstdout
     SubunitTestProgram(module=None, argv=sys.argv, testRunner=runner,
         stdout=sys.stdout)
 
=== modified file 'python/subunit/test_results.py'
--- python/subunit/test_results.py	2012-12-17 07:24:28 +0000
+++ python/subunit/test_results.py	2013-03-31 05:51:20 +0000
@@ -25,8 +25,10 @@
     text_content,
     TracebackContent,
     )
+from testtools import StreamResult
 
 from subunit import iso8601
+import subunit
 
 
 # NOT a TestResult, because we are implementing the interface, not inheriting
@@ -525,16 +527,24 @@
 
 
 class TestIdPrintingResult(testtools.TestResult):
-
-    def __init__(self, stream, show_times=False):
+    """Print test ids to a stream.
+
+    Implements both TestResult and StreamResult, for compatibility.
+    """
+
+    def __init__(self, stream, show_times=False, show_exists=False):
         """Create a FilterResult object outputting to stream."""
         super(TestIdPrintingResult, self).__init__()
         self._stream = stream
+        self.show_exists = show_exists
+        self.show_times = show_times
+
+    def startTestRun(self):
         self.failed_tests = 0
         self.__time = None
-        self.show_times = show_times
         self._test = None
         self._test_duration = 0
+        self._active_tests = {}
 
     def addError(self, test, err):
         self.failed_tests += 1
@@ -557,21 +567,44 @@
     def addExpectedFailure(self, test, err=None, details=None):
         self._test = test
 
-    def reportTest(self, test, duration):
+    def reportTest(self, test_id, duration):
         if self.show_times:
             seconds = duration.seconds
             seconds += duration.days * 3600 * 24
             seconds += duration.microseconds / 1000000.0
-            self._stream.write(test.id() + ' %0.3f\n' % seconds)
+            self._stream.write(test_id + ' %0.3f\n' % seconds)
         else:
-            self._stream.write(test.id() + '\n')
+            self._stream.write(test_id + '\n')
 
     def startTest(self, test):
         self._start_time = self._time()
 
+    def status(self, test_id=None, test_status=None, test_tags=None,
+        runnable=True, file_name=None, file_bytes=None, eof=False,
+        mime_type=None, route_code=None, timestamp=None):
+        if not test_id:
+            return
+        if timestamp is not None:
+            self.time(timestamp)
+        if test_status=='exists':
+            if self.show_exists:
+                self.reportTest(test_id, 0)
+        elif test_status in ('inprogress', None):
+            self._active_tests[test_id] = self._time()
+        else:
+            self._end_test(test_id)
+
+    def _end_test(self, test_id):
+        test_start = self._active_tests.pop(test_id, None)
+        if not test_start:
+            test_duration = 0
+        else:
+            test_duration = self._time() - test_start
+        self.reportTest(test_id, test_duration)
+
     def stopTest(self, test):
         test_duration = self._time() - self._start_time
-        self.reportTest(self._test, test_duration)
+        self.reportTest(self._test.id(), test_duration)
 
     def time(self, time):
         self.__time = time
@@ -583,6 +616,10 @@
         "Tells whether or not this result was a success"
         return self.failed_tests == 0
 
+    def stopTestRun(self):
+        for test_id in list(self._active_tests.keys()):
+            self._end_test(test_id)
+
 
 class TestByTestResult(testtools.TestResult):
     """Call something every time a test completes."""
@@ -676,3 +713,17 @@
     def startTestRun(self):
         super(CsvResult, self).startTestRun()
         self._write_row(['test', 'status', 'start_time', 'stop_time'])
+
+
+class CatFiles(StreamResult):
+    """Cat file attachments received to a stream."""
+
+    def __init__(self, byte_stream):
+        self.stream = subunit.make_stream_binary(byte_stream)
+
+    def status(self, test_id=None, test_status=None, test_tags=None,
+        runnable=True, file_name=None, file_bytes=None, eof=False,
+        mime_type=None, route_code=None, timestamp=None):
+        if file_name is not None:
+            self.stream.write(file_bytes)
+            self.stream.flush()
 
=== modified file 'python/subunit/tests/__init__.py'
--- python/subunit/tests/__init__.py	2011-11-01 15:59:58 +0000
+++ python/subunit/tests/__init__.py	2013-03-31 05:51:20 +0000
@@ -25,6 +25,7 @@
     test_subunit_tags,
     test_tap2subunit,
     test_test_protocol,
+    test_test_protocol2,
     test_test_results,
     )
 
@@ -35,6 +36,7 @@
     result.addTest(test_progress_model.test_suite())
     result.addTest(test_test_results.test_suite())
     result.addTest(test_test_protocol.test_suite())
+    result.addTest(test_test_protocol2.test_suite())
     result.addTest(test_tap2subunit.test_suite())
     result.addTest(test_subunit_filter.test_suite())
     result.addTest(test_subunit_tags.test_suite())
 
=== modified file 'python/subunit/tests/test_run.py'
--- python/subunit/tests/test_run.py	2012-05-07 19:36:05 +0000
+++ python/subunit/tests/test_run.py	2013-03-31 05:51:20 +0000
@@ -18,6 +18,7 @@
 import unittest
 
 from testtools import PlaceHolder
+from testtools.testresult.doubles import StreamResult
 
 import subunit
 from subunit.run import SubunitTestRunner
@@ -29,16 +30,6 @@
     return result
 
 
-class TimeCollectingTestResult(unittest.TestResult):
-
-    def __init__(self, *args, **kwargs):
-        super(TimeCollectingTestResult, self).__init__(*args, **kwargs)
-        self.time_called = []
-
-    def time(self, a_time):
-        self.time_called.append(a_time)
-
-
 class TestSubunitTestRunner(unittest.TestCase):
 
     def test_includes_timing_output(self):
@@ -46,7 +37,24 @@
         runner = SubunitTestRunner(stream=io)
         test = PlaceHolder('name')
         runner.run(test)
-        client = TimeCollectingTestResult()
-        io.seek(0)
-        subunit.TestProtocolServer(client).readFrom(io)
-        self.assertTrue(len(client.time_called) > 0)
+        io.seek(0)
+        eventstream = StreamResult()
+        subunit.ByteStreamToStreamResult(io).run(eventstream)
+        timestamps = [event[-1] for event in eventstream._events
+            if event is not None]
+        self.assertNotEqual([], timestamps)
+
+    def test_enumerates_tests_before_run(self):
+        io = BytesIO()
+        runner = SubunitTestRunner(stream=io)
+        test1 = PlaceHolder('name1')
+        test2 = PlaceHolder('name2')
+        case = unittest.TestSuite([test1, test2])
+        runner.run(case)
+        io.seek(0)
+        eventstream = StreamResult()
+        subunit.ByteStreamToStreamResult(io).run(eventstream)
+        self.assertEqual([
+            ('status', 'name1', 'exists'),
+            ('status', 'name2', 'exists'),
+            ], [event[:3] for event in eventstream._events[:2]])
 
=== modified file 'python/subunit/tests/test_subunit_filter.py'
--- python/subunit/tests/test_subunit_filter.py	2012-05-07 22:53:53 +0000
+++ python/subunit/tests/test_subunit_filter.py	2013-03-31 05:51:20 +0000
@@ -25,10 +25,11 @@
 
 from testtools import TestCase
 from testtools.compat import _b, BytesIO
-from testtools.testresult.doubles import ExtendedTestResult
+from testtools.testresult.doubles import ExtendedTestResult, StreamResult
 
 import subunit
 from subunit.test_results import make_tag_filter, TestResultFilter
+from subunit import ByteStreamToStreamResult, StreamResultToBytes
 
 
 class TestTestResultFilter(TestCase):
@@ -286,23 +287,6 @@
 
 class TestFilterCommand(TestCase):
 
-    example_subunit_stream = _b("""\
-tags: global
-test passed
-success passed
-test failed
-tags: local
-failure failed
-test error
-error error [
-error details
-]
-test skipped
-skip skipped
-test todo
-xfail todo
-""")
-
     def run_command(self, args, stream):
         root = os.path.dirname(
             os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
@@ -316,52 +300,50 @@
             raise RuntimeError("%s failed: %s" % (command, err))
         return out
 
-    def to_events(self, stream):
-        test = subunit.ProtocolTestCase(BytesIO(stream))
-        result = ExtendedTestResult()
-        test.run(result)
-        return result._events
-
     def test_default(self):
-        output = self.run_command([], _b(
-                "test: foo\n"
-                "skip: foo\n"
-                ))
-        events = self.to_events(output)
-        foo = subunit.RemotedTestCase('foo')
-        self.assertEqual(
-            [('startTest', foo),
-             ('addSkip', foo, {}),
-             ('stopTest', foo)],
-            events)
+        byte_stream = BytesIO()
+        stream = StreamResultToBytes(byte_stream)
+        stream.status(test_id="foo", test_status="inprogress")
+        stream.status(test_id="foo", test_status="skip")
+        output = self.run_command([], byte_stream.getvalue())
+        events = StreamResult()
+        ByteStreamToStreamResult(BytesIO(output)).run(events)
+        ids = set(event[1] for event in events._events)
+        self.assertEqual([
+            ('status', 'foo', 'inprogress'),
+            ('status', 'foo', 'skip'),
+            ], [event[:3] for event in events._events])
 
     def test_tags(self):
-        output = self.run_command(['-s', '--with-tag', 'a'], _b(
-                "tags: a\n"
-                "test: foo\n"
-                "success: foo\n"
-                "tags: -a\n"
-                "test: bar\n"
-                "success: bar\n"
-                "test: baz\n"
-                "tags: a\n"
-                "success: baz\n"
-                ))
-        events = self.to_events(output)
-        foo = subunit.RemotedTestCase('foo')
-        baz = subunit.RemotedTestCase('baz')
-        self.assertEqual(
-            [('tags', set(['a']), set()),
-             ('startTest', foo),
-             ('addSuccess', foo),
-             ('stopTest', foo),
-             ('tags', set(), set(['a'])),
-             ('startTest', baz),
-             ('tags', set(['a']), set()),
-             ('addSuccess', baz),
-             ('stopTest', baz),
-             ],
-            events)
+        byte_stream = BytesIO()
+        stream = StreamResultToBytes(byte_stream)
+        stream.status(
+            test_id="foo", test_status="inprogress", test_tags=set(["a"]))
+        stream.status(
+            test_id="foo", test_status="success", test_tags=set(["a"]))
+        stream.status(test_id="bar", test_status="inprogress")
+        stream.status(test_id="bar", test_status="inprogress")
+        stream.status(
+            test_id="baz", test_status="inprogress", test_tags=set(["a"]))
+        stream.status(
+            test_id="baz", test_status="success", test_tags=set(["a"]))
+        output = self.run_command(
+            ['-s', '--with-tag', 'a'], byte_stream.getvalue())
+        events = StreamResult()
+        ByteStreamToStreamResult(BytesIO(output)).run(events)
+        ids = set(event[1] for event in events._events)
+        self.assertEqual(set(['foo', 'baz']), ids)
+
+    def test_no_passthrough(self):
+        output = self.run_command(['--no-passthrough'], b'hi thar')
+        self.assertEqual(b'', output)
+
+    def test_passthrough(self):
+        output = self.run_command([], b'hi thar')
+        byte_stream = BytesIO()
+        stream = StreamResultToBytes(byte_stream)
+        stream.status(file_name="stdout", file_bytes=b'hi thar')
+        self.assertEqual(byte_stream.getvalue(), output)
 
 
 def test_suite():
 
=== modified file 'python/subunit/tests/test_subunit_tags.py'
--- python/subunit/tests/test_subunit_tags.py	2011-04-24 21:40:52 +0000
+++ python/subunit/tests/test_subunit_tags.py	2013-03-31 05:51:20 +0000
@@ -16,10 +16,9 @@
 
 """Tests for subunit.tag_stream."""
 
+from io import BytesIO
 import unittest
 
-from testtools.compat import StringIO
-
 import subunit
 import subunit.test_results
 
@@ -27,40 +26,42 @@
 class TestSubUnitTags(unittest.TestCase):
 
     def setUp(self):
-        self.original = StringIO()
-        self.filtered = StringIO()
+        self.original = BytesIO()
+        self.filtered = BytesIO()
 
     def test_add_tag(self):
-        self.original.write("tags: foo\n")
-        self.original.write("test: test\n")
-        self.original.write("tags: bar -quux\n")
-        self.original.write("success: test\n")
+        reference = BytesIO()
+        stream = subunit.StreamResultToBytes(reference)
+        stream.status(
+            test_id='test', test_status='inprogress', test_tags=set(['quux', 'foo']))
+        stream.status(
+            test_id='test', test_status='success', test_tags=set(['bar', 'quux', 'foo']))
+        stream = subunit.StreamResultToBytes(self.original)
+        stream.status(
+            test_id='test', test_status='inprogress', test_tags=set(['foo']))
+        stream.status(
+            test_id='test', test_status='success', test_tags=set(['foo', 'bar']))
         self.original.seek(0)
-        result = subunit.tag_stream(self.original, self.filtered, ["quux"])
-        self.assertEqual([
-            "tags: quux",
-            "tags: foo",
-            "test: test",
-            "tags: bar",
-            "success: test",
-            ],
-            self.filtered.getvalue().splitlines())
+        self.assertEqual(
+            0, subunit.tag_stream(self.original, self.filtered, ["quux"]))
+        self.assertEqual(reference.getvalue(), self.filtered.getvalue())
 
     def test_remove_tag(self):
-        self.original.write("tags: foo\n")
-        self.original.write("test: test\n")
-        self.original.write("tags: bar -quux\n")
-        self.original.write("success: test\n")
+        reference = BytesIO()
+        stream = subunit.StreamResultToBytes(reference)
+        stream.status(
+            test_id='test', test_status='inprogress', test_tags=set(['foo']))
+        stream.status(
+            test_id='test', test_status='success', test_tags=set(['foo']))
+        stream = subunit.StreamResultToBytes(self.original)
+        stream.status(
+            test_id='test', test_status='inprogress', test_tags=set(['foo']))
+        stream.status(
+            test_id='test', test_status='success', test_tags=set(['foo', 'bar']))
         self.original.seek(0)
-        result = subunit.tag_stream(self.original, self.filtered, ["-bar"])
-        self.assertEqual([
-            "tags: -bar",
-            "tags: foo",
-            "test: test",
-            "tags: -quux",
-            "success: test",
-            ],
-            self.filtered.getvalue().splitlines())
+        self.assertEqual(
+            0, subunit.tag_stream(self.original, self.filtered, ["-bar"]))
+        self.assertEqual(reference.getvalue(), self.filtered.getvalue())
 
 
 def test_suite():
 
=== modified file 'python/subunit/tests/test_tap2subunit.py'
--- python/subunit/tests/test_tap2subunit.py	2011-04-24 21:40:52 +0000
+++ python/subunit/tests/test_tap2subunit.py	2013-03-31 05:51:20 +0000
@@ -16,14 +16,19 @@
 
 """Tests for TAP2SubUnit."""
 
+from io import BytesIO, StringIO
 import unittest
 
-from testtools.compat import StringIO
+from testtools import TestCase
+from testtools.compat import _u
+from testtools.testresult.doubles import StreamResult
 
 import subunit
 
-
-class TestTAP2SubUnit(unittest.TestCase):
+UTF8_TEXT = 'text/plain; charset=UTF8'
+
+
+class TestTAP2SubUnit(TestCase):
     """Tests for TAP2SubUnit.
 
     These tests test TAP string data in, and subunit string data out.
@@ -34,24 +39,21 @@
     """
 
     def setUp(self):
+        super(TestTAP2SubUnit, self).setUp()
         self.tap = StringIO()
-        self.subunit = StringIO()
+        self.subunit = BytesIO()
 
     def test_skip_entire_file(self):
         # A file
         # 1..- # Skipped: comment
         # results in a single skipped test.
-        self.tap.write("1..0 # Skipped: entire file skipped\n")
+        self.tap.write(_u("1..0 # Skipped: entire file skipped\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test file skip",
-            "skip file skip [",
-            "Skipped: entire file skipped",
-            "]",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'file skip', 'skip', None, True,
+            'tap comment', b'Skipped: entire file skipped', True, None, None,
+            None)])
 
     def test_ok_test_pass(self):
         # A file
@@ -59,164 +61,128 @@
         # results in a passed test with name 'test 1' (a synthetic name as tap
         # does not require named fixtures - it is the first test in the tap
         # stream).
-        self.tap.write("ok\n")
+        self.tap.write(_u("ok\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1",
-            "success test 1",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1', 'success', None, False, None,
+            None, True, None, None, None)])
 
     def test_ok_test_number_pass(self):
         # A file
         # ok 1
         # results in a passed test with name 'test 1'
-        self.tap.write("ok 1\n")
+        self.tap.write(_u("ok 1\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1",
-            "success test 1",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1', 'success', None, False, None,
+            None, True, None, None, None)])
 
     def test_ok_test_number_description_pass(self):
         # A file
         # ok 1 - There is a description
         # results in a passed test with name 'test 1 - There is a description'
-        self.tap.write("ok 1 - There is a description\n")
+        self.tap.write(_u("ok 1 - There is a description\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1 - There is a description",
-            "success test 1 - There is a description",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1 - There is a description',
+            'success', None, False, None, None, True, None, None, None)])
 
     def test_ok_test_description_pass(self):
         # A file
         # ok There is a description
         # results in a passed test with name 'test 1 There is a description'
-        self.tap.write("ok There is a description\n")
+        self.tap.write(_u("ok There is a description\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1 There is a description",
-            "success test 1 There is a description",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1 There is a description',
+            'success', None, False, None, None, True, None, None, None)])
 
     def test_ok_SKIP_skip(self):
         # A file
         # ok # SKIP
         # results in a skkip test with name 'test 1'
-        self.tap.write("ok # SKIP\n")
+        self.tap.write(_u("ok # SKIP\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1",
-            "skip test 1",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1', 'skip', None, False, None,
+            None, True, None, None, None)])
 
     def test_ok_skip_number_comment_lowercase(self):
-        self.tap.write("ok 1 # skip no samba environment available, skipping compilation\n")
+        self.tap.write(_u("ok 1 # skip no samba environment available, skipping compilation\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1",
-            "skip test 1 [", 
-            "no samba environment available, skipping compilation",
-            "]"
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1', 'skip', None, False, 'tap comment',
+            b'no samba environment available, skipping compilation', True,
+            'text/plain; charset=UTF8', None, None)])
 
     def test_ok_number_description_SKIP_skip_comment(self):
         # A file
         # ok 1 foo  # SKIP Not done yet
         # results in a skip test with name 'test 1 foo' and a log of
         # Not done yet
-        self.tap.write("ok 1 foo  # SKIP Not done yet\n")
+        self.tap.write(_u("ok 1 foo  # SKIP Not done yet\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1 foo",
-            "skip test 1 foo [",
-            "Not done yet",
-            "]",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1 foo', 'skip', None, False,
+            'tap comment', b'Not done yet', True, 'text/plain; charset=UTF8',
+            None, None)])
 
     def test_ok_SKIP_skip_comment(self):
         # A file
         # ok # SKIP Not done yet
         # results in a skip test with name 'test 1' and a log of Not done yet
-        self.tap.write("ok # SKIP Not done yet\n")
+        self.tap.write(_u("ok # SKIP Not done yet\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1",
-            "skip test 1 [",
-            "Not done yet",
-            "]",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1', 'skip', None, False,
+            'tap comment', b'Not done yet', True, 'text/plain; charset=UTF8',
+            None, None)])
 
     def test_ok_TODO_xfail(self):
         # A file
         # ok # TODO
         # results in a xfail test with name 'test 1'
-        self.tap.write("ok # TODO\n")
+        self.tap.write(_u("ok # TODO\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1",
-            "xfail test 1",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1', 'xfail', None, False, None,
+            None, True, None, None, None)])
 
     def test_ok_TODO_xfail_comment(self):
         # A file
         # ok # TODO Not done yet
         # results in a xfail test with name 'test 1' and a log of Not done yet
-        self.tap.write("ok # TODO Not done yet\n")
+        self.tap.write(_u("ok # TODO Not done yet\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1",
-            "xfail test 1 [",
-            "Not done yet",
-            "]",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([('status', 'test 1', 'xfail', None, False,
+            'tap comment', b'Not done yet', True, 'text/plain; charset=UTF8',
+            None, None)])
 
     def test_bail_out_errors(self):
         # A file with line in it
         # Bail out! COMMENT
         # is treated as an error
-        self.tap.write("ok 1 foo\n")
-        self.tap.write("Bail out! Lifejacket engaged\n")
+        self.tap.write(_u("ok 1 foo\n"))
+        self.tap.write(_u("Bail out! Lifejacket engaged\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            "test test 1 foo",
-            "success test 1 foo",
-            "test Bail out! Lifejacket engaged",
-            "error Bail out! Lifejacket engaged",
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1 foo', 'success', None, False, None, None, True,
+             None, None, None),
+            ('status', 'Bail out! Lifejacket engaged', 'fail', None, False,
+             None, None, True, None, None, None)])
 
     def test_missing_test_at_end_with_plan_adds_error(self):
         # A file
@@ -224,23 +190,20 @@
         # ok first test
         # not ok third test
         # results in three tests, with the third being created
-        self.tap.write('1..3\n')
-        self.tap.write('ok first test\n')
-        self.tap.write('not ok second test\n')
+        self.tap.write(_u('1..3\n'))
+        self.tap.write(_u('ok first test\n'))
+        self.tap.write(_u('not ok second test\n'))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1 first test',
-            'success test 1 first test',
-            'test test 2 second test',
-            'failure test 2 second test',
-            'test test 3',
-            'error test 3 [',
-            'test missing from TAP output',
-            ']',
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1 first test', 'success', None, False, None,
+             None, True, None, None, None),
+            ('status', 'test 2 second test', 'fail', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 3', 'fail', None, False, 'tap meta',
+             b'test missing from TAP output', True, 'text/plain; charset=UTF8',
+             None, None)])
 
     def test_missing_test_with_plan_adds_error(self):
         # A file
@@ -248,45 +211,39 @@
         # ok first test
         # not ok 3 third test
         # results in three tests, with the second being created
-        self.tap.write('1..3\n')
-        self.tap.write('ok first test\n')
-        self.tap.write('not ok 3 third test\n')
+        self.tap.write(_u('1..3\n'))
+        self.tap.write(_u('ok first test\n'))
+        self.tap.write(_u('not ok 3 third test\n'))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1 first test',
-            'success test 1 first test',
-            'test test 2',
-            'error test 2 [',
-            'test missing from TAP output',
-            ']',
-            'test test 3 third test',
-            'failure test 3 third test',
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1 first test', 'success', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 2', 'fail', None, False, 'tap meta',
+             b'test missing from TAP output', True, 'text/plain; charset=UTF8',
+             None, None),
+            ('status', 'test 3 third test', 'fail', None, False, None, None,
+             True, None, None, None)])
 
     def test_missing_test_no_plan_adds_error(self):
         # A file
         # ok first test
         # not ok 3 third test
         # results in three tests, with the second being created
-        self.tap.write('ok first test\n')
-        self.tap.write('not ok 3 third test\n')
+        self.tap.write(_u('ok first test\n'))
+        self.tap.write(_u('not ok 3 third test\n'))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1 first test',
-            'success test 1 first test',
-            'test test 2',
-            'error test 2 [',
-            'test missing from TAP output',
-            ']',
-            'test test 3 third test',
-            'failure test 3 third test',
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1 first test', 'success', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 2', 'fail', None, False, 'tap meta',
+             b'test missing from TAP output', True, 'text/plain; charset=UTF8',
+             None, None),
+            ('status', 'test 3 third test', 'fail', None, False, None, None,
+             True, None, None, None)])
 
     def test_four_tests_in_a_row_trailing_plan(self):
         # A file
@@ -296,25 +253,23 @@
         # not ok 4 - fourth
         # 1..4
         # results in four tests numbered and named
-        self.tap.write('ok 1 - first test in a script with trailing plan\n')
-        self.tap.write('not ok 2 - second\n')
-        self.tap.write('ok 3 - third\n')
-        self.tap.write('not ok 4 - fourth\n')
-        self.tap.write('1..4\n')
+        self.tap.write(_u('ok 1 - first test in a script with trailing plan\n'))
+        self.tap.write(_u('not ok 2 - second\n'))
+        self.tap.write(_u('ok 3 - third\n'))
+        self.tap.write(_u('not ok 4 - fourth\n'))
+        self.tap.write(_u('1..4\n'))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1 - first test in a script with trailing plan',
-            'success test 1 - first test in a script with trailing plan',
-            'test test 2 - second',
-            'failure test 2 - second',
-            'test test 3 - third',
-            'success test 3 - third',
-            'test test 4 - fourth',
-            'failure test 4 - fourth'
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1 - first test in a script with trailing plan',
+             'success', None, False, None, None, True, None, None, None),
+            ('status', 'test 2 - second', 'fail', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 3 - third', 'success', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 4 - fourth', 'fail', None, False, None, None,
+             True, None, None, None)])
 
     def test_four_tests_in_a_row_with_plan(self):
         # A file
@@ -324,25 +279,23 @@
         # ok 3 - third
         # not ok 4 - fourth
         # results in four tests numbered and named
-        self.tap.write('1..4\n')
-        self.tap.write('ok 1 - first test in a script with a plan\n')
-        self.tap.write('not ok 2 - second\n')
-        self.tap.write('ok 3 - third\n')
-        self.tap.write('not ok 4 - fourth\n')
+        self.tap.write(_u('1..4\n'))
+        self.tap.write(_u('ok 1 - first test in a script with a plan\n'))
+        self.tap.write(_u('not ok 2 - second\n'))
+        self.tap.write(_u('ok 3 - third\n'))
+        self.tap.write(_u('not ok 4 - fourth\n'))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1 - first test in a script with a plan',
-            'success test 1 - first test in a script with a plan',
-            'test test 2 - second',
-            'failure test 2 - second',
-            'test test 3 - third',
-            'success test 3 - third',
-            'test test 4 - fourth',
-            'failure test 4 - fourth'
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1 - first test in a script with a plan',
+             'success', None, False, None, None, True, None, None, None),
+            ('status', 'test 2 - second', 'fail', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 3 - third', 'success', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 4 - fourth', 'fail', None, False, None, None,
+             True, None, None, None)])
 
     def test_four_tests_in_a_row_no_plan(self):
         # A file
@@ -351,46 +304,43 @@
         # ok 3 - third
         # not ok 4 - fourth
         # results in four tests numbered and named
-        self.tap.write('ok 1 - first test in a script with no plan at all\n')
-        self.tap.write('not ok 2 - second\n')
-        self.tap.write('ok 3 - third\n')
-        self.tap.write('not ok 4 - fourth\n')
+        self.tap.write(_u('ok 1 - first test in a script with no plan at all\n'))
+        self.tap.write(_u('not ok 2 - second\n'))
+        self.tap.write(_u('ok 3 - third\n'))
+        self.tap.write(_u('not ok 4 - fourth\n'))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1 - first test in a script with no plan at all',
-            'success test 1 - first test in a script with no plan at all',
-            'test test 2 - second',
-            'failure test 2 - second',
-            'test test 3 - third',
-            'success test 3 - third',
-            'test test 4 - fourth',
-            'failure test 4 - fourth'
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1 - first test in a script with no plan at all',
+             'success', None, False, None, None, True, None, None, None),
+            ('status', 'test 2 - second', 'fail', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 3 - third', 'success', None, False, None, None,
+             True, None, None, None),
+            ('status', 'test 4 - fourth', 'fail', None, False, None, None,
+             True, None, None, None)])
 
     def test_todo_and_skip(self):
         # A file
         # not ok 1 - a fail but # TODO but is TODO
         # not ok 2 - another fail # SKIP instead
         # results in two tests, numbered and commented.
-        self.tap.write("not ok 1 - a fail but # TODO but is TODO\n")
-        self.tap.write("not ok 2 - another fail # SKIP instead\n")
+        self.tap.write(_u("not ok 1 - a fail but # TODO but is TODO\n"))
+        self.tap.write(_u("not ok 2 - another fail # SKIP instead\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1 - a fail but',
-            'xfail test 1 - a fail but [',
-            'but is TODO',
-            ']',
-            'test test 2 - another fail',
-            'skip test 2 - another fail [',
-            'instead',
-            ']',
-            ],
-            self.subunit.getvalue().splitlines())
+        self.subunit.seek(0)
+        events = StreamResult()
+        subunit.ByteStreamToStreamResult(self.subunit).run(events)
+        self.check_events([
+            ('status', 'test 1 - a fail but', 'xfail', None, False,
+             'tap comment', b'but is TODO', True, 'text/plain; charset=UTF8',
+             None, None),
+            ('status', 'test 2 - another fail', 'skip', None, False,
+             'tap comment', b'instead', True, 'text/plain; charset=UTF8',
+             None, None)])
 
     def test_leading_comments_add_to_next_test_log(self):
         # A file
@@ -399,21 +349,17 @@
         # ok
         # results in a single test with the comment included
         # in the first test and not the second.
-        self.tap.write("# comment\n")
-        self.tap.write("ok\n")
-        self.tap.write("ok\n")
+        self.tap.write(_u("# comment\n"))
+        self.tap.write(_u("ok\n"))
+        self.tap.write(_u("ok\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1',
-            'success test 1 [',
-            '# comment',
-            ']',
-            'test test 2',
-            'success test 2',
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1', 'success', None, False, 'tap comment',
+             b'# comment', True, 'text/plain; charset=UTF8', None, None),
+            ('status', 'test 2', 'success', None, False, None, None, True,
+             None, None, None)])
     
     def test_trailing_comments_are_included_in_last_test_log(self):
         # A file
@@ -422,21 +368,23 @@
         # # comment
         # results in a two tests, with the second having the comment
         # attached to its log.
-        self.tap.write("ok\n")
-        self.tap.write("ok\n")
-        self.tap.write("# comment\n")
+        self.tap.write(_u("ok\n"))
+        self.tap.write(_u("ok\n"))
+        self.tap.write(_u("# comment\n"))
         self.tap.seek(0)
         result = subunit.TAP2SubUnit(self.tap, self.subunit)
         self.assertEqual(0, result)
-        self.assertEqual([
-            'test test 1',
-            'success test 1',
-            'test test 2',
-            'success test 2 [',
-            '# comment',
-            ']',
-            ],
-            self.subunit.getvalue().splitlines())
+        self.check_events([
+            ('status', 'test 1', 'success', None, False, None, None, True,
+             None, None, None),
+            ('status', 'test 2', 'success', None, False, 'tap comment',
+             b'# comment', True, 'text/plain; charset=UTF8', None, None)])
+
+    def check_events(self, events):
+        self.subunit.seek(0)
+        eventstream = StreamResult()
+        subunit.ByteStreamToStreamResult(self.subunit).run(eventstream)
+        self.assertEqual(events, eventstream._events)
 
 
 def test_suite():
 
=== added file 'python/subunit/tests/test_test_protocol2.py'
--- python/subunit/tests/test_test_protocol2.py	1970-01-01 00:00:00 +0000
+++ python/subunit/tests/test_test_protocol2.py	2013-03-31 05:51:20 +0000
@@ -0,0 +1,415 @@
+#
+#  subunit: extensions to Python unittest to get test results from subprocesses.
+#  Copyright (C) 2013  Robert Collins <[e-mail]>
+#
+#  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
+#  license at the users choice. A copy of both licenses are available in the
+#  project source as Apache-2.0 and BSD. You may not use this file except in
+#  compliance with one of these two licences.
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+#  license you chose for the specific language governing permissions and
+#  limitations under that license.
+#
+
+from io import BytesIO
+import datetime
+
+from testtools import TestCase
+from testtools.matchers import HasLength
+from testtools.tests.test_testresult import TestStreamResultContract
+from testtools.testresult.doubles import StreamResult
+
+import subunit
+import subunit.iso8601 as iso8601
+
+CONSTANT_ENUM = b'\xb3)\x01\x0c\x03foo\x08U_\x1b'
+CONSTANT_INPROGRESS = b'\xb3)\x02\x0c\x03foo\x8e\xc1-\xb5'
+CONSTANT_SUCCESS = b'\xb3)\x03\x0c\x03fooE\x9d\xfe\x10'
+CONSTANT_UXSUCCESS = b'\xb3)\x04\x0c\x03fooX\x98\xce\xa8'
+CONSTANT_SKIP = b'\xb3)\x05\x0c\x03foo\x93\xc4\x1d\r'
+CONSTANT_FAIL = b'\xb3)\x06\x0c\x03foo\x15Po\xa3'
+CONSTANT_XFAIL = b'\xb3)\x07\x0c\x03foo\xde\x0c\xbc\x06'
+CONSTANT_EOF = b'\xb3!\x10\x08S\x15\x88\xdc'
+CONSTANT_FILE_CONTENT = b'\xb3!@\x13\x06barney\x03wooA5\xe3\x8c'
+CONSTANT_MIME = b'\xb3! #\x1aapplication/foo; charset=1x3Q\x15'
+CONSTANT_TIMESTAMP = b'\xb3+\x03\x13<\x17T\xcf\x80\xaf\xc8\x03barI\x96>-'
+CONSTANT_ROUTE_CODE = b'\xb3-\x03\x13\x03bar\x06source\x9cY9\x19'
+CONSTANT_RUNNABLE = b'\xb3(\x03\x0c\x03foo\xe3\xea\xf5\xa4'
+CONSTANT_TAGS = b'\xb3)\x80\x15\x03bar\x02\x03foo\x03barTHn\xb4'
+
+
+class TestStreamResultToBytesContract(TestCase, TestStreamResultContract):
+    """Check that StreamResult behaves as testtools expects."""
+
+    def _make_result(self):
+        return subunit.StreamResultToBytes(BytesIO())
+
+
+class TestStreamResultToBytes(TestCase):
+
+    def _make_result(self):
+        output = BytesIO()
+        return subunit.StreamResultToBytes(output), output
+
+    def test_numbers(self):
+        result = subunit.StreamResultToBytes(BytesIO())
+        packet = []
+        self.assertRaises(Exception, result._write_number, -1, packet)
+        self.assertEqual([], packet)
+        result._write_number(0, packet)
+        self.assertEqual([b'\x00'], packet)
+        del packet[:]
+        result._write_number(63, packet)
+        self.assertEqual([b'\x3f'], packet)
+        del packet[:]
+        result._write_number(64, packet)
+        self.assertEqual([b'\x40\x40'], packet)
+        del packet[:]
+        result._write_number(16383, packet)
+        self.assertEqual([b'\x7f\xff'], packet)
+        del packet[:]
+        result._write_number(16384, packet)
+        self.assertEqual([b'\x80\x40', b'\x00'], packet)
+        del packet[:]
+        result._write_number(4194303, packet)
+        self.assertEqual([b'\xbf\xff', b'\xff'], packet)
+        del packet[:]
+        result._write_number(4194304, packet)
+        self.assertEqual([b'\xc0\x40\x00\x00'], packet)
+        del packet[:]
+        result._write_number(1073741823, packet)
+        self.assertEqual([b'\xff\xff\xff\xff'], packet)
+        del packet[:]
+        self.assertRaises(Exception, result._write_number, 1073741824, packet)
+        self.assertEqual([], packet)
+
+    def test_volatile_length(self):
+        # if the length of the packet data before the length itself is
+        # considered is right on the boundary for length's variable length
+        # encoding, it is easy to get the length wrong by not accounting for
+        # length itself.
+        # that is, the encoder has to ensure that length == sum (length_of_rest
+        # + length_of_length)
+        result, output = self._make_result()
+        # 1 byte short:
+        result.status(file_name="", file_bytes=b'\xff'*0)
+        self.assertThat(output.getvalue(), HasLength(10))
+        self.assertEqual(b'\x0a', output.getvalue()[3:4])
+        output.seek(0)
+        output.truncate()
+        # 1 byte long:
+        result.status(file_name="", file_bytes=b'\xff'*53)
+        self.assertThat(output.getvalue(), HasLength(63))
+        self.assertEqual(b'\x3f', output.getvalue()[3:4])
+        output.seek(0)
+        output.truncate()
+        # 2 bytes short
+        result.status(file_name="", file_bytes=b'\xff'*54)
+        self.assertThat(output.getvalue(), HasLength(65))
+        self.assertEqual(b'\x40\x41', output.getvalue()[3:5])
+        output.seek(0)
+        output.truncate()
+        # 2 bytes long
+        result.status(file_name="", file_bytes=b'\xff'*16371)
+        self.assertThat(output.getvalue(), HasLength(16383))
+        self.assertEqual(b'\x7f\xff', output.getvalue()[3:5])
+        output.seek(0)
+        output.truncate()
+        # 3 bytes short
+        result.status(file_name="", file_bytes=b'\xff'*16372)
+        self.assertThat(output.getvalue(), HasLength(16385))
+        self.assertEqual(b'\x80\x40\x01', output.getvalue()[3:6])
+        output.seek(0)
+        output.truncate()
+        # 3 bytes long
+        result.status(file_name="", file_bytes=b'\xff'*4194289)
+        self.assertThat(output.getvalue(), HasLength(4194303))
+        self.assertEqual(b'\xbf\xff\xff', output.getvalue()[3:6])
+        output.seek(0)
+        output.truncate()
+        self.assertRaises(Exception, result.status, file_name="",
+            file_bytes=b'\xff'*4194290)
+
+    def test_trivial_enumeration(self):
+        result, output = self._make_result()
+        result.status("foo", 'exists')
+        self.assertEqual(CONSTANT_ENUM, output.getvalue())
+
+    def test_inprogress(self):
+        result, output = self._make_result()
+        result.status("foo", 'inprogress')
+        self.assertEqual(CONSTANT_INPROGRESS, output.getvalue())
+
+    def test_success(self):
+        result, output = self._make_result()
+        result.status("foo", 'success')
+        self.assertEqual(CONSTANT_SUCCESS, output.getvalue())
+
+    def test_uxsuccess(self):
+        result, output = self._make_result()
+        result.status("foo", 'uxsuccess')
+        self.assertEqual(CONSTANT_UXSUCCESS, output.getvalue())
+
+    def test_skip(self):
+        result, output = self._make_result()
+        result.status("foo", 'skip')
+        self.assertEqual(CONSTANT_SKIP, output.getvalue())
+
+    def test_fail(self):
+        result, output = self._make_result()
+        result.status("foo", 'fail')
+        self.assertEqual(CONSTANT_FAIL, output.getvalue())
+
+    def test_xfail(self):
+        result, output = self._make_result()
+        result.status("foo", 'xfail')
+        self.assertEqual(CONSTANT_XFAIL, output.getvalue())
+
+    def test_unknown_status(self):
+        result, output = self._make_result()
+        self.assertRaises(Exception, result.status, "foo", 'boo')
+        self.assertEqual(b'', output.getvalue())
+
+    def test_eof(self):
+        result, output = self._make_result()
+        result.status(eof=True)
+        self.assertEqual(CONSTANT_EOF, output.getvalue())
+
+    def test_file_content(self):
+        result, output = self._make_result()
+        result.status(file_name="barney", file_bytes=b"woo")
+        self.assertEqual(CONSTANT_FILE_CONTENT, output.getvalue())
+
+    def test_mime(self):
+        result, output = self._make_result()
+        result.status(mime_type="application/foo; charset=1")
+        self.assertEqual(CONSTANT_MIME, output.getvalue())
+
+    def test_route_code(self):
+        result, output = self._make_result()
+        result.status(test_id="bar", test_status='success',
+            route_code="source")
+        self.assertEqual(CONSTANT_ROUTE_CODE, output.getvalue())
+
+    def test_runnable(self):
+        result, output = self._make_result()
+        result.status("foo", 'success', runnable=False)
+        self.assertEqual(CONSTANT_RUNNABLE, output.getvalue())
+
+    def test_tags(self):
+        result, output = self._make_result()
+        result.status(test_id="bar", test_tags=set(['foo', 'bar']))
+        self.assertEqual(CONSTANT_TAGS, output.getvalue())
+
+    def test_timestamp(self):
+        timestamp = datetime.datetime(2001, 12, 12, 12, 59, 59, 45,
+            iso8601.Utc())
+        result, output = self._make_result()
+        result.status(test_id="bar", test_status='success', timestamp=timestamp)
+        self.assertEqual(CONSTANT_TIMESTAMP, output.getvalue())
+
+
+class TestByteStreamToStreamResult(TestCase):
+
+    def test_non_subunit_encapsulated(self):
+        source = BytesIO(b"foo\nbar\n")
+        result = StreamResult()
+        subunit.ByteStreamToStreamResult(
+            source, non_subunit_name="stdout").run(result)
+        self.assertEqual([
+            ('status', None, None, None, True, 'stdout', b'f', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'o', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'o', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'\n', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'b', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'a', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'r', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'\n', False, None, None, None),
+            ], result._events)
+        self.assertEqual(b'', source.read())
+
+    def test_signature_middle_utf8_char(self):
+        utf8_bytes = b'\xe3\xb3\x8a'
+        source = BytesIO(utf8_bytes)
+        # Should be treated as one character (it is u'\u3cca') and wrapped
+        result = StreamResult()
+        subunit.ByteStreamToStreamResult(
+            source, non_subunit_name="stdout").run(
+            result)
+        self.assertEqual([
+            ('status', None, None, None, True, 'stdout', b'\xe3', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'\xb3', False, None, None, None),
+            ('status', None, None, None, True, 'stdout', b'\x8a', False, None, None, None),
+            ], result._events)
+
+    def test_non_subunit_disabled_raises(self):
+        source = BytesIO(b"foo\nbar\n")
+        result = StreamResult()
+        case = subunit.ByteStreamToStreamResult(source)
+        e = self.assertRaises(Exception, case.run, result)
+        self.assertEqual(b'f', e.args[1])
+        self.assertEqual(b'oo\nbar\n', source.read())
+        self.assertEqual([], result._events)
+
+    def test_trivial_enumeration(self):
+        source = BytesIO(CONSTANT_ENUM)
+        result = StreamResult()
+        subunit.ByteStreamToStreamResult(
+            source, non_subunit_name="stdout").run(result)
+        self.assertEqual(b'', source.read())
+        self.assertEqual([
+            ('status', 'foo', 'exists', None, True, None, None, False, None, None, None),
+            ], result._events)
+
+    def test_multiple_events(self):
+        source = BytesIO(CONSTANT_ENUM + CONSTANT_ENUM)
+        result = StreamResult()
+        subunit.ByteStreamToStreamResult(
+            source, non_subunit_name="stdout").run(result)
+        self.assertEqual(b'', source.read())
+        self.assertEqual([
+            ('status', 'foo', 'exists', None, True, None, None, False, None, None, None),
+            ('status', 'foo', 'exists', None, True, None, None, False, None, None, None),
+            ], result._events)
+
+    def test_inprogress(self):
+        self.check_event(CONSTANT_INPROGRESS, 'inprogress')
+
+    def test_success(self):
+        self.check_event(CONSTANT_SUCCESS, 'success')
+
+    def test_uxsuccess(self):
+        self.check_event(CONSTANT_UXSUCCESS, 'uxsuccess')
+
+    def test_skip(self):
+        self.check_event(CONSTANT_SKIP, 'skip')
+
+    def test_fail(self):
+        self.check_event(CONSTANT_FAIL, 'fail')
+
+    def test_xfail(self):
+        self.check_event(CONSTANT_XFAIL, 'xfail')
+
+    def check_events(self, source_bytes, events):
+        source = BytesIO(source_bytes)
+        result = StreamResult()
+        subunit.ByteStreamToStreamResult(
+            source, non_subunit_name="stdout").run(result)
+        self.assertEqual(b'', source.read())
+        self.assertEqual(events, result._events)
+
+    def check_event(self, source_bytes, test_status=None, test_id="foo",
+        route_code=None, timestamp=None, tags=None, mime_type=None,
+        file_name=None, file_bytes=None, eof=False, runnable=True):
+        event = self._event(test_id=test_id, test_status=test_status,
+            tags=tags, runnable=runnable, file_name=file_name,
+            file_bytes=file_bytes, eof=eof, mime_type=mime_type,
+            route_code=route_code, timestamp=timestamp)
+        self.check_events(source_bytes, [event])
+
+    def _event(self, test_status=None, test_id=None, route_code=None,
+        timestamp=None, tags=None, mime_type=None, file_name=None,
+        file_bytes=None, eof=False, runnable=True):
+        return ('status', test_id, test_status, tags, runnable, file_name,
+            file_bytes, eof, mime_type, route_code, timestamp)
+
+    def test_eof(self):
+        self.check_event(CONSTANT_EOF, test_id=None, eof=True)
+
+    def test_file_content(self):
+        self.check_event(CONSTANT_FILE_CONTENT,
+            test_id=None, file_name="barney", file_bytes=b"woo")
+
+    def test_file_content_length_into_checksum(self):
+        # A bad file content length which creeps into the checksum.
+        bad_file_length_content = b'\xb3!@\x13\x06barney\x04woo\xdc\xe2\xdb\x35'
+        self.check_events(bad_file_length_content, [
+            self._event(test_id="subunit.parser", eof=True,
+                file_name="Packet data", file_bytes=bad_file_length_content),
+            self._event(test_id="subunit.parser", test_status="fail", eof=True,
+                file_name="Parser Error",
+                file_bytes=b"File content extends past end of packet: claimed 4 bytes, 3 available"),
+            ])
+
+    def test_packet_length_4_word_varint(self):
+        packet_data = b'\xb3!@\xc0\x00\x11'
+        self.check_events(packet_data, [
+            self._event(test_id="subunit.parser", eof=True,
+                file_name="Packet data", file_bytes=packet_data),
+            self._event(test_id="subunit.parser", test_status="fail", eof=True,
+                file_name="Parser Error",
+                file_bytes=b"3 byte maximum given but 4 byte value found."),
+            ])
+
+    def test_mime(self):
+        self.check_event(CONSTANT_MIME,
+            test_id=None, mime_type='application/foo; charset=1')
+
+    def test_route_code(self):
+        self.check_event(CONSTANT_ROUTE_CODE,
+            'success', route_code="source", test_id="bar")
+
+    def test_runnable(self):
+        self.check_event(CONSTANT_RUNNABLE,
+            test_status='success', runnable=False)
+
+    def test_tags(self):
+        self.check_event(CONSTANT_TAGS,
+            None, tags=set(['foo', 'bar']), test_id="bar")
+
+    def test_timestamp(self):
+        timestamp = datetime.datetime(2001, 12, 12, 12, 59, 59, 45,
+            iso8601.Utc())
+        self.check_event(CONSTANT_TIMESTAMP,
+            'success', test_id='bar', timestamp=timestamp)
+
+    def test_bad_crc_errors_via_status(self):
+        file_bytes = CONSTANT_MIME[:-1] + b'\x00'
+        self.check_events( file_bytes, [
+            self._event(test_id="subunit.parser", eof=True,
+                file_name="Packet data", file_bytes=file_bytes),
+            self._event(test_id="subunit.parser", test_status="fail", eof=True,
+                file_name="Parser Error",
+                file_bytes=b'Bad checksum - calculated (0x78335115), '
+                    b'stored (0x78335100)'),
+            ])
+
+    def test_not_utf8_in_string(self):
+        file_bytes = CONSTANT_ROUTE_CODE[:5] + b'\xb4' + CONSTANT_ROUTE_CODE[6:-4] + b'\xce\x56\xc6\x17'
+        self.check_events(file_bytes, [
+            self._event(test_id="subunit.parser", eof=True,
+                file_name="Packet data", file_bytes=file_bytes),
+            self._event(test_id="subunit.parser", test_status="fail", eof=True,
+                file_name="Parser Error",
+                file_bytes=b'UTF8 string at offset 2 is not UTF8'),
+            ])
+
+    def test_NULL_in_string(self):
+        file_bytes = CONSTANT_ROUTE_CODE[:6] + b'\x00' + CONSTANT_ROUTE_CODE[7:-4] + b'\xd7\x41\xac\xfe'
+        self.check_events(file_bytes, [
+            self._event(test_id="subunit.parser", eof=True,
+                file_name="Packet data", file_bytes=file_bytes),
+            self._event(test_id="subunit.parser", test_status="fail", eof=True,
+                file_name="Parser Error",
+                file_bytes=b'UTF8 string at offset 2 contains NUL byte'),
+            ])
+
+    def test_bad_utf8_stringlength(self):
+        file_bytes = CONSTANT_ROUTE_CODE[:4] + b'\x3f' + CONSTANT_ROUTE_CODE[5:-4] + b'\xbe\x29\xe0\xc2'
+        self.check_events(file_bytes, [
+            self._event(test_id="subunit.parser", eof=True,
+                file_name="Packet data", file_bytes=file_bytes),
+            self._event(test_id="subunit.parser", test_status="fail", eof=True,
+                file_name="Parser Error",
+                file_bytes=b'UTF8 string at offset 2 extends past end of '
+                    b'packet: claimed 63 bytes, 10 available'),
+            ])
+
+
+def test_suite():
+    loader = subunit.tests.TestUtil.TestLoader()
+    result = loader.loadTestsFromName(__name__)
+    return result
 
=== added file 'python/subunit/v2.py'
--- python/subunit/v2.py	1970-01-01 00:00:00 +0000
+++ python/subunit/v2.py	2013-03-31 05:51:20 +0000
@@ -0,0 +1,458 @@
+#
+#  subunit: extensions to Python unittest to get test results from subprocesses.
+#  Copyright (C) 2013  Robert Collins <[e-mail]>
+#
+#  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
+#  license at the users choice. A copy of both licenses are available in the
+#  project source as Apache-2.0 and BSD. You may not use this file except in
+#  compliance with one of these two licences.
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+#  license you chose for the specific language governing permissions and
+#  limitations under that license.
+#
+
+import codecs
+import datetime
+from io import UnsupportedOperation
+import os
+import select
+import struct
+import zlib
+
+from extras import safe_hasattr
+
+import subunit
+import subunit.iso8601 as iso8601
+
+__all__ = [
+    'ByteStreamToStreamResult',
+    'StreamResultToBytes',
+    ]
+
+SIGNATURE = b'\xb3'
+FMT_8  = '>B'
+FMT_16 = '>H'
+FMT_24 = '>HB'
+FMT_32 = '>I'
+FMT_TIMESTAMP = '>II'
+FLAG_TEST_ID = 0x0800
+FLAG_ROUTE_CODE = 0x0400
+FLAG_TIMESTAMP = 0x0200
+FLAG_RUNNABLE = 0x0100
+FLAG_TAGS = 0x0080
+FLAG_MIME_TYPE = 0x0020
+FLAG_EOF = 0x0010
+FLAG_FILE_CONTENT = 0x0040
+EPOCH = datetime.datetime.utcfromtimestamp(0).replace(tzinfo=iso8601.Utc())
+NUL_ELEMENT = b'\0'[0]
+
+
+class ParseError(Exception):
+    """Used to pass error messages within the parser."""
+
+
+class StreamResultToBytes(object):
+    """Convert StreamResult API calls to bytes.
+    
+    The StreamResult API is defined by testtools.StreamResult.
+    """
+
+    status_mask = {
+        None: 0,
+        'exists': 0x1,
+        'inprogress': 0x2,
+        'success': 0x3,
+        'uxsuccess': 0x4,
+        'skip': 0x5,
+        'fail': 0x6,
+        'xfail': 0x7,
+        }
+
+    zero_b = b'\0'[0]
+
+    def __init__(self, output_stream):
+        """Create a StreamResultToBytes with output written to output_stream.
+
+        :param output_stream: A file-like object. Must support write(bytes)
+            and flush() methods. Flush will be called after each write.
+            The stream will be passed through subunit.make_stream_binary,
+            to handle regular cases such as stdout.
+        """
+        self.output_stream = subunit.make_stream_binary(output_stream)
+
+    def startTestRun(self):
+        pass
+
+    def stopTestRun(self):
+        pass
+
+    def status(self, test_id=None, test_status=None, test_tags=None,
+        runnable=True, file_name=None, file_bytes=None, eof=False,
+        mime_type=None, route_code=None, timestamp=None):
+        self._write_packet(test_id=test_id, test_status=test_status,
+            test_tags=test_tags, runnable=runnable, file_name=file_name,
+            file_bytes=file_bytes, eof=eof, mime_type=mime_type,
+            route_code=route_code, timestamp=timestamp)
+
+    def _write_utf8(self, a_string, packet):
+        utf8 = a_string.encode('utf-8')
+        self._write_number(len(utf8), packet)
+        packet.append(utf8)
+
+    def _write_len16(self, length, packet):
+        assert length < 65536
+        packet.append(struct.pack(FMT_16, length))
+
+    def _write_number(self, value, packet):
+        packet.extend(self._encode_number(value))
+
+    def _encode_number(self, value):
+        assert value >= 0
+        if value < 64:
+            return [struct.pack(FMT_8, value)]
+        elif value < 16384:
+            value = value | 0x4000
+            return [struct.pack(FMT_16, value)]
+        elif value < 4194304:
+            value = value | 0x800000
+            return [struct.pack(FMT_16, value >> 8),
+                    struct.pack(FMT_8, value & 0xff)]
+        elif value < 1073741824:
+            value = value | 0xc0000000
+            return [struct.pack(FMT_32, value)]
+        else:
+            raise ValueError('value too large to encode: %r' % (value,))
+
+    def _write_packet(self, test_id=None, test_status=None, test_tags=None,
+        runnable=True, file_name=None, file_bytes=None, eof=False,
+        mime_type=None, route_code=None, timestamp=None):
+        packet = [SIGNATURE]
+        packet.append(b'FF') # placeholder for flags
+        # placeholder for length, but see below as length is variable.
+        packet.append(b'')
+        flags = 0x2000 # Version 0x2
+        if timestamp is not None:
+            flags = flags | FLAG_TIMESTAMP
+            since_epoch = timestamp - EPOCH
+            nanoseconds = since_epoch.microseconds * 1000
+            seconds = (since_epoch.seconds + since_epoch.days * 24 * 3600)
+            packet.append(struct.pack(FMT_32, seconds))
+            self._write_number(nanoseconds, packet)
+        if test_id is not None:
+            flags = flags | FLAG_TEST_ID
+            self._write_utf8(test_id, packet)
+        if test_tags:
+            flags = flags | FLAG_TAGS
+            self._write_number(len(test_tags), packet)
+            for tag in test_tags:
+                self._write_utf8(tag, packet)
+        if runnable:
+            flags = flags | FLAG_RUNNABLE
+        if mime_type:
+            flags = flags | FLAG_MIME_TYPE
+            self._write_utf8(mime_type, packet)
+        if file_name is not None:
+            flags = flags | FLAG_FILE_CONTENT
+            self._write_utf8(file_name, packet)
+            self._write_number(len(file_bytes), packet)
+            packet.append(file_bytes)
+        if eof: 
+           flags = flags | FLAG_EOF
+        if route_code is not None:
+            flags = flags | FLAG_ROUTE_CODE
+            self._write_utf8(route_code, packet)
+        # 0x0008 - not used in v2.
+        flags = flags | self.status_mask[test_status]
+        packet[1] = struct.pack(FMT_16, flags)
+        base_length = sum(map(len, packet)) + 4
+        if base_length <= 62:
+            # one byte to encode length, 62+1 = 63
+            length_length = 1
+        elif base_length <= 16381:
+            # two bytes to encode length, 16381+2 = 16383
+            length_length = 2
+        elif base_length <= 4194300:
+            # three bytes to encode length, 419430+3=4194303
+            length_length = 3
+        else:
+            # Longer than policy:
+            # TODO: chunk the packet automatically?
+            # - strip all but file data
+            # - do 4M chunks of that till done
+            # - include original data in final chunk.
+            raise ValueError("Length too long: %r" % base_length)
+        packet[2:3] = self._encode_number(base_length + length_length)
+        # We could either do a partial application of crc32 over each chunk
+        # or a single join to a temp variable then a final join 
+        # or two writes (that python might then split).
+        # For now, simplest code: join, crc32, join, output
+        content = b''.join(packet)
+        self.output_stream.write(content + struct.pack(
+            FMT_32, zlib.crc32(content) & 0xffffffff))
+        self.output_stream.flush()
+
+
+class ByteStreamToStreamResult(object):
+    """Parse a subunit byte stream.
+
+    Mixed streams that contain non-subunit content is supported when a
+    non_subunit_name is passed to the contructor. The default is to raise an
+    error containing the non-subunit byte after it has been read from the
+    stream.
+
+    Typical use:
+
+       >>> case = ByteStreamToStreamResult(sys.stdin.buffer)
+       >>> result = StreamResult()
+       >>> result.startTestRun()
+       >>> case.run(result)
+       >>> result.stopTestRun()
+    """
+
+    status_lookup = {
+        0x0: None,
+        0x1: 'exists',
+        0x2: 'inprogress',
+        0x3: 'success',
+        0x4: 'uxsuccess',
+        0x5: 'skip',
+        0x6: 'fail',
+        0x7: 'xfail',
+        }
+
+    def __init__(self, source, non_subunit_name=None):
+        """Create a ByteStreamToStreamResult.
+
+        :param source: A file like object to read bytes from. Must support
+            read(<count>) and return bytes. The file is not closed by
+            ByteStreamToStreamResult. subunit.make_stream_binary() is
+            called on the stream to get it into bytes mode.
+        :param non_subunit_name: If set to non-None, non subunit content
+            encountered in the stream will be converted into file packets
+            labelled with this name.
+        """
+        self.non_subunit_name = non_subunit_name
+        self.source = subunit.make_stream_binary(source)
+        self.codec = codecs.lookup('utf8').incrementaldecoder()
+
+    def run(self, result):
+        """Parse source and emit events to result.
+        
+        This is a blocking call: it will run until EOF is detected on source.
+        """
+        self.codec.reset()
+        mid_character = False
+        while True:
+            # We're in blocking mode; read one char
+            content = self.source.read(1)
+            if not content:
+                # EOF
+                return
+            if not mid_character and content[0] == SIGNATURE[0]:
+                self._parse_packet(result)
+                continue
+            if self.non_subunit_name is None:
+                raise Exception("Non subunit content", content)
+            try:
+                if self.codec.decode(content):
+                    # End of a character
+                    mid_character = False
+                else:
+                    mid_character = True
+            except UnicodeDecodeError:
+                # Bad unicode, not our concern.
+                mid_character = False
+            # Aggregate all content that is not subunit until either
+            # 1MiB is accumulated or 50ms has passed with no input.
+            # Both are arbitrary amounts intended to give a simple
+            # balance between efficiency (avoiding death by a thousand
+            # one-byte packets), buffering (avoiding overlarge state
+            # being hidden on intermediary nodes) and interactivity
+            # (when driving a debugger, slow response to typing is
+            # annoying).
+            buffered = [content]
+            while len(buffered[-1]):
+                try:
+                    self.source.fileno()
+                except:
+                    # Won't be able to select, fallback to
+                    # one-byte-at-a-time.
+                    break
+                # Note: this has a very low timeout because with stdin, the
+                # BufferedIO layer typically has all the content available
+                # from the stream when e.g. pdb is dropped into, leading to
+                # select always timing out when in fact we could have read
+                # (from the buffer layer) - we typically fail to aggregate
+                # any content on 3.x Pythons.
+                readable = select.select([self.source], [], [], 0.000001)[0]
+                if readable:
+                    content = self.source.read(1)
+                    if not len(content):
+                        # EOF - break and emit buffered.
+                        break
+                    if not mid_character and content[0] == SIGNATURE[0]:
+                        # New packet, break, emit buffered, then parse.
+                        break
+                    buffered.append(content)
+                    # Feed into the codec.
+                    try:
+                        if self.codec.decode(content):
+                            # End of a character
+                            mid_character = False
+                        else:
+                            mid_character = True
+                    except UnicodeDecodeError:
+                        # Bad unicode, not our concern.
+                        mid_character = False
+                if not readable or len(buffered) >= 1048576:
+                    # timeout or too much data, emit what we have.
+                    break
+            result.status(
+                file_name=self.non_subunit_name,
+                file_bytes=b''.join(buffered))
+            if mid_character or not len(content) or content[0] != SIGNATURE[0]:
+                continue
+            # Otherwise, parse a data packet.
+            self._parse_packet(result)
+
+    def _parse_packet(self, result):
+        try:
+            packet = [SIGNATURE]
+            self._parse(packet, result)
+        except ParseError as error:
+            result.status(test_id="subunit.parser", eof=True,
+                file_name="Packet data", file_bytes=b''.join(packet))
+            result.status(test_id="subunit.parser", test_status='fail',
+                eof=True, file_name="Parser Error",
+                file_bytes=(error.args[0]).encode('utf8'))
+
+    def _parse_varint(self, data, pos, max_3_bytes=False):
+        # because the only incremental IO we do is at the start, and the 32 bit
+        # CRC means we can always safely read enough to cover any varint, we
+        # can be sure that there should be enough data - and if not it is an
+        # error not a normal situation.
+        data_0 = struct.unpack(FMT_8, data[pos:pos+1])[0]
+        typeenum = data_0 & 0xc0
+        value_0 = data_0 & 0x3f
+        if typeenum == 0x00:
+            return value_0, 1
+        elif typeenum == 0x40:
+            data_1 = struct.unpack(FMT_8, data[pos+1:pos+2])[0]
+            return (value_0 << 8) | data_1, 2
+        elif typeenum == 0x80:
+            data_1 = struct.unpack(FMT_16, data[pos+1:pos+3])[0]
+            return (value_0 << 16) | data_1, 3
+        else:
+            if max_3_bytes:
+                raise ParseError('3 byte maximum given but 4 byte value found.')
+            data_1, data_2 = struct.unpack(FMT_24, data[pos+1:pos+4])
+            result = (value_0 << 24) | data_1 << 8 | data_2
+            return result, 4
+
+    def _parse(self, packet, result):
+            # 2 bytes flags, at most 3 bytes length.
+            packet.append(self.source.read(5))
+            flags = struct.unpack(FMT_16, packet[-1][:2])[0]
+            length, consumed = self._parse_varint(
+                packet[-1], 2, max_3_bytes=True)
+            remainder = self.source.read(length - 6)
+            if len(remainder) != length - 6:
+                raise ParseError(
+                    'Short read - got %d bytes, wanted %d bytes' % (
+                    len(remainder), length - 6))
+            if consumed != 3:
+                # Avoid having to parse torn values
+                packet[-1] += remainder
+                pos = 2 + consumed
+            else:
+                # Avoid copying potentially lots of data.
+                packet.append(remainder)
+                pos = 0
+            crc = zlib.crc32(packet[0])
+            for fragment in packet[1:-1]:
+                crc = zlib.crc32(fragment, crc)
+            crc = zlib.crc32(packet[-1][:-4], crc) & 0xffffffff
+            packet_crc = struct.unpack(FMT_32, packet[-1][-4:])[0]
+            if crc != packet_crc:
+                # Bad CRC, report it and stop parsing the packet.
+                raise ParseError(
+                    'Bad checksum - calculated (0x%x), stored (0x%x)'
+                        % (crc, packet_crc))
+            if safe_hasattr(__builtins__, 'memoryview'):
+                body = memoryview(packet[-1])
+            else:
+                body = packet[-1]
+            # Discard CRC-32
+            body = body[:-4]
+            # One packet could have both file and status data; the Python API
+            # presents these separately (perhaps it shouldn't?)
+            if flags & FLAG_TIMESTAMP:
+                seconds = struct.unpack(FMT_32, body[pos:pos+4])[0]
+                nanoseconds, consumed = self._parse_varint(body, pos+4)
+                pos = pos + 4 + consumed
+                timestamp = EPOCH + datetime.timedelta(
+                    seconds=seconds, microseconds=nanoseconds/1000)
+            else:
+                timestamp = None
+            if flags & FLAG_TEST_ID:
+                test_id, pos = self._read_utf8(body, pos)
+            else:
+                test_id = None
+            if flags & FLAG_TAGS:
+                tag_count, consumed = self._parse_varint(body, pos)
+                pos += consumed
+                test_tags = set()
+                for _ in range(tag_count):
+                    tag, pos = self._read_utf8(body, pos)
+                    test_tags.add(tag)
+            else:
+                test_tags = None
+            if flags & FLAG_MIME_TYPE:
+                mime_type, pos = self._read_utf8(body, pos)
+            else:
+                mime_type = None
+            if flags & FLAG_FILE_CONTENT:
+                file_name, pos = self._read_utf8(body, pos)
+                content_length, consumed = self._parse_varint(body, pos)
+                pos += consumed
+                file_bytes = body[pos:pos+content_length]
+                if len(file_bytes) != content_length:
+                    raise ParseError('File content extends past end of packet: '
+                        'claimed %d bytes, %d available' % (
+                        content_length, len(file_bytes)))
+            else:
+                file_name = None
+                file_bytes = None
+            if flags & FLAG_ROUTE_CODE:
+                route_code, pos = self._read_utf8(body, pos)
+            else:
+                route_code = None
+            runnable = bool(flags & FLAG_RUNNABLE)
+            eof = bool(flags & FLAG_EOF)
+            test_status = self.status_lookup[flags & 0x0007]
+            result.status(test_id=test_id, test_status=test_status,
+                test_tags=test_tags, runnable=runnable, mime_type=mime_type,
+                eof=eof, file_name=file_name, file_bytes=file_bytes,
+                route_code=route_code, timestamp=timestamp)
+    __call__ = run
+
+    def _read_utf8(self, buf, pos):
+        length, consumed = self._parse_varint(buf, pos)
+        pos += consumed
+        utf8_bytes = buf[pos:pos+length]
+        if length != len(utf8_bytes):
+            raise ParseError(
+                'UTF8 string at offset %d extends past end of packet: '
+                'claimed %d bytes, %d available' % (pos - 2, length,
+                len(utf8_bytes)))
+        if NUL_ELEMENT in utf8_bytes:
+            raise ParseError('UTF8 string at offset %d contains NUL byte' % (
+                pos-2,))
+        try:
+            return utf8_bytes.decode('utf-8'), length+pos
+        except UnicodeDecodeError:
+            raise ParseError('UTF8 string at offset %d is not UTF8' % (pos-2,))
+
 
=== modified file 'setup.py'
--- setup.py	2012-12-17 08:12:44 +0000
+++ setup.py	2013-03-31 05:51:20 +0000
@@ -9,6 +9,7 @@
 else:
     extra = {
         'install_requires': [
+            'extras',
             'testtools>=0.9.23',
         ]
     }
@@ -49,6 +50,8 @@
     packages=['subunit', 'subunit.tests'],
     package_dir={'subunit': 'python/subunit'},
     scripts = [
+        'filters/subunit-1to2',
+        'filters/subunit-2to1',
         'filters/subunit2gtk',
         'filters/subunit2junitxml',
         'filters/subunit2pyunit',