# needed and report to your result object.
suite.run(result)
-subunit includes extensions to the python ``TestResult`` protocol. The
-``time(a_datetime)`` method is called (if present) when a ``time:``
+subunit includes extensions to the python ``TestResult`` protocol.
+
+The ``time(a_datetime)`` method is called (if present) when a ``time:``
directive is encountered in a subunit stream. This is used to tell a TestResult
about the time that events in the stream occured at, to allow reconstructing
test timing from a stream.
+The ``progress(offset, whence)`` method controls progress data for a stream.
+The offset parameter is an int, and whence is one of subunit.SEEK_CUR,
+subunit.SEEK_SET.
+
Finally, subunit.run is a convenience wrapper to run a python test suite via
the command line, reporting via subunit::
import iso8601
+SEEK_CUR = os.SEEK_CUR
+SEEK_SET = os.SEEK_SET
+
+
def test_suite():
import subunit.tests
return subunit.tests.test_suite()
else:
self.stdOutLineReceived(line)
+ def _handleProgress(self, offset, line):
+ """Process a progress directive."""
+ line = line[offset:].strip()
+ if line[0] in '+-':
+ whence = SEEK_CUR
+ else:
+ whence = SEEK_SET
+ delta = int(line)
+ progress_method = getattr(self.client, 'progress', None)
+ if callable(progress_method):
+ progress_method(delta, whence)
+
def _handleTags(self, offset, line):
"""Process a tags command."""
tags = line[offset:].split()
self._addError(offset, line)
elif cmd == 'failure':
self._addFailure(offset, line)
+ elif cmd == 'progress':
+ self._handleProgress(offset, line)
elif cmd == 'skip':
self._addSkip(offset, line)
elif cmd in ('success', 'successful'):
"""Mark a test as starting its test run."""
self._stream.write("test: %s\n" % test.id())
+ def progress(self, offset, whence):
+ """Provide indication about the progress/length of the test run.
+
+ :param offset: Information about the number of tests remaining. If
+ whence is SEEK_CUR, then offset increases/decreases the remaining
+ test count. If whence is SEEK_SET, then offset specifies exactly
+ the remaining test count.
+ :param whence: One of SEEK_CUR or SEEK_SET.
+ """
+ if whence == SEEK_CUR and offset > -1:
+ prefix = "+"
+ else:
+ prefix = ""
+ self._stream.write("progress: %s%s\n" % (prefix, offset))
+
def time(self, a_datetime):
"""Inform the client of the time.
self._before_event()
return self._call_maybe("addUnexpectedSuccess", test)
+ def progress(self, offset, whence):
+ self._before_event()
+ return self._call_maybe("progress", offset, whence)
+
def wasSuccessful(self):
self._before_event()
return self.decorated.wasSuccessful()
time = datetime.datetime.utcnow().replace(tzinfo=iso8601.Utc())
self._call_maybe("time", time)
+ def progress(self, offset, whence):
+ return self._call_maybe("progress", offset, whence)
+
@property
def shouldStop(self):
return self.decorated.shouldStop
self.skip_calls = []
self.start_calls = []
self.success_calls = []
+ self.progress_calls = []
self._time = None
super(MockTestProtocolServerClient, self).__init__()
def startTest(self, test):
self.start_calls.append(test)
+ def progress(self, offset, whence):
+ self.progress_calls.append((offset, whence))
+
def time(self, time):
self._time = time
self.assertEqual(protocol.success_calls, [])
self.assertEqual(protocol.start_calls, [])
+ def test_progress(self):
+ protocol = MockTestProtocolServerClient()
+ protocol.progress(-1, subunit.SEEK_CUR)
+ self.assertEqual(protocol.progress_calls, [(-1, subunit.SEEK_CUR)])
+
class TestTestImports(unittest.TestCase):
self.success_quoted_bracket("success:")
+class TestTestProtocolServerProgress(unittest.TestCase):
+ """Test receipt of progress: directives."""
+
+ def test_progress_accepted_stdlib(self):
+ # With a stdlib TestResult, progress events are swallowed.
+ self.result = unittest.TestResult()
+ self.stream = StringIO()
+ self.protocol = subunit.TestProtocolServer(self.result,
+ stream=self.stream)
+ self.protocol.lineReceived("progress: 23")
+ self.protocol.lineReceived("progress: -2")
+ self.protocol.lineReceived("progress: +4")
+ self.assertEqual("", self.stream.getvalue())
+
+ def test_progress_accepted_extended(self):
+ # With a progress capable TestResult, progress events are emitted.
+ self.result = MockTestProtocolServerClient()
+ self.stream = StringIO()
+ self.protocol = subunit.TestProtocolServer(self.result,
+ stream=self.stream)
+ self.protocol.lineReceived("progress: 23")
+ self.protocol.lineReceived("progress: -2")
+ self.protocol.lineReceived("progress: +4")
+ self.assertEqual("", self.stream.getvalue())
+ self.assertEqual(
+ [(23, subunit.SEEK_SET), (-2, subunit.SEEK_CUR),
+ (4, subunit.SEEK_CUR)],
+ self.result.progress_calls)
+
+
class TestTestProtocolServerStreamTags(unittest.TestCase):
"""Test managing tags on the protocol level."""
self.io.getvalue(),
'skip: %s [\nHas it really?\n]\n' % self.test.id())
+ def test_progress_set(self):
+ self.protocol.progress(23, subunit.SEEK_SET)
+ self.assertEqual(self.io.getvalue(), 'progress: 23\n')
+
+ def test_progress_neg_cur(self):
+ self.protocol.progress(-23, subunit.SEEK_CUR)
+ self.assertEqual(self.io.getvalue(), 'progress: -23\n')
+
+ def test_progress_pos_cur(self):
+ self.protocol.progress(23, subunit.SEEK_CUR)
+ self.assertEqual(self.io.getvalue(), 'progress: +23\n')
+
def test_time(self):
# Calling time() outputs a time signal immediately.
self.protocol.time(
def test_addUnexpectedSuccess(self):
self.result.addUnexpectedSuccess(self)
+ def test_progress(self):
+ self.result.progress(1, os.SEEK_SET)
+
def test_wasSuccessful(self):
self.result.wasSuccessful()
self.assertEqual(1, len(self.result.decorated._calls))
self.assertNotEqual(None, self.result.decorated._calls[0])
+ def test_no_time_from_progress(self):
+ self.result.progress(1, os.SEEK_CUR)
+ self.assertEqual(0, len(self.result.decorated._calls))
+
def test_no_time_from_shouldStop(self):
self.result.decorated.stop()
self.result.shouldStop