* Expects subunit v2 if the local library has v2 support in the subunit

  library. This should be seamless if the system under test shares the
  Python libraries. If it doesn't, either arrange to use ``subunit-2to1``
  or upgrade the subunit libraries for the system under test.
  (Robert Collins)

11 files changed, 261 insertions, 42 deletions

diff --git a/NEWS b/NEWS
index 01d8747..51cb818 100644
--- a/NEWS
+++ b/NEWS
@@ -5,6 +5,15 @@ testrepository release notes
 NEXT (In development)
 +++++++++++++++++++++
 
+CHANGES
+-------
+
+* Expects subunit v2 if the local library has v2 support in the subunit
+  library. This should be seamless if the system under test shares the
+  Python libraries. If it doesn't, either arrange to use ``subunit-2to1``
+  or upgrade the subunit libraries for the system under test.
+  (Robert Collins)
+
 0.0.14
 ++++++
 
diff --git a/testrepository/commands/load.py b/testrepository/commands/load.py
index 074ec1e..fecbc1a 100644
--- a/testrepository/commands/load.py
+++ b/testrepository/commands/load.py
@@ -15,9 +15,14 @@
 """Load data into a repository."""
 
 from functools import partial
+from operator import methodcaller
 import optparse
 
+from extras import try_import
+v2_avail = try_import('subunit.ByteStreamToStreamResult')
+
 import subunit.test_results
+import testtools
 from testtools import ConcurrentTestSuite, MultiTestResult, Tagger
 
 from testrepository.arguments.path import ExistingPathArgument
@@ -83,11 +88,24 @@ class load(Command):
             cases = lambda:map(opener, self.ui.arguments['streams'])
         else:
             cases = lambda:self.ui.iter_streams('subunit')
-        def make_tests(suite):
-            streams = list(suite)[0]
-            for stream in streams():
-                yield subunit.ProtocolTestCase(stream)
-        case = ConcurrentTestSuite(cases, make_tests, _wrap_result)
+        if v2_avail:
+            def make_tests(suite):
+                streams = list(suite)[0]
+                for pos, stream in enumerate(streams()):
+                    case = subunit.ByteStreamToStreamResult(
+                        stream, non_subunit_name='stdout')
+                    case = testtools.DecorateTestCaseResult(case,
+                        lambda result:testtools.StreamTagger(
+                            [result], add=['worker-%d' % pos]))
+                    yield (case, str(pos))
+            case = testtools.ConcurrentStreamTestSuite(cases, make_tests)
+        else:
+            # TODO: switch this to the new threadsafesuite thingy. Do the decorate on the backend.
+            def make_tests(suite):
+                streams = list(suite)[0]
+                for stream in streams():
+                    yield subunit.ProtocolTestCase(stream)
+            case = ConcurrentTestSuite(cases, make_tests, _wrap_result)
         # One copy of the stream to repository storage
         inserter = repo.get_inserter(partial=self.ui.options.partial)
         # One copy of the stream to the UI layer after performing global
@@ -99,6 +117,8 @@ class load(Command):
         output_result = self.ui.make_result(
             lambda: run_id, testcommand, previous_run=previous_run)
         result = MultiTestResult(inserter, output_result)
+        if v2_avail:
+            result = testtools.StreamToExtendedDecorator(result)
         result.startTestRun()
         try:
             case.run(result)
diff --git a/testrepository/commands/run.py b/testrepository/commands/run.py
index b28b7b6..8a68b2d 100644
--- a/testrepository/commands/run.py
+++ b/testrepository/commands/run.py
@@ -19,6 +19,9 @@ from math import ceil
 import optparse
 import re
 
+from extras import try_import
+import subunit
+v2_avail = try_import('subunit.ByteStreamToStreamResult')
 from testtools import (
     TestResult,
     TestByTestResult,
@@ -67,10 +70,16 @@ class ReturnCodeToSubunit(object):
             if self.lastoutput != LINEFEED:
                 # Subunit is line orientated, it has to start on a fresh line.
                 self.source.write(_b('\n'))
-            self.source.write(_b('test: process-returncode\n'
-                'error: process-returncode [\n'
-                ' returncode %d\n'
-                ']\n' % returncode))
+            if v2_avail:
+                stream = subunit.StreamResultToBytes(self.source)
+                stream.status(test_id='process-returncode', test_status='fail',
+                    file_name='traceback', mime_type='test/plain;charset=utf8',
+                    file_bytes=('returncode %d' % returncode).encode('utf8'))
+            else:
+                self.source.write(_b('test: process-returncode\n'
+                    'failure: process-returncode [\n'
+                    ' returncode %d\n'
+                    ']\n' % returncode))
         self.source.seek(0)
         self.done = True
 
diff --git a/testrepository/testcommand.py b/testrepository/testcommand.py
index 9fc994f..e4364cd 100644
--- a/testrepository/testcommand.py
+++ b/testrepository/testcommand.py
@@ -31,7 +31,7 @@ from testtools.compat import _b
 
 from testrepository.results import TestResultFilter
 from testrepository.testlist import (
-    parse_list,
+    parse_enumeration,
     write_list,
     )
 
@@ -288,7 +288,7 @@ class TestListingFixture(Fixture):
                 stdout=subprocess.PIPE, stdin=subprocess.PIPE)
             out, err = run_proc.communicate()
             # Should we raise on non-zero exit?
-            ids = parse_list(out)
+            ids = parse_enumeration(out)
             return ids
         finally:
             if instance:
diff --git a/testrepository/testlist.py b/testrepository/testlist.py
index 96174c2..e1b4aca 100644
--- a/testrepository/testlist.py
+++ b/testrepository/testlist.py
@@ -14,8 +14,15 @@
 
 """Handling of lists of tests - common code to --load-list etc."""
 
+from io import BytesIO
+
+from extras import try_import
+bytestream_to_streamresult = try_import('subunit.ByteStreamToStreamResult')
+stream_result = try_import('testtools.testresult.doubles.StreamResult')
+
 from testtools.compat import _b, _u
 
+
 def write_list(stream, test_ids):
     """Write test_ids out to stream.
 
@@ -28,5 +35,26 @@ def write_list(stream, test_ids):
 
 def parse_list(list_bytes):
     """Parse list_bytes into a list of test ids."""
+    return _v1(list_bytes)
+
+
+def parse_enumeration(enumeration_bytes):
+    """Parse enumeration_bytes into a list of test_ids."""
+    # If subunit v2 is available, use it.
+    if bytestream_to_streamresult is not None:
+        return _v2(enumeration_bytes)
+    else:
+        return _v1(enumeration_bytes)
+
+
+def _v1(list_bytes):
     return [id.strip() for id in list_bytes.decode('utf8').split(_u('\n'))
         if id.strip()]
+
+
+def _v2(list_bytes):
+    parser = bytestream_to_streamresult(BytesIO(list_bytes),
+        non_subunit_name='stdout')
+    result = stream_result()
+    parser.run(result)
+    return [event[1] for event in result._events if event[2]=='exists']
diff --git a/testrepository/tests/commands/test_list_tests.py b/testrepository/tests/commands/test_list_tests.py
index 9120955..6ee2010 100644
--- a/testrepository/tests/commands/test_list_tests.py
+++ b/testrepository/tests/commands/test_list_tests.py
@@ -14,9 +14,13 @@
 
 """Tests for the list_tests command."""
 
+from io import BytesIO
 import os.path
 from subprocess import PIPE
 
+from extras import try_import
+import subunit
+v2_avail = try_import('subunit.ByteStreamToStreamResult')
 from testtools.compat import _b
 from testtools.matchers import MatchesException
 
@@ -71,7 +75,15 @@ class TestCommand(ResourcedTestCase):
     def test_calls_list_tests(self):
         ui, cmd = self.get_test_ui_and_cmd(args=('--', 'bar', 'quux'))
         cmd.repository_factory = memory.RepositoryFactory()
-        ui.proc_outputs = [_b('returned\n\nvalues\n')]
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='returned', test_status='exists')
+            stream.status(test_id='values', test_status='exists')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = _b('returned\n\nvalues\n')
+        ui.proc_outputs = [subunit_bytes]
         self.setup_repo(cmd, ui)
         self.set_config(
             '[DEFAULT]\ntest_command=foo $LISTOPT $IDOPTION\n'
@@ -91,7 +103,15 @@ class TestCommand(ResourcedTestCase):
         ui, cmd = self.get_test_ui_and_cmd(
             args=('returned', '--', 'bar', 'quux'))
         cmd.repository_factory = memory.RepositoryFactory()
-        ui.proc_outputs = [_b('returned\n\nvalues\n')]
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='returned', test_status='exists')
+            stream.status(test_id='values', test_status='exists')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = _b('returned\nvalues\n')
+        ui.proc_outputs = [subunit_bytes]
         self.setup_repo(cmd, ui)
         self.set_config(
             '[DEFAULT]\ntest_command=foo $LISTOPT $IDOPTION\n'
diff --git a/testrepository/tests/commands/test_load.py b/testrepository/tests/commands/test_load.py
index ad94cb5..36cf82b 100644
--- a/testrepository/tests/commands/test_load.py
+++ b/testrepository/tests/commands/test_load.py
@@ -14,9 +14,13 @@
 
 """Tests for the load command."""
 
-from datetime import datetime
+from datetime import datetime, timedelta
+from io import BytesIO
 from tempfile import NamedTemporaryFile
 
+from extras import try_import
+v2_avail = try_import('subunit.ByteStreamToStreamResult')
+import subunit
 from subunit import iso8601
 
 import testtools
@@ -111,7 +115,15 @@ class TestCommandLoad(ResourcedTestCase):
         self.assertEqual(0, cmd.execute())
 
     def test_load_returns_1_on_failed_stream(self):
-        ui = UI([('subunit', _b('test: foo\nfailure: foo\n'))])
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='foo', test_status='inprogress')
+            stream.status(test_id='foo', test_status='fail')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = _b('test: foo\nfailure: foo\n')
+        ui = UI([('subunit', subunit_bytes)])
         cmd = load.load(ui)
         ui.set_command(cmd)
         cmd.repository_factory = memory.RepositoryFactory()
@@ -119,7 +131,15 @@ class TestCommandLoad(ResourcedTestCase):
         self.assertEqual(1, cmd.execute())
 
     def test_load_new_shows_test_failures(self):
-        ui = UI([('subunit', _b('test: foo\nfailure: foo\n'))])
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='foo', test_status='inprogress')
+            stream.status(test_id='foo', test_status='fail')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = b'test: foo\nfailure: foo\n'
+        ui = UI([('subunit', subunit_bytes)])
         cmd = load.load(ui)
         ui.set_command(cmd)
         cmd.repository_factory = memory.RepositoryFactory()
@@ -131,7 +151,17 @@ class TestCommandLoad(ResourcedTestCase):
             ui.outputs[1:])
 
     def test_load_new_shows_test_failure_details(self):
-        ui = UI([('subunit', _b('test: foo\nfailure: foo [\narg\n]\n'))])
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='foo', test_status='inprogress')
+            stream.status(test_id='foo', test_status='fail',
+                file_name="traceback", mime_type='text/plain;charset=utf8',
+                file_bytes=b'arg\n')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = b'test: foo\nfailure: foo [\narg\n]\n'
+        ui = UI([('subunit', subunit_bytes)])
         cmd = load.load(ui)
         ui.set_command(cmd)
         cmd.repository_factory = memory.RepositoryFactory()
@@ -153,7 +183,15 @@ class TestCommandLoad(ResourcedTestCase):
         self.assertEqual(1, len(result.failures))
 
     def test_load_new_shows_test_skips(self):
-        ui = UI([('subunit', _b('test: foo\nskip: foo\n'))])
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='foo', test_status='inprogress')
+            stream.status(test_id='foo', test_status='skip')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = b'test: foo\nskip: foo\n'
+        ui = UI([('subunit', subunit_bytes)])
         cmd = load.load(ui)
         ui.set_command(cmd)
         cmd.repository_factory = memory.RepositoryFactory()
@@ -199,13 +237,22 @@ class TestCommandLoad(ResourcedTestCase):
             cmd.repository_factory.repos[ui.here].get_test_run(0)._partial)
 
     def test_load_timed_run(self):
-        ui = UI(
-            [('subunit',
-              _b('time: 2011-01-01 00:00:01.000000Z\n'
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            time = datetime(2011, 1, 1, 0, 0, 1, tzinfo=iso8601.Utc())
+            stream.status(test_id='foo', test_status='inprogress', timestamp=time)
+            stream.status(test_id='foo', test_status='success',
+                timestamp=time+timedelta(seconds=2))
+            timed_bytes = buffer.getvalue()
+        else:
+            timed_bytes = _b('time: 2011-01-01 00:00:01.000000Z\n'
                'test: foo\n'
                'time: 2011-01-01 00:00:03.000000Z\n'
                'success: foo\n'
-               'time: 2011-01-01 00:00:06.000000Z\n'))])
+               'time: 2011-01-01 00:00:06.000000Z\n')
+        ui = UI(
+            [('subunit', timed_bytes)])
         cmd = load.load(ui)
         ui.set_command(cmd)
         cmd.repository_factory = memory.RepositoryFactory()
@@ -237,17 +284,29 @@ class TestCommandLoad(ResourcedTestCase):
         # If there's a previous run in the database, then show information
         # about the high level differences in the test run: how many more
         # tests, how many more failures, how much longer it takes.
-        ui = UI(
-            [('subunit',
-              _b('time: 2011-01-02 00:00:01.000000Z\n'
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            time = datetime(2011, 1, 2, 0, 0, 1, tzinfo=iso8601.Utc())
+            stream.status(test_id='foo', test_status='inprogress', timestamp=time)
+            stream.status(test_id='foo', test_status='fail',
+                timestamp=time+timedelta(seconds=2))
+            stream.status(test_id='bar', test_status='inprogress',
+                timestamp=time+timedelta(seconds=4))
+            stream.status(test_id='bar', test_status='fail',
+                timestamp=time+timedelta(seconds=6))
+            timed_bytes = buffer.getvalue()
+        else:
+            timed_bytes = _b('time: 2011-01-02 00:00:01.000000Z\n'
                'test: foo\n'
                'time: 2011-01-02 00:00:03.000000Z\n'
                'error: foo\n'
                'time: 2011-01-02 00:00:05.000000Z\n'
                'test: bar\n'
                'time: 2011-01-02 00:00:07.000000Z\n'
-               'error: bar\n'
-               ))])
+               'error: bar\n')
+        ui = UI(
+            [('subunit', timed_bytes)])
         cmd = load.load(ui)
         ui.set_command(cmd)
         cmd.repository_factory = memory.RepositoryFactory()
diff --git a/testrepository/tests/commands/test_run.py b/testrepository/tests/commands/test_run.py
index 5eb9d5b..26eba32 100644
--- a/testrepository/tests/commands/test_run.py
+++ b/testrepository/tests/commands/test_run.py
@@ -14,14 +14,18 @@
 
 """Tests for the run command."""
 
+from io import BytesIO
 import os.path
 from subprocess import PIPE
 import tempfile
 
+from extras import try_import
 from fixtures import (
     Fixture,
     MonkeyPatch,
     )
+import subunit
+v2_avail = try_import('subunit.ByteStreamToStreamResult')
 from subunit import RemotedTestCase
 from testscenarios.scenarios import multiply_scenarios
 from testtools.compat import _b
@@ -294,8 +298,16 @@ class TestCommand(ResourcedTestCase):
             cmd.repository_factory.repos[ui.here].get_test_run(1)._partial)
 
     def test_load_failure_exposed(self):
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='foo', test_status='inprogress')
+            stream.status(test_id='foo', test_status='fail')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = b'test: foo\nfailure: foo\n'
         ui, cmd = self.get_test_ui_and_cmd(options=[('quiet', True),],
-            proc_outputs=[_b('test: foo\nfailure: foo\n')])
+            proc_outputs=[subunit_bytes])
         cmd.repository_factory = memory.RepositoryFactory()
         self.setup_repo(cmd, ui)
         self.set_config('[DEFAULT]\ntest_command=foo\n')
@@ -304,8 +316,16 @@ class TestCommand(ResourcedTestCase):
         self.assertEqual(1, result)
 
     def test_process_exit_code_nonzero_causes_synthetic_error_test(self):
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='foo', test_status='inprogress')
+            stream.status(test_id='foo', test_status='success')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = b'test: foo\nsuccess: foo\n'
         ui, cmd = self.get_test_ui_and_cmd(options=[('quiet', True),],
-            proc_outputs=[_b('test: foo\nsuccess: foo\n')],
+            proc_outputs=[subunit_bytes],
             proc_results=[2])
             # 2 is non-zero, and non-zero triggers the behaviour of exiting
             # with 1 - but we want to see that it doesn't pass-through the
@@ -316,10 +336,8 @@ class TestCommand(ResourcedTestCase):
         result = cmd.execute()
         self.assertEqual(1, result)
         run = cmd.repository_factory.repos[ui.here].get_test_run(1)
-        self.assertEqual([
-            Wildcard,
-            ('Error', RemotedTestCase('process-returncode'), Wildcard)],
-            run._outcomes)
+        self.assertEqual([Wildcard, 'Failure'],
+            [outcome[0] for outcome in run._outcomes])
 
     def test_regex_test_filter(self):
         ui, cmd = self.get_test_ui_and_cmd(args=('ab.*cd', '--', 'bar', 'quux'))
@@ -372,10 +390,25 @@ class TestCommand(ResourcedTestCase):
 
     def test_until_failure(self):
         ui, cmd = self.get_test_ui_and_cmd(options=[('until_failure', True)])
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='foo', test_status='inprogress')
+            stream.status(test_id='foo', test_status='success')
+            subunit_bytes1 = buffer.getvalue()
+            buffer.seek(0)
+            buffer.truncate()
+            stream.status(test_id='foo', test_status='inprogress')
+            stream.status(test_id='foo', test_status='fail')
+            subunit_bytes2 = buffer.getvalue()
+        else:
+            subunit_bytes1 = b'test: foo\nsuccess: foo\n'
+            subunit_bytes2 = b'test: foo\nfailure: foo\n'
         ui.proc_outputs = [
-            _b('test: foo\nsuccess: foo\n'), # stream one, works
-            _b('test: foo\nfailure: foo\n') # stream two, fails
+            subunit_bytes1, # stream one, works
+            subunit_bytes2, # stream two, fails
             ]
+        ui.require_proc_stdout = True
         cmd.repository_factory = memory.RepositoryFactory()
         self.setup_repo(cmd, ui)
         self.set_config(
@@ -465,7 +498,15 @@ class TestReturnCodeToSubunit(ResourcedTestCase):
         proc.returncode = 1
         stream = run.ReturnCodeToSubunit(proc)
         content = accumulate(stream, self.reader)
-        self.assertEqual(
-            _b('foo\nbar\ntest: process-returncode\n'
-            'error: process-returncode [\n returncode 1\n]\n'),
-            content)
+        if v2_avail:
+            buffer = BytesIO()
+            buffer.write(b'foo\nbar\n')
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='process-returncode', test_status='fail',
+                file_name='traceback', mime_type='test/plain;charset=utf8',
+                file_bytes=b'returncode 1')
+            expected_content = buffer.getvalue()
+        else:
+            expected_content = _b('foo\nbar\ntest: process-returncode\n'
+                'failure: process-returncode [\n returncode 1\n]\n')
+        self.assertEqual(expected_content, content)
diff --git a/testrepository/tests/test_testcommand.py b/testrepository/tests/test_testcommand.py
index c664305..21d5c7a 100644
--- a/testrepository/tests/test_testcommand.py
+++ b/testrepository/tests/test_testcommand.py
@@ -14,10 +14,14 @@
 
 """Tests for the testcommand module."""
 
+from io import BytesIO
 import os.path
 import optparse
 import re
 
+from extras import try_import
+import subunit
+v2_avail = try_import('subunit.ByteStreamToStreamResult')
 from testtools.compat import _b
 from testtools.matchers import (
     Equals,
@@ -181,7 +185,15 @@ class TestTestCommand(ResourcedTestCase):
 
     def test_get_run_command_default_and_list_expands(self):
         ui, command = self.get_test_ui_and_cmd()
-        ui.proc_outputs = [_b('returned\nids\n')]
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='returned', test_status='exists')
+            stream.status(test_id='ids', test_status='exists')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = _b('returned\nids\n')
+        ui.proc_outputs = [subunit_bytes]
         ui.options = optparse.Values()
         ui.options.parallel = True
         ui.options.concurrency = 2
@@ -283,8 +295,16 @@ class TestTestCommand(ResourcedTestCase):
         self.assertEqual(expected_cmd, fixture.list_cmd)
 
     def test_list_tests_parsing(self):
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='returned', test_status='exists')
+            stream.status(test_id='ids', test_status='exists')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = _b('returned\nids\n')
         ui, command = self.get_test_ui_and_cmd()
-        ui.proc_outputs = [_b('returned\nids\n')]
+        ui.proc_outputs = [subunit_bytes]
         self.set_config(
             '[DEFAULT]\ntest_command=foo $LISTOPT $IDLIST\ntest_id_list_default=whoo yea\n'
             'test_list_option=--list\n')
@@ -515,8 +535,16 @@ class TestTestCommand(ResourcedTestCase):
             log._events)
 
     def test_filter_tests_by_regex_only(self):
+        if v2_avail:
+            buffer = BytesIO()
+            stream = subunit.StreamResultToBytes(buffer)
+            stream.status(test_id='returned', test_status='exists')
+            stream.status(test_id='ids', test_status='exists')
+            subunit_bytes = buffer.getvalue()
+        else:
+            subunit_bytes = _b('returned\nids\n')
         ui, command = self.get_test_ui_and_cmd()
-        ui.proc_outputs = [_b('returned\nids\n')]
+        ui.proc_outputs = [subunit_bytes]
         self.set_config(
             '[DEFAULT]\ntest_command=foo $LISTOPT $IDLIST\ntest_id_list_default=whoo yea\n'
             'test_list_option=--list\n')
diff --git a/testrepository/tests/test_ui.py b/testrepository/tests/test_ui.py
index 9c62466..520a2da 100644
--- a/testrepository/tests/test_ui.py
+++ b/testrepository/tests/test_ui.py
@@ -19,6 +19,7 @@ import optparse
 import subprocess
 import sys
 
+from fixtures import EnvironmentVariable
 from testtools.compat import _b, _u
 from testtools.content import text_content
 from testtools.matchers import raises
@@ -100,6 +101,7 @@ class TestUIContract(ResourcedTestCase):
         self.assertThat(lambda: ui.iter_streams('subunit'), raises(KeyError))
 
     def test_output_error(self):
+        self.useFixture(EnvironmentVariable('TESTR_PDB'))
         try:
             raise Exception('fooo')
         except Exception:
diff --git a/testrepository/ui/model.py b/testrepository/ui/model.py
index 0c0f580..8063520 100644
--- a/testrepository/ui/model.py
+++ b/testrepository/ui/model.py
@@ -113,6 +113,7 @@ class UI(ui.AbstractUI):
         # Could take parsed args, but for now this is easier.
         self.unparsed_args = args
         self.proc_outputs = list(proc_outputs)
+        self.require_proc_stdout = False
         self.proc_results = list(proc_results)
 
     def _check_cmd(self):
@@ -180,6 +181,8 @@ class UI(ui.AbstractUI):
         result = ProcessModel(self)
         if self.proc_outputs:
             result.stdout = BytesIO(self.proc_outputs.pop(0))
+        elif self.require_proc_stdout:
+            raise Exception("No process output available")
         if self.proc_results:
             result.returncode = self.proc_results.pop(0)
         return result