summaryrefslogtreecommitdiff
path: root/lib/subunit
diff options
context:
space:
mode:
Diffstat (limited to 'lib/subunit')
-rwxr-xr-xlib/subunit/filters/tap2subunit26
-rw-r--r--lib/subunit/python/iso8601/LICENSE20
-rw-r--r--lib/subunit/python/iso8601/README26
-rw-r--r--lib/subunit/python/iso8601/README.subunit5
-rw-r--r--lib/subunit/python/iso8601/setup.py58
-rw-r--r--lib/subunit/python/iso8601/test_iso8601.py111
-rw-r--r--lib/subunit/python/subunit/__init__.py1107
-rw-r--r--lib/subunit/python/subunit/chunked.py164
-rw-r--r--lib/subunit/python/subunit/details.py109
-rw-r--r--lib/subunit/python/subunit/iso8601.py123
-rw-r--r--lib/subunit/python/subunit/progress_model.py106
-rwxr-xr-xlib/subunit/python/subunit/run.py46
-rw-r--r--lib/subunit/python/subunit/test_results.py334
-rw-r--r--lib/subunit/python/subunit/tests/__init__.py44
-rwxr-xr-xlib/subunit/python/subunit/tests/sample-script.py7
-rw-r--r--lib/subunit/python/subunit/tests/test_chunked.py127
-rw-r--r--lib/subunit/python/subunit/tests/test_details.py110
-rw-r--r--lib/subunit/python/subunit/tests/test_progress_model.py118
-rw-r--r--lib/subunit/python/subunit/tests/test_subunit_filter.py136
-rw-r--r--lib/subunit/python/subunit/tests/test_subunit_stats.py83
-rw-r--r--lib/subunit/python/subunit/tests/test_subunit_tags.py68
-rw-r--r--lib/subunit/python/subunit/tests/test_tap2subunit.py432
-rw-r--r--lib/subunit/python/subunit/tests/test_test_protocol.py967
-rw-r--r--lib/subunit/python/subunit/tests/test_test_results.py199
-rwxr-xr-xlib/subunit/tap2subunit35
-rwxr-xr-xlib/subunit/update.sh4
26 files changed, 4061 insertions, 504 deletions
diff --git a/lib/subunit/filters/tap2subunit b/lib/subunit/filters/tap2subunit
new file mode 100755
index 0000000000..c571972225
--- /dev/null
+++ b/lib/subunit/filters/tap2subunit
@@ -0,0 +1,26 @@
+#!/usr/bin/env python
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""A filter that reads a TAP stream and outputs a subunit stream.
+
+More information on TAP is available at
+http://testanything.org/wiki/index.php/Main_Page.
+"""
+
+import sys
+
+from subunit import TAP2SubUnit
+sys.exit(TAP2SubUnit(sys.stdin, sys.stdout))
diff --git a/lib/subunit/python/iso8601/LICENSE b/lib/subunit/python/iso8601/LICENSE
new file mode 100644
index 0000000000..5ca93dae79
--- /dev/null
+++ b/lib/subunit/python/iso8601/LICENSE
@@ -0,0 +1,20 @@
+Copyright (c) 2007 Michael Twomey
+
+Permission is hereby granted, free of charge, to any person obtaining a
+copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be included
+in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
diff --git a/lib/subunit/python/iso8601/README b/lib/subunit/python/iso8601/README
new file mode 100644
index 0000000000..5ec9d45597
--- /dev/null
+++ b/lib/subunit/python/iso8601/README
@@ -0,0 +1,26 @@
+A simple package to deal with ISO 8601 date time formats.
+
+ISO 8601 defines a neutral, unambiguous date string format, which also
+has the property of sorting naturally.
+
+e.g. YYYY-MM-DDTHH:MM:SSZ or 2007-01-25T12:00:00Z
+
+Currently this covers only the most common date formats encountered, not
+all of ISO 8601 is handled.
+
+Currently the following formats are handled:
+
+* 2006-01-01T00:00:00Z
+* 2006-01-01T00:00:00[+-]00:00
+
+I'll add more as I encounter them in my day to day life. Patches with
+new formats and tests will be gratefully accepted of course :)
+
+References:
+
+* http://www.cl.cam.ac.uk/~mgk25/iso-time.html - simple overview
+
+* http://hydracen.com/dx/iso8601.htm - more detailed enumeration of
+ valid formats.
+
+See the LICENSE file for the license this package is released under.
diff --git a/lib/subunit/python/iso8601/README.subunit b/lib/subunit/python/iso8601/README.subunit
new file mode 100644
index 0000000000..d1ed8a11a6
--- /dev/null
+++ b/lib/subunit/python/iso8601/README.subunit
@@ -0,0 +1,5 @@
+This is a [slightly rearranged] import of http://pypi.python.org/pypi/iso8601/
+version 0.1.4. The OS X hidden files have been stripped, and the package
+turned into a single module, to simplify installation. The remainder of the
+source distribution is included in the subunit source tree at python/iso8601
+for reference.
diff --git a/lib/subunit/python/iso8601/setup.py b/lib/subunit/python/iso8601/setup.py
new file mode 100644
index 0000000000..cdb61ecf6a
--- /dev/null
+++ b/lib/subunit/python/iso8601/setup.py
@@ -0,0 +1,58 @@
+try:
+ from setuptools import setup
+except ImportError:
+ from distutils import setup
+
+long_description="""Simple module to parse ISO 8601 dates
+
+This module parses the most common forms of ISO 8601 date strings (e.g.
+2007-01-14T20:34:22+00:00) into datetime objects.
+
+>>> import iso8601
+>>> iso8601.parse_date("2007-01-25T12:00:00Z")
+datetime.datetime(2007, 1, 25, 12, 0, tzinfo=<iso8601.iso8601.Utc ...>)
+>>>
+
+Changes
+=======
+
+0.1.4
+-----
+
+* The default_timezone argument wasn't being passed through correctly,
+ UTC was being used in every case. Fixes issue 10.
+
+0.1.3
+-----
+
+* Fixed the microsecond handling, the generated microsecond values were
+ way too small. Fixes issue 9.
+
+0.1.2
+-----
+
+* Adding ParseError to __all__ in iso8601 module, allows people to import it.
+ Addresses issue 7.
+* Be a little more flexible when dealing with dates without leading zeroes.
+ This violates the spec a little, but handles more dates as seen in the
+ field. Addresses issue 6.
+* Allow date/time separators other than T.
+
+0.1.1
+-----
+
+* When parsing dates without a timezone the specified default is used. If no
+ default is specified then UTC is used. Addresses issue 4.
+"""
+
+setup(
+ name="iso8601",
+ version="0.1.4",
+ description=long_description.split("\n")[0],
+ long_description=long_description,
+ author="Michael Twomey",
+ author_email="micktwomey+iso8601@gmail.com",
+ url="http://code.google.com/p/pyiso8601/",
+ packages=["iso8601"],
+ license="MIT",
+)
diff --git a/lib/subunit/python/iso8601/test_iso8601.py b/lib/subunit/python/iso8601/test_iso8601.py
new file mode 100644
index 0000000000..ff9e2731cf
--- /dev/null
+++ b/lib/subunit/python/iso8601/test_iso8601.py
@@ -0,0 +1,111 @@
+import iso8601
+
+def test_iso8601_regex():
+ assert iso8601.ISO8601_REGEX.match("2006-10-11T00:14:33Z")
+
+def test_timezone_regex():
+ assert iso8601.TIMEZONE_REGEX.match("+01:00")
+ assert iso8601.TIMEZONE_REGEX.match("+00:00")
+ assert iso8601.TIMEZONE_REGEX.match("+01:20")
+ assert iso8601.TIMEZONE_REGEX.match("-01:00")
+
+def test_parse_date():
+ d = iso8601.parse_date("2006-10-20T15:34:56Z")
+ assert d.year == 2006
+ assert d.month == 10
+ assert d.day == 20
+ assert d.hour == 15
+ assert d.minute == 34
+ assert d.second == 56
+ assert d.tzinfo == iso8601.UTC
+
+def test_parse_date_fraction():
+ d = iso8601.parse_date("2006-10-20T15:34:56.123Z")
+ assert d.year == 2006
+ assert d.month == 10
+ assert d.day == 20
+ assert d.hour == 15
+ assert d.minute == 34
+ assert d.second == 56
+ assert d.microsecond == 123000
+ assert d.tzinfo == iso8601.UTC
+
+def test_parse_date_fraction_2():
+ """From bug 6
+
+ """
+ d = iso8601.parse_date("2007-5-7T11:43:55.328Z'")
+ assert d.year == 2007
+ assert d.month == 5
+ assert d.day == 7
+ assert d.hour == 11
+ assert d.minute == 43
+ assert d.second == 55
+ assert d.microsecond == 328000
+ assert d.tzinfo == iso8601.UTC
+
+def test_parse_date_tz():
+ d = iso8601.parse_date("2006-10-20T15:34:56.123+02:30")
+ assert d.year == 2006
+ assert d.month == 10
+ assert d.day == 20
+ assert d.hour == 15
+ assert d.minute == 34
+ assert d.second == 56
+ assert d.microsecond == 123000
+ assert d.tzinfo.tzname(None) == "+02:30"
+ offset = d.tzinfo.utcoffset(None)
+ assert offset.days == 0
+ assert offset.seconds == 60 * 60 * 2.5
+
+def test_parse_invalid_date():
+ try:
+ iso8601.parse_date(None)
+ except iso8601.ParseError:
+ pass
+ else:
+ assert 1 == 2
+
+def test_parse_invalid_date2():
+ try:
+ iso8601.parse_date("23")
+ except iso8601.ParseError:
+ pass
+ else:
+ assert 1 == 2
+
+def test_parse_no_timezone():
+ """issue 4 - Handle datetime string without timezone
+
+ This tests what happens when you parse a date with no timezone. While not
+ strictly correct this is quite common. I'll assume UTC for the time zone
+ in this case.
+ """
+ d = iso8601.parse_date("2007-01-01T08:00:00")
+ assert d.year == 2007
+ assert d.month == 1
+ assert d.day == 1
+ assert d.hour == 8
+ assert d.minute == 0
+ assert d.second == 0
+ assert d.microsecond == 0
+ assert d.tzinfo == iso8601.UTC
+
+def test_parse_no_timezone_different_default():
+ tz = iso8601.FixedOffset(2, 0, "test offset")
+ d = iso8601.parse_date("2007-01-01T08:00:00", default_timezone=tz)
+ assert d.tzinfo == tz
+
+def test_space_separator():
+ """Handle a separator other than T
+
+ """
+ d = iso8601.parse_date("2007-06-23 06:40:34.00Z")
+ assert d.year == 2007
+ assert d.month == 6
+ assert d.day == 23
+ assert d.hour == 6
+ assert d.minute == 40
+ assert d.second == 34
+ assert d.microsecond == 0
+ assert d.tzinfo == iso8601.UTC
diff --git a/lib/subunit/python/subunit/__init__.py b/lib/subunit/python/subunit/__init__.py
index 406cd8765b..6e8df90317 100644
--- a/lib/subunit/python/subunit/__init__.py
+++ b/lib/subunit/python/subunit/__init__.py
@@ -1,28 +1,151 @@
#
-# subunit: extensions to python unittest to get test results from subprocesses.
+# subunit: extensions to Python unittest to get test results from subprocesses.
# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
-# Copyright (C) 2007 Jelmer Vernooij <jelmer@samba.org>
#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation; either version 3 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+# 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.
#
+"""Subunit - a streaming test protocol
+
+Overview
+++++++++
+
+The ``subunit`` Python package provides a number of ``unittest`` extensions
+which can be used to cause tests to output Subunit, to parse Subunit streams
+into test activity, perform seamless test isolation within a regular test
+case and variously sort, filter and report on test runs.
+
+
+Key Classes
+-----------
+
+The ``subunit.TestProtocolClient`` class is a ``unittest.TestResult``
+extension which will translate a test run into a Subunit stream.
+
+The ``subunit.ProtocolTestCase`` class is an adapter between the Subunit wire
+protocol and the ``unittest.TestCase`` object protocol. It is used to translate
+a stream into a test run, which regular ``unittest.TestResult`` objects can
+process and report/inspect.
+
+Subunit has support for non-blocking usage too, for use with asyncore or
+Twisted. See the ``TestProtocolServer`` parser class for more details.
+
+Subunit includes extensions to the Python ``TestResult`` protocol. These are
+all done in a compatible manner: ``TestResult`` objects that do not implement
+the extension methods will not cause errors to be raised, instead the extension
+will either lose fidelity (for instance, folding expected failures to success
+in Python versions < 2.7 or 3.1), or discard the extended data (for extra
+details, tags, timestamping and progress markers).
+
+The test outcome methods ``addSuccess``, ``addError``, ``addExpectedFailure``,
+``addFailure``, ``addSkip`` take an optional keyword parameter ``details``
+which can be used instead of the usual python unittest parameter.
+When used the value of details should be a dict from ``string`` to
+``testtools.content.Content`` objects. This is a draft API being worked on with
+the Python Testing In Python mail list, with the goal of permitting a common
+way to provide additional data beyond a traceback, such as captured data from
+disk, logging messages etc. The reference for this API is in testtools (0.9.0
+and newer).
+
+The ``tags(new_tags, gone_tags)`` method is called (if present) to add or
+remove tags in the test run that is currently executing. If called when no
+test is in progress (that is, if called outside of the ``startTest``,
+``stopTest`` pair), the the tags apply to all sebsequent tests. If called
+when a test is in progress, then the tags only apply to that test.
+
+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.PROGRESS_CUR,
+subunit.PROGRESS_SET, PROGRESS_PUSH, PROGRESS_POP. Push and pop operations
+ignore the offset parameter.
+
+
+Python test support
+-------------------
+
+``subunit.run`` is a convenience wrapper to run a Python test suite via
+the command line, reporting via Subunit::
+
+ $ python -m subunit.run mylib.tests.test_suite
+
+The ``IsolatedTestSuite`` class is a TestSuite that forks before running its
+tests, allowing isolation between the test runner and some tests.
+
+Similarly, ``IsolatedTestCase`` is a base class which can be subclassed to get
+tests that will fork() before that individual test is run.
+
+`ExecTestCase`` is a convenience wrapper for running an external
+program to get a Subunit stream and then report that back to an arbitrary
+result object::
+
+ class AggregateTests(subunit.ExecTestCase):
+
+ def test_script_one(self):
+ './bin/script_one'
+
+ def test_script_two(self):
+ './bin/script_two'
+
+ # Normally your normal test loading would take of this automatically,
+ # It is only spelt out in detail here for clarity.
+ suite = unittest.TestSuite([AggregateTests("test_script_one"),
+ AggregateTests("test_script_two")])
+ # Create any TestResult class you like.
+ result = unittest._TextTestResult(sys.stdout)
+ # And run your suite as normal, Subunit will exec each external script as
+ # needed and report to your result object.
+ suite.run(result)
+
+Utility modules
+---------------
+
+* subunit.chunked contains HTTP chunked encoding/decoding logic.
+* subunit.test_results contains TestResult helper classes.
+"""
+
+import datetime
import os
+import re
from StringIO import StringIO
+import subprocess
import sys
import unittest
+import iso8601
+from testtools import content, content_type, ExtendedToOriginalDecorator
+try:
+ from testtools.testresult.real import _StringException
+ RemoteException = _StringException
+ _remote_exception_str = '_StringException' # For testing.
+except ImportError:
+ raise ImportError ("testtools.testresult.real does not contain "
+ "_StringException, check your version.")
+
+
+from testtools.testresult.real import _StringException
+
+import chunked, details, test_results
+
+
+PROGRESS_SET = 0
+PROGRESS_CUR = 1
+PROGRESS_PUSH = 2
+PROGRESS_POP = 3
+
+
def test_suite():
import subunit.tests
return subunit.tests.test_suite()
@@ -42,211 +165,558 @@ def join_dir(base_path, path):
return os.path.join(os.path.dirname(os.path.abspath(base_path)), path)
-class TestProtocolServer(object):
- """A class for receiving results from a TestProtocol client."""
-
- OUTSIDE_TEST = 0
- TEST_STARTED = 1
- READING_FAILURE = 2
- READING_ERROR = 3
-
- def __init__(self, client, stream=sys.stdout):
- """Create a TestProtocol server instance.
-
- client should be an object that provides
- - startTest
- - addSuccess
- - addFailure
- - addError
- - stopTest
- methods, i.e. a TestResult.
- """
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.client = client
- self._stream = stream
-
- def _addError(self, offset, line):
- if (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description == line[offset:-1]):
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.current_test_description = None
- self.client.addError(self._current_test, RemoteError(""))
- self.client.stopTest(self._current_test)
- self._current_test = None
- elif (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description + " [" == line[offset:-1]):
- self.state = TestProtocolServer.READING_ERROR
- self._message = ""
- else:
- self.stdOutLineReceived(line)
-
- def _addFailure(self, offset, line):
- if (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description == line[offset:-1]):
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.current_test_description = None
- self.client.addFailure(self._current_test, RemoteError())
- self.client.stopTest(self._current_test)
- elif (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description + " [" == line[offset:-1]):
- self.state = TestProtocolServer.READING_FAILURE
- self._message = ""
+def tags_to_new_gone(tags):
+ """Split a list of tags into a new_set and a gone_set."""
+ new_tags = set()
+ gone_tags = set()
+ for tag in tags:
+ if tag[0] == '-':
+ gone_tags.add(tag[1:])
else:
- self.stdOutLineReceived(line)
-
- def _addSuccess(self, offset, line):
- if (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description == line[offset:-1]):
- self.client.addSuccess(self._current_test)
- self.client.stopTest(self._current_test)
- self.current_test_description = None
- self._current_test = None
- self.state = TestProtocolServer.OUTSIDE_TEST
+ new_tags.add(tag)
+ return new_tags, gone_tags
+
+
+class DiscardStream(object):
+ """A filelike object which discards what is written to it."""
+
+ def write(self, bytes):
+ pass
+
+
+class _ParserState(object):
+ """State for the subunit parser."""
+
+ def __init__(self, parser):
+ self.parser = parser
+
+ def addError(self, offset, line):
+ """An 'error:' directive has been read."""
+ self.parser.stdOutLineReceived(line)
+
+ def addExpectedFail(self, offset, line):
+ """An 'xfail:' directive has been read."""
+ self.parser.stdOutLineReceived(line)
+
+ def addFailure(self, offset, line):
+ """A 'failure:' directive has been read."""
+ self.parser.stdOutLineReceived(line)
+
+ def addSkip(self, offset, line):
+ """A 'skip:' directive has been read."""
+ self.parser.stdOutLineReceived(line)
+
+ def addSuccess(self, offset, line):
+ """A 'success:' directive has been read."""
+ self.parser.stdOutLineReceived(line)
+
+ def lineReceived(self, line):
+ """a line has been received."""
+ parts = line.split(None, 1)
+ if len(parts) == 2:
+ cmd, rest = parts
+ offset = len(cmd) + 1
+ cmd = cmd.strip(':')
+ if cmd in ('test', 'testing'):
+ self.startTest(offset, line)
+ elif cmd == 'error':
+ self.addError(offset, line)
+ elif cmd == 'failure':
+ self.addFailure(offset, line)
+ elif cmd == 'progress':
+ self.parser._handleProgress(offset, line)
+ elif cmd == 'skip':
+ self.addSkip(offset, line)
+ elif cmd in ('success', 'successful'):
+ self.addSuccess(offset, line)
+ elif cmd in ('tags',):
+ self.parser._handleTags(offset, line)
+ self.parser.subunitLineReceived(line)
+ elif cmd in ('time',):
+ self.parser._handleTime(offset, line)
+ self.parser.subunitLineReceived(line)
+ elif cmd == 'xfail':
+ self.addExpectedFail(offset, line)
+ else:
+ self.parser.stdOutLineReceived(line)
else:
- self.stdOutLineReceived(line)
+ self.parser.stdOutLineReceived(line)
+
+ def lostConnection(self):
+ """Connection lost."""
+ self.parser._lostConnectionInTest('unknown state of ')
+
+ def startTest(self, offset, line):
+ """A test start command received."""
+ self.parser.stdOutLineReceived(line)
+
- def _appendMessage(self, line):
- if line[0:2] == " ]":
- # quoted ] start
- self._message += line[1:]
+class _InTest(_ParserState):
+ """State for the subunit parser after reading a test: directive."""
+
+ def _outcome(self, offset, line, no_details, details_state):
+ """An outcome directive has been read.
+
+ :param no_details: Callable to call when no details are presented.
+ :param details_state: The state to switch to for details
+ processing of this outcome.
+ """
+ if self.parser.current_test_description == line[offset:-1]:
+ self.parser._state = self.parser._outside_test
+ self.parser.current_test_description = None
+ no_details()
+ self.parser.client.stopTest(self.parser._current_test)
+ self.parser._current_test = None
+ self.parser.subunitLineReceived(line)
+ elif self.parser.current_test_description + " [" == line[offset:-1]:
+ self.parser._state = details_state
+ details_state.set_simple()
+ self.parser.subunitLineReceived(line)
+ elif self.parser.current_test_description + " [ multipart" == \
+ line[offset:-1]:
+ self.parser._state = details_state
+ details_state.set_multipart()
+ self.parser.subunitLineReceived(line)
else:
- self._message += line
-
- def endQuote(self, line):
- if self.state == TestProtocolServer.READING_FAILURE:
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.current_test_description = None
- self.client.addFailure(self._current_test,
- RemoteError(self._message))
- self.client.stopTest(self._current_test)
- elif self.state == TestProtocolServer.READING_ERROR:
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.current_test_description = None
- self.client.addError(self._current_test,
- RemoteError(self._message))
- self.client.stopTest(self._current_test)
+ self.parser.stdOutLineReceived(line)
+
+ def _error(self):
+ self.parser.client.addError(self.parser._current_test,
+ details={})
+
+ def addError(self, offset, line):
+ """An 'error:' directive has been read."""
+ self._outcome(offset, line, self._error,
+ self.parser._reading_error_details)
+
+ def _xfail(self):
+ self.parser.client.addExpectedFailure(self.parser._current_test,
+ details={})
+
+ def addExpectedFail(self, offset, line):
+ """An 'xfail:' directive has been read."""
+ self._outcome(offset, line, self._xfail,
+ self.parser._reading_xfail_details)
+
+ def _failure(self):
+ self.parser.client.addFailure(self.parser._current_test, details={})
+
+ def addFailure(self, offset, line):
+ """A 'failure:' directive has been read."""
+ self._outcome(offset, line, self._failure,
+ self.parser._reading_failure_details)
+
+ def _skip(self):
+ self.parser.client.addSkip(self.parser._current_test, details={})
+
+ def addSkip(self, offset, line):
+ """A 'skip:' directive has been read."""
+ self._outcome(offset, line, self._skip,
+ self.parser._reading_skip_details)
+
+ def _succeed(self):
+ self.parser.client.addSuccess(self.parser._current_test, details={})
+
+ def addSuccess(self, offset, line):
+ """A 'success:' directive has been read."""
+ self._outcome(offset, line, self._succeed,
+ self.parser._reading_success_details)
+
+ def lostConnection(self):
+ """Connection lost."""
+ self.parser._lostConnectionInTest('')
+
+
+class _OutSideTest(_ParserState):
+ """State for the subunit parser outside of a test context."""
+
+ def lostConnection(self):
+ """Connection lost."""
+
+ def startTest(self, offset, line):
+ """A test start command received."""
+ self.parser._state = self.parser._in_test
+ self.parser._current_test = RemotedTestCase(line[offset:-1])
+ self.parser.current_test_description = line[offset:-1]
+ self.parser.client.startTest(self.parser._current_test)
+ self.parser.subunitLineReceived(line)
+
+
+class _ReadingDetails(_ParserState):
+ """Common logic for readin state details."""
+
+ def endDetails(self):
+ """The end of a details section has been reached."""
+ self.parser._state = self.parser._outside_test
+ self.parser.current_test_description = None
+ self._report_outcome()
+ self.parser.client.stopTest(self.parser._current_test)
+
+ def lineReceived(self, line):
+ """a line has been received."""
+ self.details_parser.lineReceived(line)
+ self.parser.subunitLineReceived(line)
+
+ def lostConnection(self):
+ """Connection lost."""
+ self.parser._lostConnectionInTest('%s report of ' %
+ self._outcome_label())
+
+ def _outcome_label(self):
+ """The label to describe this outcome."""
+ raise NotImplementedError(self._outcome_label)
+
+ def set_simple(self):
+ """Start a simple details parser."""
+ self.details_parser = details.SimpleDetailsParser(self)
+
+ def set_multipart(self):
+ """Start a multipart details parser."""
+ self.details_parser = details.MultipartDetailsParser(self)
+
+
+class _ReadingFailureDetails(_ReadingDetails):
+ """State for the subunit parser when reading failure details."""
+
+ def _report_outcome(self):
+ self.parser.client.addFailure(self.parser._current_test,
+ details=self.details_parser.get_details())
+
+ def _outcome_label(self):
+ return "failure"
+
+
+class _ReadingErrorDetails(_ReadingDetails):
+ """State for the subunit parser when reading error details."""
+
+ def _report_outcome(self):
+ self.parser.client.addError(self.parser._current_test,
+ details=self.details_parser.get_details())
+
+ def _outcome_label(self):
+ return "error"
+
+
+class _ReadingExpectedFailureDetails(_ReadingDetails):
+ """State for the subunit parser when reading xfail details."""
+
+ def _report_outcome(self):
+ self.parser.client.addExpectedFailure(self.parser._current_test,
+ details=self.details_parser.get_details())
+
+ def _outcome_label(self):
+ return "xfail"
+
+
+class _ReadingSkipDetails(_ReadingDetails):
+ """State for the subunit parser when reading skip details."""
+
+ def _report_outcome(self):
+ self.parser.client.addSkip(self.parser._current_test,
+ details=self.details_parser.get_details("skip"))
+
+ def _outcome_label(self):
+ return "skip"
+
+
+class _ReadingSuccessDetails(_ReadingDetails):
+ """State for the subunit parser when reading success details."""
+
+ def _report_outcome(self):
+ self.parser.client.addSuccess(self.parser._current_test,
+ details=self.details_parser.get_details("success"))
+
+ def _outcome_label(self):
+ return "success"
+
+
+class TestProtocolServer(object):
+ """A parser for subunit.
+
+ :ivar tags: The current tags associated with the protocol stream.
+ """
+
+ def __init__(self, client, stream=None, forward_stream=None):
+ """Create a TestProtocolServer instance.
+
+ :param client: An object meeting the unittest.TestResult protocol.
+ :param stream: The stream that lines received which are not part of the
+ subunit protocol should be written to. This allows custom handling
+ of mixed protocols. By default, sys.stdout will be used for
+ convenience.
+ :param forward_stream: A stream to forward subunit lines to. This
+ allows a filter to forward the entire stream while still parsing
+ and acting on it. By default forward_stream is set to
+ DiscardStream() and no forwarding happens.
+ """
+ self.client = ExtendedToOriginalDecorator(client)
+ if stream is None:
+ stream = sys.stdout
+ self._stream = stream
+ self._forward_stream = forward_stream or DiscardStream()
+ # state objects we can switch too
+ self._in_test = _InTest(self)
+ self._outside_test = _OutSideTest(self)
+ self._reading_error_details = _ReadingErrorDetails(self)
+ self._reading_failure_details = _ReadingFailureDetails(self)
+ self._reading_skip_details = _ReadingSkipDetails(self)
+ self._reading_success_details = _ReadingSuccessDetails(self)
+ self._reading_xfail_details = _ReadingExpectedFailureDetails(self)
+ # start with outside test.
+ self._state = self._outside_test
+
+ def _handleProgress(self, offset, line):
+ """Process a progress directive."""
+ line = line[offset:].strip()
+ if line[0] in '+-':
+ whence = PROGRESS_CUR
+ delta = int(line)
+ elif line == "push":
+ whence = PROGRESS_PUSH
+ delta = None
+ elif line == "pop":
+ whence = PROGRESS_POP
+ delta = None
else:
- self.stdOutLineReceived(line)
+ whence = PROGRESS_SET
+ delta = int(line)
+ self.client.progress(delta, whence)
+
+ def _handleTags(self, offset, line):
+ """Process a tags command."""
+ tags = line[offset:].split()
+ new_tags, gone_tags = tags_to_new_gone(tags)
+ self.client.tags(new_tags, gone_tags)
+
+ def _handleTime(self, offset, line):
+ # Accept it, but do not do anything with it yet.
+ try:
+ event_time = iso8601.parse_date(line[offset:-1])
+ except TypeError, e:
+ raise TypeError("Failed to parse %r, got %r" % (line, e))
+ self.client.time(event_time)
def lineReceived(self, line):
"""Call the appropriate local method for the received line."""
- if line == "]\n":
- self.endQuote(line)
- elif (self.state == TestProtocolServer.READING_FAILURE or
- self.state == TestProtocolServer.READING_ERROR):
- self._appendMessage(line)
- else:
- parts = line.split(None, 1)
- if len(parts) == 2:
- cmd, rest = parts
- offset = len(cmd) + 1
- cmd = cmd.strip(':')
- if cmd in ('test', 'testing'):
- self._startTest(offset, line)
- elif cmd == 'error':
- self._addError(offset, line)
- elif cmd == 'failure':
- self._addFailure(offset, line)
- elif cmd in ('success', 'successful'):
- self._addSuccess(offset, line)
- else:
- self.stdOutLineReceived(line)
- else:
- self.stdOutLineReceived(line)
+ self._state.lineReceived(line)
+
+ def _lostConnectionInTest(self, state_string):
+ error_string = "lost connection during %stest '%s'" % (
+ state_string, self.current_test_description)
+ self.client.addError(self._current_test, RemoteError(error_string))
+ self.client.stopTest(self._current_test)
def lostConnection(self):
"""The input connection has finished."""
- if self.state == TestProtocolServer.TEST_STARTED:
- self.client.addError(self._current_test,
- RemoteError("lost connection during test '%s'"
- % self.current_test_description))
- self.client.stopTest(self._current_test)
- elif self.state == TestProtocolServer.READING_ERROR:
- self.client.addError(self._current_test,
- RemoteError("lost connection during "
- "error report of test "
- "'%s'" %
- self.current_test_description))
- self.client.stopTest(self._current_test)
- elif self.state == TestProtocolServer.READING_FAILURE:
- self.client.addError(self._current_test,
- RemoteError("lost connection during "
- "failure report of test "
- "'%s'" %
- self.current_test_description))
- self.client.stopTest(self._current_test)
+ self._state.lostConnection()
def readFrom(self, pipe):
+ """Blocking convenience API to parse an entire stream.
+
+ :param pipe: A file-like object supporting readlines().
+ :return: None.
+ """
for line in pipe.readlines():
self.lineReceived(line)
self.lostConnection()
def _startTest(self, offset, line):
"""Internal call to change state machine. Override startTest()."""
- if self.state == TestProtocolServer.OUTSIDE_TEST:
- self.state = TestProtocolServer.TEST_STARTED
- self._current_test = RemotedTestCase(line[offset:-1])
- self.current_test_description = line[offset:-1]
- self.client.startTest(self._current_test)
- else:
- self.stdOutLineReceived(line)
+ self._state.startTest(offset, line)
+
+ def subunitLineReceived(self, line):
+ self._forward_stream.write(line)
def stdOutLineReceived(self, line):
self._stream.write(line)
-class RemoteException(Exception):
- """An exception that occured remotely to python."""
-
- def __eq__(self, other):
- try:
- return self.args == other.args
- except AttributeError:
- return False
-
-
class TestProtocolClient(unittest.TestResult):
- """A class that looks like a TestResult and informs a TestProtocolServer."""
+ """A TestResult which generates a subunit stream for a test run.
+
+ # Get a TestSuite or TestCase to run
+ suite = make_suite()
+ # Create a stream (any object with a 'write' method)
+ stream = file('tests.log', 'wb')
+ # Create a subunit result object which will output to the stream
+ result = subunit.TestProtocolClient(stream)
+ # Optionally, to get timing data for performance analysis, wrap the
+ # serialiser with a timing decorator
+ result = subunit.test_results.AutoTimingTestResultDecorator(result)
+ # Run the test suite reporting to the subunit result object
+ suite.run(result)
+ # Close the stream.
+ stream.close()
+ """
def __init__(self, stream):
- super(TestProtocolClient, self).__init__()
+ unittest.TestResult.__init__(self)
self._stream = stream
- def addError(self, test, error):
- """Report an error in test test."""
- self._stream.write("error: %s [\n" % (test.shortDescription() or str(test)))
- for line in self._exc_info_to_string(error, test).splitlines():
- self._stream.write("%s\n" % line)
+ def addError(self, test, error=None, details=None):
+ """Report an error in test test.
+
+ Only one of error and details should be provided: conceptually there
+ are two separate methods:
+ addError(self, test, error)
+ addError(self, test, details)
+
+ :param error: Standard unittest positional argument form - an
+ exc_info tuple.
+ :param details: New Testing-in-python drafted API; a dict from string
+ to subunit.Content objects.
+ """
+ self._addOutcome("error", test, error=error, details=details)
+
+ def addExpectedFailure(self, test, error=None, details=None):
+ """Report an expected failure in test test.
+
+ Only one of error and details should be provided: conceptually there
+ are two separate methods:
+ addError(self, test, error)
+ addError(self, test, details)
+
+ :param error: Standard unittest positional argument form - an
+ exc_info tuple.
+ :param details: New Testing-in-python drafted API; a dict from string
+ to subunit.Content objects.
+ """
+ self._addOutcome("xfail", test, error=error, details=details)
+
+ def addFailure(self, test, error=None, details=None):
+ """Report a failure in test test.
+
+ Only one of error and details should be provided: conceptually there
+ are two separate methods:
+ addFailure(self, test, error)
+ addFailure(self, test, details)
+
+ :param error: Standard unittest positional argument form - an
+ exc_info tuple.
+ :param details: New Testing-in-python drafted API; a dict from string
+ to subunit.Content objects.
+ """
+ self._addOutcome("failure", test, error=error, details=details)
+
+ def _addOutcome(self, outcome, test, error=None, details=None):
+ """Report a failure in test test.
+
+ Only one of error and details should be provided: conceptually there
+ are two separate methods:
+ addOutcome(self, test, error)
+ addOutcome(self, test, details)
+
+ :param outcome: A string describing the outcome - used as the
+ event name in the subunit stream.
+ :param error: Standard unittest positional argument form - an
+ exc_info tuple.
+ :param details: New Testing-in-python drafted API; a dict from string
+ to subunit.Content objects.
+ """
+ self._stream.write("%s: %s" % (outcome, test.id()))
+ if error is None and details is None:
+ raise ValueError
+ if error is not None:
+ self._stream.write(" [\n")
+ for line in self._exc_info_to_string(error, test).splitlines():
+ self._stream.write("%s\n" % line)
+ else:
+ self._write_details(details)
self._stream.write("]\n")
- super(TestProtocolClient, self).addError(test, error)
- def addFailure(self, test, error):
- """Report a failure in test test."""
- self._stream.write("failure: %s [\n" % (test.shortDescription() or str(test)))
- for line in self._exc_info_to_string(error, test).splitlines():
- self._stream.write("%s\n" % line)
- self._stream.write("]\n")
- super(TestProtocolClient, self).addFailure(test, error)
+ def addSkip(self, test, reason=None, details=None):
+ """Report a skipped test."""
+ if reason is None:
+ self._addOutcome("skip", test, error=None, details=details)
+ else:
+ self._stream.write("skip: %s [\n" % test.id())
+ self._stream.write("%s\n" % reason)
+ self._stream.write("]\n")
- def addSuccess(self, test):
+ def addSuccess(self, test, details=None):
"""Report a success in a test."""
- self._stream.write("successful: %s\n" % (test.shortDescription() or str(test)))
- super(TestProtocolClient, self).addSuccess(test)
+ self._stream.write("successful: %s" % test.id())
+ if not details:
+ self._stream.write("\n")
+ else:
+ self._write_details(details)
+ self._stream.write("]\n")
+ addUnexpectedSuccess = addSuccess
def startTest(self, test):
"""Mark a test as starting its test run."""
- self._stream.write("test: %s\n" % (test.shortDescription() or str(test)))
- super(TestProtocolClient, self).startTest(test)
+ 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 PROGRESS_CUR, then offset increases/decreases the
+ remaining test count. If whence is PROGRESS_SET, then offset
+ specifies exactly the remaining test count.
+ :param whence: One of PROGRESS_CUR, PROGRESS_SET, PROGRESS_PUSH,
+ PROGRESS_POP.
+ """
+ if whence == PROGRESS_CUR and offset > -1:
+ prefix = "+"
+ elif whence == PROGRESS_PUSH:
+ prefix = ""
+ offset = "push"
+ elif whence == PROGRESS_POP:
+ prefix = ""
+ offset = "pop"
+ else:
+ prefix = ""
+ self._stream.write("progress: %s%s\n" % (prefix, offset))
+
+ def time(self, a_datetime):
+ """Inform the client of the time.
+
+ ":param datetime: A datetime.datetime object.
+ """
+ time = a_datetime.astimezone(iso8601.Utc())
+ self._stream.write("time: %04d-%02d-%02d %02d:%02d:%02d.%06dZ\n" % (
+ time.year, time.month, time.day, time.hour, time.minute,
+ time.second, time.microsecond))
+
+ def _write_details(self, details):
+ """Output details to the stream.
+
+ :param details: An extended details dict for a test outcome.
+ """
+ self._stream.write(" [ multipart\n")
+ for name, content in sorted(details.iteritems()):
+ self._stream.write("Content-Type: %s/%s" %
+ (content.content_type.type, content.content_type.subtype))
+ parameters = content.content_type.parameters
+ if parameters:
+ self._stream.write(";")
+ param_strs = []
+ for param, value in parameters.iteritems():
+ param_strs.append("%s=%s" % (param, value))
+ self._stream.write(",".join(param_strs))
+ self._stream.write("\n%s\n" % name)
+ encoder = chunked.Encoder(self._stream)
+ map(encoder.write, content.iter_bytes())
+ encoder.close()
+
+ def done(self):
+ """Obey the testtools result.done() interface."""
def RemoteError(description=""):
- if description == "":
- description = "\n"
- return (RemoteException, RemoteException(description), None)
+ return (_StringException, _StringException(description), None)
class RemotedTestCase(unittest.TestCase):
- """A class to represent test cases run in child processes."""
+ """A class to represent test cases run in child processes.
+
+ Instances of this class are used to provide the Python test API a TestCase
+ that can be printed to the screen, introspected for metadata and so on.
+ However, as they are a simply a memoisation of a test that was actually
+ run in the past by a separate process, they cannot perform any interactive
+ actions.
+ """
def __eq__ (self, other):
try:
@@ -272,7 +742,7 @@ class RemotedTestCase(unittest.TestCase):
return self.__description
def id(self):
- return "%s.%s" % (self._strclass(), self.__description)
+ return "%s" % (self.__description,)
def __str__(self):
return "%s (%s)" % (self.__description, self._strclass())
@@ -318,12 +788,18 @@ class ExecTestCase(unittest.TestCase):
def _run(self, result):
protocol = TestProtocolServer(result)
- output = os.popen(self.script, mode='r')
- protocol.readFrom(output)
+ output = subprocess.Popen(self.script, shell=True,
+ stdout=subprocess.PIPE).communicate()[0]
+ protocol.readFrom(StringIO(output))
class IsolatedTestCase(unittest.TestCase):
- """A TestCase which runs its tests in a forked process."""
+ """A TestCase which executes in a forked process.
+
+ Each test gets its own process, which has a performance overhead but will
+ provide excellent isolation from global state (such as django configs,
+ zope utilities and so on).
+ """
def run(self, result=None):
if result is None: result = self.defaultTestResult()
@@ -331,7 +807,13 @@ class IsolatedTestCase(unittest.TestCase):
class IsolatedTestSuite(unittest.TestSuite):
- """A TestCase which runs its tests in a forked process."""
+ """A TestSuite which runs its tests in a forked process.
+
+ This decorator that will fork() before running the tests and report the
+ results from the child process using a Subunit stream. This is useful for
+ handling tests that mutate global state, or are testing C extensions that
+ could crash the VM.
+ """
def run(self, result=None):
if result is None: result = unittest.TestResult()
@@ -376,13 +858,256 @@ def run_isolated(klass, self, result):
return result
-class SubunitTestRunner(object):
- def __init__(self, stream=sys.stdout):
- self.stream = stream
+def TAP2SubUnit(tap, subunit):
+ """Filter a TAP pipe into a subunit pipe.
+
+ :param tap: A tap pipe/stream/file object.
+ :param subunit: A pipe/stream/file object to write subunit results to.
+ :return: The exit code to exit with.
+ """
+ BEFORE_PLAN = 0
+ AFTER_PLAN = 1
+ SKIP_STREAM = 2
+ client = TestProtocolClient(subunit)
+ 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 _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")
+ del log[:]
+ for line in tap:
+ if state == BEFORE_PLAN:
+ match = re.match("(\d+)\.\.(\d+)\s*(?:\#\s+(.*))?\n", line)
+ if match:
+ state = AFTER_PLAN
+ _, plan_stop, comment = match.groups()
+ plan_stop = int(plan_stop)
+ 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")
+ 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)(?:\s+(.*))?)?\n", line)
+ if match:
+ # new test, emit current one.
+ _emit_test()
+ status, number, description, directive, directive_comment = match.groups()
+ if status == 'ok':
+ result = 'success'
+ else:
+ result = "failure"
+ if description is None:
+ description = ''
+ else:
+ description = ' ' + description
+ if directive is not None:
+ if directive == 'TODO':
+ result = 'xfail'
+ elif directive == 'SKIP':
+ result = 'skip'
+ if directive_comment is not None:
+ log.append(directive_comment)
+ if number is not None:
+ number = int(number)
+ while plan_start < number:
+ plan_start = _skipped_test(subunit, plan_start)
+ test_name = "test %d%s" % (plan_start, description)
+ plan_start += 1
+ continue
+ match = re.match("Bail out\!(?:\s*(.*))?\n", line)
+ if match:
+ reason, = match.groups()
+ if reason is None:
+ extra = ''
+ else:
+ extra = ' %s' % reason
+ _emit_test()
+ test_name = "Bail out!%s" % extra
+ result = "error"
+ state = SKIP_STREAM
+ continue
+ match = re.match("\#.*\n", line)
+ if match:
+ log.append(line[:-1])
+ continue
+ subunit.write(line)
+ _emit_test()
+ while plan_start <= plan_stop:
+ # record missed tests
+ plan_start = _skipped_test(subunit, plan_start)
+ return 0
+
+
+def tag_stream(original, filtered, tags):
+ """Alter tags on a stream.
+
+ :param original: The input stream.
+ :param filtered: The output stream.
+ :param tags: The tags to apply. As in a normal stream - a list of 'TAG' or
+ '-TAG' commands.
+
+ A 'TAG' command will add the tag to the output stream,
+ and override any existing '-TAG' command in that stream.
+ Specifically:
+ * A global 'tags: TAG' will be added to the start of the stream.
+ * Any tags commands with -TAG will have the -TAG removed.
+
+ A '-TAG' command will remove the TAG command from the stream.
+ Specifically:
+ * A 'tags: -TAG' command will be added to the start of the stream.
+ * Any 'tags: TAG' command will have 'TAG' removed from it.
+ Additionally, any redundant tagging commands (adding a tag globally
+ present, or removing a tag globally removed) are stripped as a
+ by-product of the filtering.
+ :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)
+ return 0
+
+
+class ProtocolTestCase(object):
+ """Subunit wire protocol to unittest.TestCase adapter.
+
+ ProtocolTestCase honours the core of ``unittest.TestCase`` protocol -
+ calling a ProtocolTestCase or invoking the run() method will make a 'test
+ run' happen. The 'test run' will simply be a replay of the test activity
+ that has been encoded into the stream. The ``unittest.TestCase`` ``debug``
+ and ``countTestCases`` methods are not supported because there isn't a
+ sensible mapping for those methods.
+
+ # Get a stream (any object with a readline() method), in this case the
+ # stream output by the example from ``subunit.TestProtocolClient``.
+ stream = file('tests.log', 'rb')
+ # Create a parser which will read from the stream and emit
+ # activity to a unittest.TestResult when run() is called.
+ suite = subunit.ProtocolTestCase(stream)
+ # Create a result object to accept the contents of that stream.
+ result = unittest._TextTestResult(sys.stdout)
+ # 'run' the tests - process the stream and feed its contents to result.
+ suite.run(result)
+ stream.close()
+
+ :seealso: TestProtocolServer (the subunit wire protocol parser).
+ """
+
+ def __init__(self, stream, passthrough=None, forward=False):
+ """Create a ProtocolTestCase reading from stream.
+
+ :param stream: A filelike object which a subunit stream can be read
+ from.
+ :param passthrough: A stream pass non subunit input on to. If not
+ supplied, the TestProtocolServer default is used.
+ :param forward: A stream to pass subunit input on to. If not supplied
+ subunit input is not forwarded.
+ """
+ self._stream = stream
+ self._passthrough = passthrough
+ self._forward = forward
+
+ def __call__(self, result=None):
+ return self.run(result)
- def run(self, test):
- "Run the given test case or test suite."
- result = TestProtocolClient(self.stream)
- test(result)
- return result
+ def run(self, result=None):
+ if result is None:
+ result = self.defaultTestResult()
+ protocol = TestProtocolServer(result, self._passthrough, self._forward)
+ line = self._stream.readline()
+ while line:
+ protocol.lineReceived(line)
+ line = self._stream.readline()
+ protocol.lostConnection()
+
+
+class TestResultStats(unittest.TestResult):
+ """A pyunit TestResult interface implementation for making statistics.
+
+ :ivar total_tests: The total tests seen.
+ :ivar passed_tests: The tests that passed.
+ :ivar failed_tests: The tests that failed.
+ :ivar seen_tags: The tags seen across all tests.
+ """
+ def __init__(self, stream):
+ """Create a TestResultStats which outputs to stream."""
+ unittest.TestResult.__init__(self)
+ self._stream = stream
+ self.failed_tests = 0
+ self.skipped_tests = 0
+ self.seen_tags = set()
+
+ @property
+ def total_tests(self):
+ return self.testsRun
+
+ def addError(self, test, err, details=None):
+ self.failed_tests += 1
+
+ def addFailure(self, test, err, details=None):
+ self.failed_tests += 1
+
+ def addSkip(self, test, reason, details=None):
+ self.skipped_tests += 1
+
+ def formatStats(self):
+ self._stream.write("Total tests: %5d\n" % self.total_tests)
+ self._stream.write("Passed tests: %5d\n" % self.passed_tests)
+ self._stream.write("Failed tests: %5d\n" % self.failed_tests)
+ self._stream.write("Skipped tests: %5d\n" % self.skipped_tests)
+ tags = sorted(self.seen_tags)
+ self._stream.write("Seen tags: %s\n" % (", ".join(tags)))
+
+ @property
+ def passed_tests(self):
+ return self.total_tests - self.failed_tests - self.skipped_tests
+
+ def tags(self, new_tags, gone_tags):
+ """Accumulate the seen tags."""
+ self.seen_tags.update(new_tags)
+
+ def wasSuccessful(self):
+ """Tells whether or not this result was a success"""
+ return self.failed_tests == 0
diff --git a/lib/subunit/python/subunit/chunked.py b/lib/subunit/python/subunit/chunked.py
new file mode 100644
index 0000000000..82e4b0ddfc
--- /dev/null
+++ b/lib/subunit/python/subunit/chunked.py
@@ -0,0 +1,164 @@
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""Encoder/decoder for http style chunked encoding."""
+
+class Decoder(object):
+ """Decode chunked content to a byte stream."""
+
+ def __init__(self, output):
+ """Create a decoder decoding to output.
+
+ :param output: A file-like object. Bytes written to the Decoder are
+ decoded to strip off the chunking and written to the output.
+ Up to a full write worth of data or a single control line may be
+ buffered (whichever is larger). The close method should be called
+ when no more data is available, to detect short streams; the
+ write method will return none-None when the end of a stream is
+ detected.
+ """
+ self.output = output
+ self.buffered_bytes = []
+ self.state = self._read_length
+ self.body_length = 0
+
+ def close(self):
+ """Close the decoder.
+
+ :raises ValueError: If the stream is incomplete ValueError is raised.
+ """
+ if self.state != self._finished:
+ raise ValueError("incomplete stream")
+
+ def _finished(self):
+ """Finished reading, return any remaining bytes."""
+ if self.buffered_bytes:
+ buffered_bytes = self.buffered_bytes
+ self.buffered_bytes = []
+ return ''.join(buffered_bytes)
+ else:
+ raise ValueError("stream is finished")
+
+ def _read_body(self):
+ """Pass body bytes to the output."""
+ while self.body_length and self.buffered_bytes:
+ if self.body_length >= len(self.buffered_bytes[0]):
+ self.output.write(self.buffered_bytes[0])
+ self.body_length -= len(self.buffered_bytes[0])
+ del self.buffered_bytes[0]
+ # No more data available.
+ if not self.body_length:
+ self.state = self._read_length
+ else:
+ self.output.write(self.buffered_bytes[0][:self.body_length])
+ self.buffered_bytes[0] = \
+ self.buffered_bytes[0][self.body_length:]
+ self.body_length = 0
+ self.state = self._read_length
+ return self.state()
+
+ def _read_length(self):
+ """Try to decode a length from the bytes."""
+ count = -1
+ match_chars = "0123456789abcdefABCDEF\r\n"
+ count_chars = []
+ for bytes in self.buffered_bytes:
+ for byte in bytes:
+ if byte not in match_chars:
+ break
+ count_chars.append(byte)
+ if byte == '\n':
+ break
+ if not count_chars:
+ return
+ if count_chars[-1][-1] != '\n':
+ return
+ count_str = ''.join(count_chars)
+ self.body_length = int(count_str[:-2], 16)
+ excess_bytes = len(count_str)
+ while excess_bytes:
+ if excess_bytes >= len(self.buffered_bytes[0]):
+ excess_bytes -= len(self.buffered_bytes[0])
+ del self.buffered_bytes[0]
+ else:
+ self.buffered_bytes[0] = self.buffered_bytes[0][excess_bytes:]
+ excess_bytes = 0
+ if not self.body_length:
+ self.state = self._finished
+ if not self.buffered_bytes:
+ # May not call into self._finished with no buffered data.
+ return ''
+ else:
+ self.state = self._read_body
+ return self.state()
+
+ def write(self, bytes):
+ """Decode bytes to the output stream.
+
+ :raises ValueError: If the stream has already seen the end of file
+ marker.
+ :returns: None, or the excess bytes beyond the end of file marker.
+ """
+ if bytes:
+ self.buffered_bytes.append(bytes)
+ return self.state()
+
+
+class Encoder(object):
+ """Encode content to a stream using HTTP Chunked coding."""
+
+ def __init__(self, output):
+ """Create an encoder encoding to output.
+
+ :param output: A file-like object. Bytes written to the Encoder
+ will be encoded using HTTP chunking. Small writes may be buffered
+ and the ``close`` method must be called to finish the stream.
+ """
+ self.output = output
+ self.buffered_bytes = []
+ self.buffer_size = 0
+
+ def flush(self, extra_len=0):
+ """Flush the encoder to the output stream.
+
+ :param extra_len: Increase the size of the chunk by this many bytes
+ to allow for a subsequent write.
+ """
+ if not self.buffer_size and not extra_len:
+ return
+ buffered_bytes = self.buffered_bytes
+ buffer_size = self.buffer_size
+ self.buffered_bytes = []
+ self.buffer_size = 0
+ self.output.write("%X\r\n" % (buffer_size + extra_len))
+ if buffer_size:
+ self.output.write(''.join(buffered_bytes))
+ return True
+
+ def write(self, bytes):
+ """Encode bytes to the output stream."""
+ bytes_len = len(bytes)
+ if self.buffer_size + bytes_len >= 65536:
+ self.flush(bytes_len)
+ self.output.write(bytes)
+ else:
+ self.buffered_bytes.append(bytes)
+ self.buffer_size += bytes_len
+
+ def close(self):
+ """Finish the stream. This does not close the output stream."""
+ self.flush()
+ self.output.write("0\r\n")
diff --git a/lib/subunit/python/subunit/details.py b/lib/subunit/python/subunit/details.py
new file mode 100644
index 0000000000..65a04046d9
--- /dev/null
+++ b/lib/subunit/python/subunit/details.py
@@ -0,0 +1,109 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""Handlers for outcome details."""
+
+from cStringIO import StringIO
+
+from testtools import content, content_type
+
+import chunked
+
+
+class DetailsParser(object):
+ """Base class/API reference for details parsing."""
+
+
+class SimpleDetailsParser(DetailsParser):
+ """Parser for single-part [] delimited details."""
+
+ def __init__(self, state):
+ self._message = ""
+ self._state = state
+
+ def lineReceived(self, line):
+ if line == "]\n":
+ self._state.endDetails()
+ return
+ if line[0:2] == " ]":
+ # quoted ] start
+ self._message += line[1:]
+ else:
+ self._message += line
+
+ def get_details(self, style=None):
+ result = {}
+ if not style:
+ result['traceback'] = content.Content(
+ content_type.ContentType("text", "x-traceback"),
+ lambda:[self._message])
+ else:
+ if style == 'skip':
+ name = 'reason'
+ else:
+ name = 'message'
+ result[name] = content.Content(
+ content_type.ContentType("text", "plain"),
+ lambda:[self._message])
+ return result
+
+ def get_message(self):
+ return self._message
+
+
+class MultipartDetailsParser(DetailsParser):
+ """Parser for multi-part [] surrounded MIME typed chunked details."""
+
+ def __init__(self, state):
+ self._state = state
+ self._details = {}
+ self._parse_state = self._look_for_content
+
+ def _look_for_content(self, line):
+ if line == "]\n":
+ self._state.endDetails()
+ return
+ # TODO error handling
+ field, value = line[:-1].split(' ', 1)
+ main, sub = value.split('/')
+ self._content_type = content_type.ContentType(main, sub)
+ self._parse_state = self._get_name
+
+ def _get_name(self, line):
+ self._name = line[:-1]
+ self._body = StringIO()
+ self._chunk_parser = chunked.Decoder(self._body)
+ self._parse_state = self._feed_chunks
+
+ def _feed_chunks(self, line):
+ residue = self._chunk_parser.write(line)
+ if residue is not None:
+ # Line based use always ends on no residue.
+ assert residue == ''
+ body = self._body
+ self._details[self._name] = content.Content(
+ self._content_type, lambda:[body.getvalue()])
+ self._chunk_parser.close()
+ self._parse_state = self._look_for_content
+
+ def get_details(self, for_skip=False):
+ return self._details
+
+ def get_message(self):
+ return None
+
+ def lineReceived(self, line):
+ self._parse_state(line)
diff --git a/lib/subunit/python/subunit/iso8601.py b/lib/subunit/python/subunit/iso8601.py
new file mode 100644
index 0000000000..93c92fb516
--- /dev/null
+++ b/lib/subunit/python/subunit/iso8601.py
@@ -0,0 +1,123 @@
+# Copyright (c) 2007 Michael Twomey
+#
+# Permission is hereby granted, free of charge, to any person obtaining a
+# copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish,
+# distribute, sublicense, and/or sell copies of the Software, and to
+# permit persons to whom the Software is furnished to do so, subject to
+# the following conditions:
+#
+# The above copyright notice and this permission notice shall be included
+# in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+"""ISO 8601 date time string parsing
+
+Basic usage:
+>>> import iso8601
+>>> iso8601.parse_date("2007-01-25T12:00:00Z")
+datetime.datetime(2007, 1, 25, 12, 0, tzinfo=<iso8601.iso8601.Utc ...>)
+>>>
+
+"""
+
+from datetime import datetime, timedelta, tzinfo
+import re
+
+__all__ = ["parse_date", "ParseError"]
+
+# Adapted from http://delete.me.uk/2005/03/iso8601.html
+ISO8601_REGEX = re.compile(r"(?P<year>[0-9]{4})(-(?P<month>[0-9]{1,2})(-(?P<day>[0-9]{1,2})"
+ r"((?P<separator>.)(?P<hour>[0-9]{2}):(?P<minute>[0-9]{2})(:(?P<second>[0-9]{2})(\.(?P<fraction>[0-9]+))?)?"
+ r"(?P<timezone>Z|(([-+])([0-9]{2}):([0-9]{2})))?)?)?)?"
+)
+TIMEZONE_REGEX = re.compile("(?P<prefix>[+-])(?P<hours>[0-9]{2}).(?P<minutes>[0-9]{2})")
+
+class ParseError(Exception):
+ """Raised when there is a problem parsing a date string"""
+
+# Yoinked from python docs
+ZERO = timedelta(0)
+class Utc(tzinfo):
+ """UTC
+
+ """
+ def utcoffset(self, dt):
+ return ZERO
+
+ def tzname(self, dt):
+ return "UTC"
+
+ def dst(self, dt):
+ return ZERO
+UTC = Utc()
+
+class FixedOffset(tzinfo):
+ """Fixed offset in hours and minutes from UTC
+
+ """
+ def __init__(self, offset_hours, offset_minutes, name):
+ self.__offset = timedelta(hours=offset_hours, minutes=offset_minutes)
+ self.__name = name
+
+ def utcoffset(self, dt):
+ return self.__offset
+
+ def tzname(self, dt):
+ return self.__name
+
+ def dst(self, dt):
+ return ZERO
+
+ def __repr__(self):
+ return "<FixedOffset %r>" % self.__name
+
+def parse_timezone(tzstring, default_timezone=UTC):
+ """Parses ISO 8601 time zone specs into tzinfo offsets
+
+ """
+ if tzstring == "Z":
+ return default_timezone
+ # This isn't strictly correct, but it's common to encounter dates without
+ # timezones so I'll assume the default (which defaults to UTC).
+ # Addresses issue 4.
+ if tzstring is None:
+ return default_timezone
+ m = TIMEZONE_REGEX.match(tzstring)
+ prefix, hours, minutes = m.groups()
+ hours, minutes = int(hours), int(minutes)
+ if prefix == "-":
+ hours = -hours
+ minutes = -minutes
+ return FixedOffset(hours, minutes, tzstring)
+
+def parse_date(datestring, default_timezone=UTC):
+ """Parses ISO 8601 dates into datetime objects
+
+ The timezone is parsed from the date string. However it is quite common to
+ have dates without a timezone (not strictly correct). In this case the
+ default timezone specified in default_timezone is used. This is UTC by
+ default.
+ """
+ if not isinstance(datestring, basestring):
+ raise ParseError("Expecting a string %r" % datestring)
+ m = ISO8601_REGEX.match(datestring)
+ if not m:
+ raise ParseError("Unable to parse date string %r" % datestring)
+ groups = m.groupdict()
+ tz = parse_timezone(groups["timezone"], default_timezone=default_timezone)
+ if groups["fraction"] is None:
+ groups["fraction"] = 0
+ else:
+ groups["fraction"] = int(float("0.%s" % groups["fraction"]) * 1e6)
+ return datetime(int(groups["year"]), int(groups["month"]), int(groups["day"]),
+ int(groups["hour"]), int(groups["minute"]), int(groups["second"]),
+ int(groups["fraction"]), tz)
diff --git a/lib/subunit/python/subunit/progress_model.py b/lib/subunit/python/subunit/progress_model.py
new file mode 100644
index 0000000000..3a6af89a33
--- /dev/null
+++ b/lib/subunit/python/subunit/progress_model.py
@@ -0,0 +1,106 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""Support for dealing with progress state."""
+
+class ProgressModel(object):
+ """A model of progress indicators as subunit defines it.
+
+ Instances of this class represent a single logical operation that is
+ progressing. The operation may have many steps, and some of those steps may
+ supply their own progress information. ProgressModel uses a nested concept
+ where the overall state can be pushed, creating new starting state, and
+ later pushed to return to the prior state. Many user interfaces will want
+ to display an overall summary though, and accordingly the pos() and width()
+ methods return overall summary information rather than information on the
+ current subtask.
+
+ The default state is 0/0 - indicating that the overall progress is unknown.
+ Anytime the denominator of pos/width is 0, rendering of a ProgressModel
+ should should take this into consideration.
+
+ :ivar: _tasks. This private attribute stores the subtasks. Each is a tuple:
+ pos, width, overall_numerator, overall_denominator. The overall fields
+ store the calculated overall numerator and denominator for the state
+ that was pushed.
+ """
+
+ def __init__(self):
+ """Create a ProgressModel.
+
+ The new model has no progress data at all - it will claim a summary
+ width of zero and position of 0.
+ """
+ self._tasks = []
+ self.push()
+
+ def adjust_width(self, offset):
+ """Adjust the with of the current subtask."""
+ self._tasks[-1][1] += offset
+
+ def advance(self):
+ """Advance the current subtask."""
+ self._tasks[-1][0] += 1
+
+ def pop(self):
+ """Pop a subtask off the ProgressModel.
+
+ See push for a description of how push and pop work.
+ """
+ self._tasks.pop()
+
+ def pos(self):
+ """Return how far through the operation has progressed."""
+ if not self._tasks:
+ return 0
+ task = self._tasks[-1]
+ if len(self._tasks) > 1:
+ # scale up the overall pos by the current task or preserve it if
+ # no current width is known.
+ offset = task[2] * (task[1] or 1)
+ else:
+ offset = 0
+ return offset + task[0]
+
+ def push(self):
+ """Push a new subtask.
+
+ After pushing a new subtask, the overall progress hasn't changed. Calls
+ to adjust_width, advance, set_width will only after the progress within
+ the range that calling 'advance' would have before - the subtask
+ represents progressing one step in the earlier task.
+
+ Call pop() to restore the progress model to the state before push was
+ called.
+ """
+ self._tasks.append([0, 0, self.pos(), self.width()])
+
+ def set_width(self, width):
+ """Set the width of the current subtask."""
+ self._tasks[-1][1] = width
+
+ def width(self):
+ """Return the total width of the operation."""
+ if not self._tasks:
+ return 0
+ task = self._tasks[-1]
+ if len(self._tasks) > 1:
+ # scale up the overall width by the current task or preserve it if
+ # no current width is known.
+ return task[3] * (task[1] or 1)
+ else:
+ return task[1]
+
diff --git a/lib/subunit/python/subunit/run.py b/lib/subunit/python/subunit/run.py
new file mode 100755
index 0000000000..2b90791d69
--- /dev/null
+++ b/lib/subunit/python/subunit/run.py
@@ -0,0 +1,46 @@
+#!/usr/bin/python
+#
+# Simple subunit testrunner for python
+# Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2007
+#
+# 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.
+#
+
+"""Run a unittest testcase reporting results as Subunit.
+
+ $ python -m subunit.run mylib.tests.test_suite
+"""
+
+import sys
+
+from subunit import TestProtocolClient
+
+
+class SubunitTestRunner(object):
+ def __init__(self, stream=sys.stdout):
+ self.stream = stream
+
+ def run(self, test):
+ "Run the given test case or test suite."
+ result = TestProtocolClient(self.stream)
+ test(result)
+ return result
+
+
+if __name__ == '__main__':
+ import optparse
+ from unittest import TestProgram
+ parser = optparse.OptionParser(__doc__)
+ args = parser.parse_args()[1]
+ runner = SubunitTestRunner()
+ program = TestProgram(module=None, argv=[sys.argv[0]] + args,
+ testRunner=runner)
diff --git a/lib/subunit/python/subunit/test_results.py b/lib/subunit/python/subunit/test_results.py
new file mode 100644
index 0000000000..4ccc2aab35
--- /dev/null
+++ b/lib/subunit/python/subunit/test_results.py
@@ -0,0 +1,334 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""TestResult helper classes used to by subunit."""
+
+import datetime
+
+import iso8601
+import testtools
+
+import subunit
+
+
+# NOT a TestResult, because we are implementing the interface, not inheriting
+# it.
+class TestResultDecorator(object):
+ """General pass-through decorator.
+
+ This provides a base that other TestResults can inherit from to
+ gain basic forwarding functionality. It also takes care of
+ handling the case where the target doesn't support newer methods
+ or features by degrading them.
+ """
+
+ def __init__(self, decorated):
+ """Create a TestResultDecorator forwarding to decorated."""
+ # Make every decorator degrade gracefully.
+ self.decorated = testtools.ExtendedToOriginalDecorator(decorated)
+
+ def startTest(self, test):
+ return self.decorated.startTest(test)
+
+ def startTestRun(self):
+ return self.decorated.startTestRun()
+
+ def stopTest(self, test):
+ return self.decorated.stopTest(test)
+
+ def stopTestRun(self):
+ return self.decorated.stopTestRun()
+
+ def addError(self, test, err=None, details=None):
+ return self.decorated.addError(test, err, details=details)
+
+ def addFailure(self, test, err=None, details=None):
+ return self.decorated.addFailure(test, err, details=details)
+
+ def addSuccess(self, test, details=None):
+ return self.decorated.addSuccess(test, details=details)
+
+ def addSkip(self, test, reason=None, details=None):
+ return self.decorated.addSkip(test, reason, details=details)
+
+ def addExpectedFailure(self, test, err=None, details=None):
+ return self.decorated.addExpectedFailure(test, err, details=details)
+
+ def addUnexpectedSuccess(self, test, details=None):
+ return self.decorated.addUnexpectedSuccess(test, details=details)
+
+ def progress(self, offset, whence):
+ return self.decorated.progress(offset, whence)
+
+ def wasSuccessful(self):
+ return self.decorated.wasSuccessful()
+
+ @property
+ def shouldStop(self):
+ return self.decorated.shouldStop
+
+ def stop(self):
+ return self.decorated.stop()
+
+ def tags(self, gone_tags, new_tags):
+ return self.decorated.time(gone_tags, new_tags)
+
+ def time(self, a_datetime):
+ return self.decorated.time(a_datetime)
+
+
+class HookedTestResultDecorator(TestResultDecorator):
+ """A TestResult which calls a hook on every event."""
+
+ def __init__(self, decorated):
+ self.super = super(HookedTestResultDecorator, self)
+ self.super.__init__(decorated)
+
+ def startTest(self, test):
+ self._before_event()
+ return self.super.startTest(test)
+
+ def startTestRun(self):
+ self._before_event()
+ return self.super.startTestRun()
+
+ def stopTest(self, test):
+ self._before_event()
+ return self.super.stopTest(test)
+
+ def stopTestRun(self):
+ self._before_event()
+ return self.super.stopTestRun()
+
+ def addError(self, test, err=None, details=None):
+ self._before_event()
+ return self.super.addError(test, err, details=details)
+
+ def addFailure(self, test, err=None, details=None):
+ self._before_event()
+ return self.super.addFailure(test, err, details=details)
+
+ def addSuccess(self, test, details=None):
+ self._before_event()
+ return self.super.addSuccess(test, details=details)
+
+ def addSkip(self, test, reason=None, details=None):
+ self._before_event()
+ return self.super.addSkip(test, reason, details=details)
+
+ def addExpectedFailure(self, test, err=None, details=None):
+ self._before_event()
+ return self.super.addExpectedFailure(test, err, details=details)
+
+ def addUnexpectedSuccess(self, test, details=None):
+ self._before_event()
+ return self.super.addUnexpectedSuccess(test, details=details)
+
+ def progress(self, offset, whence):
+ self._before_event()
+ return self.super.progress(offset, whence)
+
+ def wasSuccessful(self):
+ self._before_event()
+ return self.super.wasSuccessful()
+
+ @property
+ def shouldStop(self):
+ self._before_event()
+ return self.super.shouldStop
+
+ def stop(self):
+ self._before_event()
+ return self.super.stop()
+
+ def time(self, a_datetime):
+ self._before_event()
+ return self.super.time(a_datetime)
+
+
+class AutoTimingTestResultDecorator(HookedTestResultDecorator):
+ """Decorate a TestResult to add time events to a test run.
+
+ By default this will cause a time event before every test event,
+ but if explicit time data is being provided by the test run, then
+ this decorator will turn itself off to prevent causing confusion.
+ """
+
+ def __init__(self, decorated):
+ self._time = None
+ super(AutoTimingTestResultDecorator, self).__init__(decorated)
+
+ def _before_event(self):
+ time = self._time
+ if time is not None:
+ return
+ time = datetime.datetime.utcnow().replace(tzinfo=iso8601.Utc())
+ self.decorated.time(time)
+
+ def progress(self, offset, whence):
+ return self.decorated.progress(offset, whence)
+
+ @property
+ def shouldStop(self):
+ return self.decorated.shouldStop
+
+ def time(self, a_datetime):
+ """Provide a timestamp for the current test activity.
+
+ :param a_datetime: If None, automatically add timestamps before every
+ event (this is the default behaviour if time() is not called at
+ all). If not None, pass the provided time onto the decorated
+ result object and disable automatic timestamps.
+ """
+ self._time = a_datetime
+ return self.decorated.time(a_datetime)
+
+
+class TestResultFilter(TestResultDecorator):
+ """A pyunit TestResult interface implementation which filters tests.
+
+ Tests that pass the filter are handed on to another TestResult instance
+ for further processing/reporting. To obtain the filtered results,
+ the other instance must be interrogated.
+
+ :ivar result: The result that tests are passed to after filtering.
+ :ivar filter_predicate: The callback run to decide whether to pass
+ a result.
+ """
+
+ def __init__(self, result, filter_error=False, filter_failure=False,
+ filter_success=True, filter_skip=False,
+ filter_predicate=None):
+ """Create a FilterResult object filtering to result.
+
+ :param filter_error: Filter out errors.
+ :param filter_failure: Filter out failures.
+ :param filter_success: Filter out successful tests.
+ :param filter_skip: Filter out skipped tests.
+ :param filter_predicate: A callable taking (test, outcome, err,
+ details) and returning True if the result should be passed
+ through. err and details may be none if no error or extra
+ metadata is available. outcome is the name of the outcome such
+ as 'success' or 'failure'.
+ """
+ TestResultDecorator.__init__(self, result)
+ self._filter_error = filter_error
+ self._filter_failure = filter_failure
+ self._filter_success = filter_success
+ self._filter_skip = filter_skip
+ if filter_predicate is None:
+ filter_predicate = lambda test, outcome, err, details: True
+ self.filter_predicate = filter_predicate
+ # The current test (for filtering tags)
+ self._current_test = None
+ # Has the current test been filtered (for outputting test tags)
+ self._current_test_filtered = None
+ # The (new, gone) tags for the current test.
+ self._current_test_tags = None
+
+ def addError(self, test, err=None, details=None):
+ if (not self._filter_error and
+ self.filter_predicate(test, 'error', err, details)):
+ self.decorated.startTest(test)
+ self.decorated.addError(test, err, details=details)
+ else:
+ self._filtered()
+
+ def addFailure(self, test, err=None, details=None):
+ if (not self._filter_failure and
+ self.filter_predicate(test, 'failure', err, details)):
+ self.decorated.startTest(test)
+ self.decorated.addFailure(test, err, details=details)
+ else:
+ self._filtered()
+
+ def addSkip(self, test, reason=None, details=None):
+ if (not self._filter_skip and
+ self.filter_predicate(test, 'skip', reason, details)):
+ self.decorated.startTest(test)
+ self.decorated.addSkip(test, reason, details=details)
+ else:
+ self._filtered()
+
+ def addSuccess(self, test, details=None):
+ if (not self._filter_success and
+ self.filter_predicate(test, 'success', None, details)):
+ self.decorated.startTest(test)
+ self.decorated.addSuccess(test, details=details)
+ else:
+ self._filtered()
+
+ def addExpectedFailure(self, test, err=None, details=None):
+ if self.filter_predicate(test, 'expectedfailure', err, details):
+ self.decorated.startTest(test)
+ return self.decorated.addExpectedFailure(test, err,
+ details=details)
+ else:
+ self._filtered()
+
+ def addUnexpectedSuccess(self, test, details=None):
+ self.decorated.startTest(test)
+ return self.decorated.addUnexpectedSuccess(test, details=details)
+
+ def _filtered(self):
+ self._current_test_filtered = True
+
+ def startTest(self, test):
+ """Start a test.
+
+ Not directly passed to the client, but used for handling of tags
+ correctly.
+ """
+ self._current_test = test
+ self._current_test_filtered = False
+ self._current_test_tags = set(), set()
+
+ def stopTest(self, test):
+ """Stop a test.
+
+ Not directly passed to the client, but used for handling of tags
+ correctly.
+ """
+ if not self._current_test_filtered:
+ # Tags to output for this test.
+ if self._current_test_tags[0] or self._current_test_tags[1]:
+ self.decorated.tags(*self._current_test_tags)
+ self.decorated.stopTest(test)
+ self._current_test = None
+ self._current_test_filtered = None
+ self._current_test_tags = None
+
+ def tags(self, new_tags, gone_tags):
+ """Handle tag instructions.
+
+ Adds and removes tags as appropriate. If a test is currently running,
+ tags are not affected for subsequent tests.
+
+ :param new_tags: Tags to add,
+ :param gone_tags: Tags to remove.
+ """
+ if self._current_test is not None:
+ # gather the tags until the test stops.
+ self._current_test_tags[0].update(new_tags)
+ self._current_test_tags[0].difference_update(gone_tags)
+ self._current_test_tags[1].update(gone_tags)
+ self._current_test_tags[1].difference_update(new_tags)
+ return self.decorated.tags(new_tags, gone_tags)
+
+ def id_to_orig_id(self, id):
+ if id.startswith("subunit.RemotedTestCase."):
+ return id[len("subunit.RemotedTestCase."):]
+ return id
diff --git a/lib/subunit/python/subunit/tests/__init__.py b/lib/subunit/python/subunit/tests/__init__.py
index 544d0e704f..a78cec8572 100644
--- a/lib/subunit/python/subunit/tests/__init__.py
+++ b/lib/subunit/python/subunit/tests/__init__.py
@@ -2,24 +2,40 @@
# subunit: extensions to python unittest to get test results from subprocesses.
# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation; either version 2 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+# 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 subunit.tests import TestUtil, test_test_protocol
+from subunit.tests import (
+ TestUtil,
+ test_chunked,
+ test_details,
+ test_progress_model,
+ test_subunit_filter,
+ test_subunit_stats,
+ test_subunit_tags,
+ test_tap2subunit,
+ test_test_protocol,
+ test_test_results,
+ )
def test_suite():
result = TestUtil.TestSuite()
+ result.addTest(test_chunked.test_suite())
+ result.addTest(test_details.test_suite())
+ result.addTest(test_progress_model.test_suite())
+ result.addTest(test_test_results.test_suite())
result.addTest(test_test_protocol.test_suite())
+ result.addTest(test_tap2subunit.test_suite())
+ result.addTest(test_subunit_filter.test_suite())
+ result.addTest(test_subunit_tags.test_suite())
+ result.addTest(test_subunit_stats.test_suite())
return result
diff --git a/lib/subunit/python/subunit/tests/sample-script.py b/lib/subunit/python/subunit/tests/sample-script.py
index 223d2f5d9f..0ee019ae4a 100755
--- a/lib/subunit/python/subunit/tests/sample-script.py
+++ b/lib/subunit/python/subunit/tests/sample-script.py
@@ -1,5 +1,12 @@
#!/usr/bin/env python
import sys
+if len(sys.argv) == 2:
+ # subunit.tests.test_test_protocol.TestExecTestCase.test_sample_method_args
+ # uses this code path to be sure that the arguments were passed to
+ # sample-script.py
+ print "test fail"
+ print "error fail"
+ sys.exit(0)
print "test old mcdonald"
print "success old mcdonald"
print "test bing crosby"
diff --git a/lib/subunit/python/subunit/tests/test_chunked.py b/lib/subunit/python/subunit/tests/test_chunked.py
new file mode 100644
index 0000000000..a24e31e0c2
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_chunked.py
@@ -0,0 +1,127 @@
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+#
+# 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 cStringIO import StringIO
+import unittest
+
+import subunit.chunked
+
+
+def test_suite():
+ loader = subunit.tests.TestUtil.TestLoader()
+ result = loader.loadTestsFromName(__name__)
+ return result
+
+
+class TestDecode(unittest.TestCase):
+
+ def setUp(self):
+ unittest.TestCase.setUp(self)
+ self.output = StringIO()
+ self.decoder = subunit.chunked.Decoder(self.output)
+
+ def test_close_read_length_short_errors(self):
+ self.assertRaises(ValueError, self.decoder.close)
+
+ def test_close_body_short_errors(self):
+ self.assertEqual(None, self.decoder.write('2\r\na'))
+ self.assertRaises(ValueError, self.decoder.close)
+
+ def test_close_body_buffered_data_errors(self):
+ self.assertEqual(None, self.decoder.write('2\r'))
+ self.assertRaises(ValueError, self.decoder.close)
+
+ def test_close_after_finished_stream_safe(self):
+ self.assertEqual(None, self.decoder.write('2\r\nab'))
+ self.assertEqual('', self.decoder.write('0\r\n'))
+ self.decoder.close()
+
+ def test_decode_nothing(self):
+ self.assertEqual('', self.decoder.write('0\r\n'))
+ self.assertEqual('', self.output.getvalue())
+
+ def test_decode_serialised_form(self):
+ self.assertEqual(None, self.decoder.write("F\r\n"))
+ self.assertEqual(None, self.decoder.write("serialised\n"))
+ self.assertEqual('', self.decoder.write("form0\r\n"))
+
+ def test_decode_short(self):
+ self.assertEqual('', self.decoder.write('3\r\nabc0\r\n'))
+ self.assertEqual('abc', self.output.getvalue())
+
+ def test_decode_combines_short(self):
+ self.assertEqual('', self.decoder.write('6\r\nabcdef0\r\n'))
+ self.assertEqual('abcdef', self.output.getvalue())
+
+ def test_decode_excess_bytes_from_write(self):
+ self.assertEqual('1234', self.decoder.write('3\r\nabc0\r\n1234'))
+ self.assertEqual('abc', self.output.getvalue())
+
+ def test_decode_write_after_finished_errors(self):
+ self.assertEqual('1234', self.decoder.write('3\r\nabc0\r\n1234'))
+ self.assertRaises(ValueError, self.decoder.write, '')
+
+ def test_decode_hex(self):
+ self.assertEqual('', self.decoder.write('A\r\n12345678900\r\n'))
+ self.assertEqual('1234567890', self.output.getvalue())
+
+ def test_decode_long_ranges(self):
+ self.assertEqual(None, self.decoder.write('10000\r\n'))
+ self.assertEqual(None, self.decoder.write('1' * 65536))
+ self.assertEqual(None, self.decoder.write('10000\r\n'))
+ self.assertEqual(None, self.decoder.write('2' * 65536))
+ self.assertEqual('', self.decoder.write('0\r\n'))
+ self.assertEqual('1' * 65536 + '2' * 65536, self.output.getvalue())
+
+
+class TestEncode(unittest.TestCase):
+
+ def setUp(self):
+ unittest.TestCase.setUp(self)
+ self.output = StringIO()
+ self.encoder = subunit.chunked.Encoder(self.output)
+
+ def test_encode_nothing(self):
+ self.encoder.close()
+ self.assertEqual('0\r\n', self.output.getvalue())
+
+ def test_encode_empty(self):
+ self.encoder.write('')
+ self.encoder.close()
+ self.assertEqual('0\r\n', self.output.getvalue())
+
+ def test_encode_short(self):
+ self.encoder.write('abc')
+ self.encoder.close()
+ self.assertEqual('3\r\nabc0\r\n', self.output.getvalue())
+
+ def test_encode_combines_short(self):
+ self.encoder.write('abc')
+ self.encoder.write('def')
+ self.encoder.close()
+ self.assertEqual('6\r\nabcdef0\r\n', self.output.getvalue())
+
+ def test_encode_over_9_is_in_hex(self):
+ self.encoder.write('1234567890')
+ self.encoder.close()
+ self.assertEqual('A\r\n12345678900\r\n', self.output.getvalue())
+
+ def test_encode_long_ranges_not_combined(self):
+ self.encoder.write('1' * 65536)
+ self.encoder.write('2' * 65536)
+ self.encoder.close()
+ self.assertEqual('10000\r\n' + '1' * 65536 + '10000\r\n' +
+ '2' * 65536 + '0\r\n', self.output.getvalue())
diff --git a/lib/subunit/python/subunit/tests/test_details.py b/lib/subunit/python/subunit/tests/test_details.py
new file mode 100644
index 0000000000..2700d4afc7
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_details.py
@@ -0,0 +1,110 @@
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+#
+# 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 cStringIO import StringIO
+import unittest
+
+import subunit.tests
+from subunit import content, content_type, details
+
+
+def test_suite():
+ loader = subunit.tests.TestUtil.TestLoader()
+ result = loader.loadTestsFromName(__name__)
+ return result
+
+
+class TestSimpleDetails(unittest.TestCase):
+
+ def test_lineReceived(self):
+ parser = details.SimpleDetailsParser(None)
+ parser.lineReceived("foo\n")
+ parser.lineReceived("bar\n")
+ self.assertEqual("foo\nbar\n", parser._message)
+
+ def test_lineReceived_escaped_bracket(self):
+ parser = details.SimpleDetailsParser(None)
+ parser.lineReceived("foo\n")
+ parser.lineReceived(" ]are\n")
+ parser.lineReceived("bar\n")
+ self.assertEqual("foo\n]are\nbar\n", parser._message)
+
+ def test_get_message(self):
+ parser = details.SimpleDetailsParser(None)
+ self.assertEqual("", parser.get_message())
+
+ def test_get_details(self):
+ parser = details.SimpleDetailsParser(None)
+ traceback = ""
+ expected = {}
+ expected['traceback'] = content.Content(
+ content_type.ContentType("text", "x-traceback"),
+ lambda:[""])
+ found = parser.get_details()
+ self.assertEqual(expected.keys(), found.keys())
+ self.assertEqual(expected['traceback'].content_type,
+ found['traceback'].content_type)
+ self.assertEqual(''.join(expected['traceback'].iter_bytes()),
+ ''.join(found['traceback'].iter_bytes()))
+
+ def test_get_details_skip(self):
+ parser = details.SimpleDetailsParser(None)
+ traceback = ""
+ expected = {}
+ expected['reason'] = content.Content(
+ content_type.ContentType("text", "plain"),
+ lambda:[""])
+ found = parser.get_details("skip")
+ self.assertEqual(expected, found)
+
+ def test_get_details_success(self):
+ parser = details.SimpleDetailsParser(None)
+ traceback = ""
+ expected = {}
+ expected['message'] = content.Content(
+ content_type.ContentType("text", "plain"),
+ lambda:[""])
+ found = parser.get_details("success")
+ self.assertEqual(expected, found)
+
+
+class TestMultipartDetails(unittest.TestCase):
+
+ def test_get_message_is_None(self):
+ parser = details.MultipartDetailsParser(None)
+ self.assertEqual(None, parser.get_message())
+
+ def test_get_details(self):
+ parser = details.MultipartDetailsParser(None)
+ self.assertEqual({}, parser.get_details())
+
+ def test_parts(self):
+ parser = details.MultipartDetailsParser(None)
+ parser.lineReceived("Content-Type: text/plain\n")
+ parser.lineReceived("something\n")
+ parser.lineReceived("F\r\n")
+ parser.lineReceived("serialised\n")
+ parser.lineReceived("form0\r\n")
+ expected = {}
+ expected['something'] = content.Content(
+ content_type.ContentType("text", "plain"),
+ lambda:["serialised\nform"])
+ found = parser.get_details()
+ self.assertEqual(expected.keys(), found.keys())
+ self.assertEqual(expected['something'].content_type,
+ found['something'].content_type)
+ self.assertEqual(''.join(expected['something'].iter_bytes()),
+ ''.join(found['something'].iter_bytes()))
diff --git a/lib/subunit/python/subunit/tests/test_progress_model.py b/lib/subunit/python/subunit/tests/test_progress_model.py
new file mode 100644
index 0000000000..76200c6107
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_progress_model.py
@@ -0,0 +1,118 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
+#
+# 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 unittest
+
+import subunit
+from subunit.progress_model import ProgressModel
+
+
+class TestProgressModel(unittest.TestCase):
+
+ def assertProgressSummary(self, pos, total, progress):
+ """Assert that a progress model has reached a particular point."""
+ self.assertEqual(pos, progress.pos())
+ self.assertEqual(total, progress.width())
+
+ def test_new_progress_0_0(self):
+ progress = ProgressModel()
+ self.assertProgressSummary(0, 0, progress)
+
+ def test_advance_0_0(self):
+ progress = ProgressModel()
+ progress.advance()
+ self.assertProgressSummary(1, 0, progress)
+
+ def test_advance_1_0(self):
+ progress = ProgressModel()
+ progress.advance()
+ self.assertProgressSummary(1, 0, progress)
+
+ def test_set_width_absolute(self):
+ progress = ProgressModel()
+ progress.set_width(10)
+ self.assertProgressSummary(0, 10, progress)
+
+ def test_set_width_absolute_preserves_pos(self):
+ progress = ProgressModel()
+ progress.advance()
+ progress.set_width(2)
+ self.assertProgressSummary(1, 2, progress)
+
+ def test_adjust_width(self):
+ progress = ProgressModel()
+ progress.adjust_width(10)
+ self.assertProgressSummary(0, 10, progress)
+ progress.adjust_width(-10)
+ self.assertProgressSummary(0, 0, progress)
+
+ def test_adjust_width_preserves_pos(self):
+ progress = ProgressModel()
+ progress.advance()
+ progress.adjust_width(10)
+ self.assertProgressSummary(1, 10, progress)
+ progress.adjust_width(-10)
+ self.assertProgressSummary(1, 0, progress)
+
+ def test_push_preserves_progress(self):
+ progress = ProgressModel()
+ progress.adjust_width(3)
+ progress.advance()
+ progress.push()
+ self.assertProgressSummary(1, 3, progress)
+
+ def test_advance_advances_substack(self):
+ progress = ProgressModel()
+ progress.adjust_width(3)
+ progress.advance()
+ progress.push()
+ progress.adjust_width(1)
+ progress.advance()
+ self.assertProgressSummary(2, 3, progress)
+
+ def test_adjust_width_adjusts_substack(self):
+ progress = ProgressModel()
+ progress.adjust_width(3)
+ progress.advance()
+ progress.push()
+ progress.adjust_width(2)
+ progress.advance()
+ self.assertProgressSummary(3, 6, progress)
+
+ def test_set_width_adjusts_substack(self):
+ progress = ProgressModel()
+ progress.adjust_width(3)
+ progress.advance()
+ progress.push()
+ progress.set_width(2)
+ progress.advance()
+ self.assertProgressSummary(3, 6, progress)
+
+ def test_pop_restores_progress(self):
+ progress = ProgressModel()
+ progress.adjust_width(3)
+ progress.advance()
+ progress.push()
+ progress.adjust_width(1)
+ progress.advance()
+ progress.pop()
+ self.assertProgressSummary(1, 3, progress)
+
+
+def test_suite():
+ loader = subunit.tests.TestUtil.TestLoader()
+ result = loader.loadTestsFromName(__name__)
+ return result
diff --git a/lib/subunit/python/subunit/tests/test_subunit_filter.py b/lib/subunit/python/subunit/tests/test_subunit_filter.py
new file mode 100644
index 0000000000..3c65ed3afc
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_subunit_filter.py
@@ -0,0 +1,136 @@
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""Tests for subunit.TestResultFilter."""
+
+import unittest
+from StringIO import StringIO
+
+import subunit
+from subunit.test_results import TestResultFilter
+
+
+class TestTestResultFilter(unittest.TestCase):
+ """Test for TestResultFilter, a TestResult object which filters tests."""
+
+ def _setUp(self):
+ self.output = StringIO()
+
+ def test_default(self):
+ """The default is to exclude success and include everything else."""
+ self.filtered_result = unittest.TestResult()
+ self.filter = TestResultFilter(self.filtered_result)
+ self.run_tests()
+ # skips are seen as success by default python TestResult.
+ self.assertEqual(['error'],
+ [error[0].id() for error in self.filtered_result.errors])
+ self.assertEqual(['failed'],
+ [failure[0].id() for failure in
+ self.filtered_result.failures])
+ self.assertEqual(4, self.filtered_result.testsRun)
+
+ def test_exclude_errors(self):
+ self.filtered_result = unittest.TestResult()
+ self.filter = TestResultFilter(self.filtered_result,
+ filter_error=True)
+ self.run_tests()
+ # skips are seen as errors by default python TestResult.
+ self.assertEqual([], self.filtered_result.errors)
+ self.assertEqual(['failed'],
+ [failure[0].id() for failure in
+ self.filtered_result.failures])
+ self.assertEqual(3, self.filtered_result.testsRun)
+
+ def test_exclude_failure(self):
+ self.filtered_result = unittest.TestResult()
+ self.filter = TestResultFilter(self.filtered_result,
+ filter_failure=True)
+ self.run_tests()
+ self.assertEqual(['error'],
+ [error[0].id() for error in self.filtered_result.errors])
+ self.assertEqual([],
+ [failure[0].id() for failure in
+ self.filtered_result.failures])
+ self.assertEqual(3, self.filtered_result.testsRun)
+
+ def test_exclude_skips(self):
+ self.filtered_result = subunit.TestResultStats(None)
+ self.filter = TestResultFilter(self.filtered_result,
+ filter_skip=True)
+ self.run_tests()
+ self.assertEqual(0, self.filtered_result.skipped_tests)
+ self.assertEqual(2, self.filtered_result.failed_tests)
+ self.assertEqual(3, self.filtered_result.testsRun)
+
+ def test_include_success(self):
+ """Success's can be included if requested."""
+ self.filtered_result = unittest.TestResult()
+ self.filter = TestResultFilter(self.filtered_result,
+ filter_success=False)
+ self.run_tests()
+ self.assertEqual(['error'],
+ [error[0].id() for error in self.filtered_result.errors])
+ self.assertEqual(['failed'],
+ [failure[0].id() for failure in
+ self.filtered_result.failures])
+ self.assertEqual(5, self.filtered_result.testsRun)
+
+ def test_filter_predicate(self):
+ """You can filter by predicate callbacks"""
+ self.filtered_result = unittest.TestResult()
+ def filter_cb(test, outcome, err, details):
+ return outcome == 'success'
+ self.filter = TestResultFilter(self.filtered_result,
+ filter_predicate=filter_cb,
+ filter_success=False)
+ self.run_tests()
+ # Only success should pass
+ self.assertEqual(1, self.filtered_result.testsRun)
+
+ def run_tests(self):
+ self.setUpTestStream()
+ self.test = subunit.ProtocolTestCase(self.input_stream)
+ self.test.run(self.filter)
+
+ def setUpTestStream(self):
+ # While TestResultFilter works on python objects, using a subunit
+ # stream is an easy pithy way of getting a series of test objects to
+ # call into the TestResult, and as TestResultFilter is intended for
+ # use with subunit also has the benefit of detecting any interface
+ # skew issues.
+ self.input_stream = StringIO()
+ self.input_stream.write("""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
+""")
+ self.input_stream.seek(0)
+
+
+def test_suite():
+ loader = subunit.tests.TestUtil.TestLoader()
+ result = loader.loadTestsFromName(__name__)
+ return result
diff --git a/lib/subunit/python/subunit/tests/test_subunit_stats.py b/lib/subunit/python/subunit/tests/test_subunit_stats.py
new file mode 100644
index 0000000000..a7f8fca675
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_subunit_stats.py
@@ -0,0 +1,83 @@
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""Tests for subunit.TestResultStats."""
+
+import unittest
+from StringIO import StringIO
+
+import subunit
+
+
+class TestTestResultStats(unittest.TestCase):
+ """Test for TestResultStats, a TestResult object that generates stats."""
+
+ def setUp(self):
+ self.output = StringIO()
+ self.result = subunit.TestResultStats(self.output)
+ self.input_stream = StringIO()
+ self.test = subunit.ProtocolTestCase(self.input_stream)
+
+ def test_stats_empty(self):
+ self.test.run(self.result)
+ self.assertEqual(0, self.result.total_tests)
+ self.assertEqual(0, self.result.passed_tests)
+ self.assertEqual(0, self.result.failed_tests)
+ self.assertEqual(set(), self.result.seen_tags)
+
+ def setUpUsedStream(self):
+ self.input_stream.write("""tags: global
+test passed
+success passed
+test failed
+tags: local
+failure failed
+test error
+error error
+test skipped
+skip skipped
+test todo
+xfail todo
+""")
+ self.input_stream.seek(0)
+ self.test.run(self.result)
+
+ def test_stats_smoke_everything(self):
+ # Statistics are calculated usefully.
+ self.setUpUsedStream()
+ self.assertEqual(5, self.result.total_tests)
+ self.assertEqual(2, self.result.passed_tests)
+ self.assertEqual(2, self.result.failed_tests)
+ self.assertEqual(1, self.result.skipped_tests)
+ self.assertEqual(set(["global", "local"]), self.result.seen_tags)
+
+ def test_stat_formatting(self):
+ expected = ("""
+Total tests: 5
+Passed tests: 2
+Failed tests: 2
+Skipped tests: 1
+Seen tags: global, local
+""")[1:]
+ self.setUpUsedStream()
+ self.result.formatStats()
+ self.assertEqual(expected, self.output.getvalue())
+
+
+def test_suite():
+ loader = subunit.tests.TestUtil.TestLoader()
+ result = loader.loadTestsFromName(__name__)
+ return result
diff --git a/lib/subunit/python/subunit/tests/test_subunit_tags.py b/lib/subunit/python/subunit/tests/test_subunit_tags.py
new file mode 100644
index 0000000000..227e2b7475
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_subunit_tags.py
@@ -0,0 +1,68 @@
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""Tests for subunit.tag_stream."""
+
+import unittest
+from StringIO import StringIO
+
+import subunit
+import subunit.test_results
+
+
+class TestSubUnitTags(unittest.TestCase):
+
+ def setUp(self):
+ self.original = StringIO()
+ self.filtered = StringIO()
+
+ 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")
+ 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())
+
+ 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")
+ 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())
+
+
+def test_suite():
+ loader = subunit.tests.TestUtil.TestLoader()
+ result = loader.loadTestsFromName(__name__)
+ return result
diff --git a/lib/subunit/python/subunit/tests/test_tap2subunit.py b/lib/subunit/python/subunit/tests/test_tap2subunit.py
new file mode 100644
index 0000000000..febfe9d441
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_tap2subunit.py
@@ -0,0 +1,432 @@
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+#
+# 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.
+#
+
+"""Tests for TAP2SubUnit."""
+
+import unittest
+from StringIO import StringIO
+import os
+import subunit
+import sys
+
+
+class TestTAP2SubUnit(unittest.TestCase):
+ """Tests for TAP2SubUnit.
+
+ These tests test TAP string data in, and subunit string data out.
+ This is ok because the subunit protocol is intended to be stable,
+ but it might be easier/pithier to write tests against TAP string in,
+ parsed subunit objects out (by hooking the subunit stream to a subunit
+ protocol server.
+ """
+
+ def setUp(self):
+ self.tap = StringIO()
+ self.subunit = StringIO()
+
+ 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.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())
+
+ def test_ok_test_pass(self):
+ # A file
+ # ok
+ # 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.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())
+
+ 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.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())
+
+ 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.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())
+
+ 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.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())
+
+ 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.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())
+
+ 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.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())
+
+ 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.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())
+
+ 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.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())
+
+ 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.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())
+
+ 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.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())
+
+ def test_missing_test_at_end_with_plan_adds_error(self):
+ # A file
+ # 1..3
+ # 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.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())
+
+ def test_missing_test_with_plan_adds_error(self):
+ # A file
+ # 1..3
+ # 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.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())
+
+ 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.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())
+
+ def test_four_tests_in_a_row_trailing_plan(self):
+ # A file
+ # ok 1 - first test in a script with no plan at all
+ # not ok 2 - second
+ # ok 3 - third
+ # 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.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())
+
+ def test_four_tests_in_a_row_with_plan(self):
+ # A file
+ # 1..4
+ # ok 1 - first test in a script with no plan at all
+ # not ok 2 - second
+ # 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.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())
+
+ def test_four_tests_in_a_row_no_plan(self):
+ # A file
+ # ok 1 - first test in a script with no plan at all
+ # not ok 2 - second
+ # 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.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())
+
+ 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.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())
+
+ def test_leading_comments_add_to_next_test_log(self):
+ # A file
+ # # comment
+ # ok
+ # 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.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())
+
+ def test_trailing_comments_are_included_in_last_test_log(self):
+ # A file
+ # ok foo
+ # ok foo
+ # # 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.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())
+
+
+def test_suite():
+ loader = subunit.tests.TestUtil.TestLoader()
+ result = loader.loadTestsFromName(__name__)
+ return result
diff --git a/lib/subunit/python/subunit/tests/test_test_protocol.py b/lib/subunit/python/subunit/tests/test_test_protocol.py
index af31584a97..9e9db18163 100644
--- a/lib/subunit/python/subunit/tests/test_test_protocol.py
+++ b/lib/subunit/python/subunit/tests/test_test_protocol.py
@@ -1,126 +1,80 @@
#
-# subunit: extensions to python unittest to get test results from subprocesses.
+# subunit: extensions to Python unittest to get test results from subprocesses.
# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation; either version 2 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+# 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 datetime
import unittest
from StringIO import StringIO
import os
-import subunit
import sys
-try:
- class MockTestProtocolServerClient(object):
- """A mock protocol server client to test callbacks."""
-
- def __init__(self):
- self.end_calls = []
- self.error_calls = []
- self.failure_calls = []
- self.start_calls = []
- self.success_calls = []
- super(MockTestProtocolServerClient, self).__init__()
-
- def addError(self, test, error):
- self.error_calls.append((test, error))
-
- def addFailure(self, test, error):
- self.failure_calls.append((test, error))
-
- def addSuccess(self, test):
- self.success_calls.append(test)
-
- def stopTest(self, test):
- self.end_calls.append(test)
-
- def startTest(self, test):
- self.start_calls.append(test)
-
-except AttributeError:
- MockTestProtocolServer = None
-
-
-class TestMockTestProtocolServer(unittest.TestCase):
-
- def test_start_test(self):
- protocol = MockTestProtocolServerClient()
- protocol.startTest(subunit.RemotedTestCase("test old mcdonald"))
- self.assertEqual(protocol.start_calls,
- [subunit.RemotedTestCase("test old mcdonald")])
- self.assertEqual(protocol.end_calls, [])
- self.assertEqual(protocol.error_calls, [])
- self.assertEqual(protocol.failure_calls, [])
- self.assertEqual(protocol.success_calls, [])
-
- def test_add_error(self):
- protocol = MockTestProtocolServerClient()
- protocol.addError(subunit.RemotedTestCase("old mcdonald"),
- subunit.RemoteError("omg it works"))
- self.assertEqual(protocol.start_calls, [])
- self.assertEqual(protocol.end_calls, [])
- self.assertEqual(protocol.error_calls, [(
- subunit.RemotedTestCase("old mcdonald"),
- subunit.RemoteError("omg it works"))])
- self.assertEqual(protocol.failure_calls, [])
- self.assertEqual(protocol.success_calls, [])
-
- def test_add_failure(self):
- protocol = MockTestProtocolServerClient()
- protocol.addFailure(subunit.RemotedTestCase("old mcdonald"),
- subunit.RemoteError("omg it works"))
- self.assertEqual(protocol.start_calls, [])
- self.assertEqual(protocol.end_calls, [])
- self.assertEqual(protocol.error_calls, [])
- self.assertEqual(protocol.failure_calls, [
- (subunit.RemotedTestCase("old mcdonald"),
- subunit.RemoteError("omg it works"))])
- self.assertEqual(protocol.success_calls, [])
+from testtools.content import Content, TracebackContent
+from testtools.content_type import ContentType
+from testtools.tests.helpers import (
+ Python26TestResult,
+ Python27TestResult,
+ ExtendedTestResult,
+ )
- def test_add_success(self):
- protocol = MockTestProtocolServerClient()
- protocol.addSuccess(subunit.RemotedTestCase("test old mcdonald"))
- self.assertEqual(protocol.start_calls, [])
- self.assertEqual(protocol.end_calls, [])
- self.assertEqual(protocol.error_calls, [])
- self.assertEqual(protocol.failure_calls, [])
- self.assertEqual(protocol.success_calls,
- [subunit.RemotedTestCase("test old mcdonald")])
-
- def test_end_test(self):
- protocol = MockTestProtocolServerClient()
- protocol.stopTest(subunit.RemotedTestCase("test old mcdonald"))
- self.assertEqual(protocol.end_calls,
- [subunit.RemotedTestCase("test old mcdonald")])
- self.assertEqual(protocol.error_calls, [])
- self.assertEqual(protocol.failure_calls, [])
- self.assertEqual(protocol.success_calls, [])
- self.assertEqual(protocol.start_calls, [])
+import subunit
+from subunit import _remote_exception_str
+import subunit.iso8601 as iso8601
class TestTestImports(unittest.TestCase):
def test_imports(self):
+ from subunit import DiscardStream
from subunit import TestProtocolServer
from subunit import RemotedTestCase
from subunit import RemoteError
from subunit import ExecTestCase
from subunit import IsolatedTestCase
from subunit import TestProtocolClient
+ from subunit import ProtocolTestCase
+
+
+class TestDiscardStream(unittest.TestCase):
+
+ def test_write(self):
+ subunit.DiscardStream().write("content")
+
+
+class TestProtocolServerForward(unittest.TestCase):
+
+ def test_story(self):
+ client = unittest.TestResult()
+ out = StringIO()
+ protocol = subunit.TestProtocolServer(client, forward_stream=out)
+ pipe = StringIO("test old mcdonald\n"
+ "success old mcdonald\n")
+ protocol.readFrom(pipe)
+ mcdonald = subunit.RemotedTestCase("old mcdonald")
+ self.assertEqual(client.testsRun, 1)
+ self.assertEqual(pipe.getvalue(), out.getvalue())
+ def test_not_command(self):
+ client = unittest.TestResult()
+ out = StringIO()
+ protocol = subunit.TestProtocolServer(client,
+ stream=subunit.DiscardStream(), forward_stream=out)
+ pipe = StringIO("success old mcdonald\n")
+ protocol.readFrom(pipe)
+ self.assertEqual(client.testsRun, 0)
+ self.assertEqual("", out.getvalue())
+
class TestTestProtocolServerPipe(unittest.TestCase):
@@ -140,47 +94,48 @@ class TestTestProtocolServerPipe(unittest.TestCase):
bing = subunit.RemotedTestCase("bing crosby")
an_error = subunit.RemotedTestCase("an error")
self.assertEqual(client.errors,
- [(an_error, 'RemoteException: \n\n')])
+ [(an_error, _remote_exception_str + '\n')])
self.assertEqual(
client.failures,
- [(bing, "RemoteException: foo.c:53:ERROR invalid state\n\n")])
+ [(bing, _remote_exception_str + ": Text attachment: traceback\n"
+ "------------\nfoo.c:53:ERROR invalid state\n"
+ "------------\n\n")])
self.assertEqual(client.testsRun, 3)
class TestTestProtocolServerStartTest(unittest.TestCase):
def setUp(self):
- self.client = MockTestProtocolServerClient()
+ self.client = Python26TestResult()
self.protocol = subunit.TestProtocolServer(self.client)
def test_start_test(self):
self.protocol.lineReceived("test old mcdonald\n")
- self.assertEqual(self.client.start_calls,
- [subunit.RemotedTestCase("old mcdonald")])
+ self.assertEqual(self.client._events,
+ [('startTest', subunit.RemotedTestCase("old mcdonald"))])
def test_start_testing(self):
self.protocol.lineReceived("testing old mcdonald\n")
- self.assertEqual(self.client.start_calls,
- [subunit.RemotedTestCase("old mcdonald")])
+ self.assertEqual(self.client._events,
+ [('startTest', subunit.RemotedTestCase("old mcdonald"))])
def test_start_test_colon(self):
self.protocol.lineReceived("test: old mcdonald\n")
- self.assertEqual(self.client.start_calls,
- [subunit.RemotedTestCase("old mcdonald")])
+ self.assertEqual(self.client._events,
+ [('startTest', subunit.RemotedTestCase("old mcdonald"))])
def test_start_testing_colon(self):
self.protocol.lineReceived("testing: old mcdonald\n")
- self.assertEqual(self.client.start_calls,
- [subunit.RemotedTestCase("old mcdonald")])
+ self.assertEqual(self.client._events,
+ [('startTest', subunit.RemotedTestCase("old mcdonald"))])
class TestTestProtocolServerPassThrough(unittest.TestCase):
def setUp(self):
- from StringIO import StringIO
self.stdout = StringIO()
self.test = subunit.RemotedTestCase("old mcdonald")
- self.client = MockTestProtocolServerClient()
+ self.client = ExtendedTestResult()
self.protocol = subunit.TestProtocolServer(self.client, self.stdout)
def keywords_before_test(self):
@@ -205,42 +160,37 @@ class TestTestProtocolServerPassThrough(unittest.TestCase):
def test_keywords_before_test(self):
self.keywords_before_test()
- self.assertEqual(self.client.start_calls, [])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.success_calls, [])
+ self.assertEqual(self.client._events, [])
def test_keywords_after_error(self):
self.protocol.lineReceived("test old mcdonald\n")
self.protocol.lineReceived("error old mcdonald\n")
self.keywords_before_test()
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls,
- [(self.test, subunit.RemoteError(""))])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.success_calls, [])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addError', self.test, {}),
+ ('stopTest', self.test),
+ ], self.client._events)
def test_keywords_after_failure(self):
self.protocol.lineReceived("test old mcdonald\n")
self.protocol.lineReceived("failure old mcdonald\n")
self.keywords_before_test()
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls,
- [(self.test, subunit.RemoteError())])
- self.assertEqual(self.client.success_calls, [])
+ self.assertEqual(self.client._events, [
+ ('startTest', self.test),
+ ('addFailure', self.test, {}),
+ ('stopTest', self.test),
+ ])
def test_keywords_after_success(self):
self.protocol.lineReceived("test old mcdonald\n")
self.protocol.lineReceived("success old mcdonald\n")
self.keywords_before_test()
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.success_calls, [self.test])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addSuccess', self.test),
+ ('stopTest', self.test),
+ ], self.client._events)
def test_keywords_after_test(self):
self.protocol.lineReceived("test old mcdonald\n")
@@ -265,14 +215,15 @@ class TestTestProtocolServerPassThrough(unittest.TestCase):
"successful a\n"
"successful: a\n"
"]\n")
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.failure_calls,
- [(self.test, subunit.RemoteError())])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.success_calls, [])
+ self.assertEqual(self.client._events, [
+ ('startTest', self.test),
+ ('addFailure', self.test, {}),
+ ('stopTest', self.test),
+ ])
def test_keywords_during_failure(self):
+ # A smoke test to make sure that the details parsers have control
+ # appropriately.
self.protocol.lineReceived("test old mcdonald\n")
self.protocol.lineReceived("failure: old mcdonald [\n")
self.protocol.lineReceived("test old mcdonald\n")
@@ -287,21 +238,24 @@ class TestTestProtocolServerPassThrough(unittest.TestCase):
self.protocol.lineReceived(" ]\n")
self.protocol.lineReceived("]\n")
self.assertEqual(self.stdout.getvalue(), "")
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.failure_calls,
- [(self.test, subunit.RemoteError("test old mcdonald\n"
- "failure a\n"
- "failure: a\n"
- "error a\n"
- "error: a\n"
- "success a\n"
- "success: a\n"
- "successful a\n"
- "successful: a\n"
- "]\n"))])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.success_calls, [])
+ details = {}
+ details['traceback'] = Content(ContentType("text", "x-traceback"),
+ lambda:[
+ "test old mcdonald\n"
+ "failure a\n"
+ "failure: a\n"
+ "error a\n"
+ "error: a\n"
+ "success a\n"
+ "success: a\n"
+ "successful a\n"
+ "successful: a\n"
+ "]\n"])
+ self.assertEqual(self.client._events, [
+ ('startTest', self.test),
+ ('addFailure', self.test, details),
+ ('stopTest', self.test),
+ ])
def test_stdout_passthrough(self):
"""Lines received which cannot be interpreted as any protocol action
@@ -315,103 +269,133 @@ class TestTestProtocolServerPassThrough(unittest.TestCase):
class TestTestProtocolServerLostConnection(unittest.TestCase):
def setUp(self):
- self.client = MockTestProtocolServerClient()
+ self.client = Python26TestResult()
self.protocol = subunit.TestProtocolServer(self.client)
self.test = subunit.RemotedTestCase("old mcdonald")
def test_lost_connection_no_input(self):
self.protocol.lostConnection()
- self.assertEqual(self.client.start_calls, [])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.success_calls, [])
+ self.assertEqual([], self.client._events)
def test_lost_connection_after_start(self):
self.protocol.lineReceived("test old mcdonald\n")
self.protocol.lostConnection()
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [
- (self.test, subunit.RemoteError("lost connection during "
- "test 'old mcdonald'"))])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.success_calls, [])
+ failure = subunit.RemoteError(
+ "lost connection during test 'old mcdonald'")
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addError', self.test, failure),
+ ('stopTest', self.test),
+ ], self.client._events)
def test_lost_connected_after_error(self):
self.protocol.lineReceived("test old mcdonald\n")
self.protocol.lineReceived("error old mcdonald\n")
self.protocol.lostConnection()
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [
- (self.test, subunit.RemoteError(""))])
- self.assertEqual(self.client.success_calls, [])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addError', self.test, subunit.RemoteError("")),
+ ('stopTest', self.test),
+ ], self.client._events)
- def test_lost_connection_during_error(self):
+ def do_connection_lost(self, outcome, opening):
self.protocol.lineReceived("test old mcdonald\n")
- self.protocol.lineReceived("error old mcdonald [\n")
+ self.protocol.lineReceived("%s old mcdonald %s" % (outcome, opening))
self.protocol.lostConnection()
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [
- (self.test, subunit.RemoteError("lost connection during error "
- "report of test 'old mcdonald'"))])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.success_calls, [])
+ failure = subunit.RemoteError(
+ "lost connection during %s report of test 'old mcdonald'" %
+ outcome)
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addError', self.test, failure),
+ ('stopTest', self.test),
+ ], self.client._events)
+
+ def test_lost_connection_during_error(self):
+ self.do_connection_lost("error", "[\n")
+
+ def test_lost_connection_during_error_details(self):
+ self.do_connection_lost("error", "[ multipart\n")
def test_lost_connected_after_failure(self):
self.protocol.lineReceived("test old mcdonald\n")
self.protocol.lineReceived("failure old mcdonald\n")
self.protocol.lostConnection()
- test = subunit.RemotedTestCase("old mcdonald")
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls,
- [(self.test, subunit.RemoteError())])
- self.assertEqual(self.client.success_calls, [])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addFailure', self.test, subunit.RemoteError("")),
+ ('stopTest', self.test),
+ ], self.client._events)
def test_lost_connection_during_failure(self):
- self.protocol.lineReceived("test old mcdonald\n")
- self.protocol.lineReceived("failure old mcdonald [\n")
- self.protocol.lostConnection()
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls,
- [(self.test,
- subunit.RemoteError("lost connection during "
- "failure report"
- " of test 'old mcdonald'"))])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.success_calls, [])
+ self.do_connection_lost("failure", "[\n")
+
+ def test_lost_connection_during_failure_details(self):
+ self.do_connection_lost("failure", "[ multipart\n")
def test_lost_connection_after_success(self):
self.protocol.lineReceived("test old mcdonald\n")
self.protocol.lineReceived("success old mcdonald\n")
self.protocol.lostConnection()
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls, [])
- self.assertEqual(self.client.success_calls, [self.test])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addSuccess', self.test),
+ ('stopTest', self.test),
+ ], self.client._events)
+
+ def test_lost_connection_during_success(self):
+ self.do_connection_lost("success", "[\n")
+
+ def test_lost_connection_during_success_details(self):
+ self.do_connection_lost("success", "[ multipart\n")
+
+ def test_lost_connection_during_skip(self):
+ self.do_connection_lost("skip", "[\n")
+
+ def test_lost_connection_during_skip_details(self):
+ self.do_connection_lost("skip", "[ multipart\n")
+
+ def test_lost_connection_during_xfail(self):
+ self.do_connection_lost("xfail", "[\n")
+
+ def test_lost_connection_during_xfail_details(self):
+ self.do_connection_lost("xfail", "[ multipart\n")
+
+
+class TestInTestMultipart(unittest.TestCase):
+
+ def setUp(self):
+ self.client = ExtendedTestResult()
+ self.protocol = subunit.TestProtocolServer(self.client)
+ self.protocol.lineReceived("test mcdonalds farm\n")
+ self.test = subunit.RemotedTestCase("mcdonalds farm")
+
+ def test__outcome_sets_details_parser(self):
+ self.protocol._reading_success_details.details_parser = None
+ self.protocol._state._outcome(0, "mcdonalds farm [ multipart\n",
+ None, self.protocol._reading_success_details)
+ parser = self.protocol._reading_success_details.details_parser
+ self.assertNotEqual(None, parser)
+ self.assertTrue(isinstance(parser,
+ subunit.details.MultipartDetailsParser))
class TestTestProtocolServerAddError(unittest.TestCase):
def setUp(self):
- self.client = MockTestProtocolServerClient()
+ self.client = ExtendedTestResult()
self.protocol = subunit.TestProtocolServer(self.client)
self.protocol.lineReceived("test mcdonalds farm\n")
self.test = subunit.RemotedTestCase("mcdonalds farm")
def simple_error_keyword(self, keyword):
self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [
- (self.test, subunit.RemoteError(""))])
- self.assertEqual(self.client.failure_calls, [])
+ details = {}
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addError', self.test, details),
+ ('stopTest', self.test),
+ ], self.client._events)
def test_simple_error(self):
self.simple_error_keyword("error")
@@ -422,21 +406,27 @@ class TestTestProtocolServerAddError(unittest.TestCase):
def test_error_empty_message(self):
self.protocol.lineReceived("error mcdonalds farm [\n")
self.protocol.lineReceived("]\n")
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [
- (self.test, subunit.RemoteError(""))])
- self.assertEqual(self.client.failure_calls, [])
+ details = {}
+ details['traceback'] = Content(ContentType("text", "x-traceback"),
+ lambda:[""])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addError', self.test, details),
+ ('stopTest', self.test),
+ ], self.client._events)
def error_quoted_bracket(self, keyword):
self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
self.protocol.lineReceived(" ]\n")
self.protocol.lineReceived("]\n")
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [
- (self.test, subunit.RemoteError("]\n"))])
- self.assertEqual(self.client.failure_calls, [])
+ details = {}
+ details['traceback'] = Content(ContentType("text", "x-traceback"),
+ lambda:["]\n"])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addError', self.test, details),
+ ('stopTest', self.test),
+ ], self.client._events)
def test_error_quoted_bracket(self):
self.error_quoted_bracket("error")
@@ -448,18 +438,22 @@ class TestTestProtocolServerAddError(unittest.TestCase):
class TestTestProtocolServerAddFailure(unittest.TestCase):
def setUp(self):
- self.client = MockTestProtocolServerClient()
+ self.client = ExtendedTestResult()
self.protocol = subunit.TestProtocolServer(self.client)
self.protocol.lineReceived("test mcdonalds farm\n")
self.test = subunit.RemotedTestCase("mcdonalds farm")
+ def assertFailure(self, details):
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addFailure', self.test, details),
+ ('stopTest', self.test),
+ ], self.client._events)
+
def simple_failure_keyword(self, keyword):
self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls,
- [(self.test, subunit.RemoteError())])
+ details = {}
+ self.assertFailure(details)
def test_simple_failure(self):
self.simple_failure_keyword("failure")
@@ -470,21 +464,19 @@ class TestTestProtocolServerAddFailure(unittest.TestCase):
def test_failure_empty_message(self):
self.protocol.lineReceived("failure mcdonalds farm [\n")
self.protocol.lineReceived("]\n")
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls,
- [(self.test, subunit.RemoteError())])
+ details = {}
+ details['traceback'] = Content(ContentType("text", "x-traceback"),
+ lambda:[""])
+ self.assertFailure(details)
def failure_quoted_bracket(self, keyword):
self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
self.protocol.lineReceived(" ]\n")
self.protocol.lineReceived("]\n")
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.failure_calls,
- [(self.test, subunit.RemoteError("]\n"))])
+ details = {}
+ details['traceback'] = Content(ContentType("text", "x-traceback"),
+ lambda:["]\n"])
+ self.assertFailure(details)
def test_failure_quoted_bracket(self):
self.failure_quoted_bracket("failure")
@@ -493,20 +485,191 @@ class TestTestProtocolServerAddFailure(unittest.TestCase):
self.failure_quoted_bracket("failure:")
+class TestTestProtocolServerAddxFail(unittest.TestCase):
+ """Tests for the xfail keyword.
+
+ In Python this can thunk through to Success due to stdlib limitations (see
+ README).
+ """
+
+ def capture_expected_failure(self, test, err):
+ self._events.append((test, err))
+
+ def setup_python26(self):
+ """Setup a test object ready to be xfailed and thunk to success."""
+ self.client = Python26TestResult()
+ self.setup_protocol()
+
+ def setup_python27(self):
+ """Setup a test object ready to be xfailed."""
+ self.client = Python27TestResult()
+ self.setup_protocol()
+
+ def setup_python_ex(self):
+ """Setup a test object ready to be xfailed with details."""
+ self.client = ExtendedTestResult()
+ self.setup_protocol()
+
+ def setup_protocol(self):
+ """Setup the protocol based on self.client."""
+ self.protocol = subunit.TestProtocolServer(self.client)
+ self.protocol.lineReceived("test mcdonalds farm\n")
+ self.test = self.client._events[-1][-1]
+
+ def simple_xfail_keyword(self, keyword, as_success):
+ self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
+ self.check_success_or_xfail(as_success)
+
+ def check_success_or_xfail(self, as_success, error_message=None):
+ if as_success:
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addSuccess', self.test),
+ ('stopTest', self.test),
+ ], self.client._events)
+ else:
+ details = {}
+ if error_message is not None:
+ details['traceback'] = Content(
+ ContentType("text", "x-traceback"), lambda:[error_message])
+ if isinstance(self.client, ExtendedTestResult):
+ value = details
+ else:
+ if error_message is not None:
+ value = subunit.RemoteError('Text attachment: traceback\n'
+ '------------\n' + error_message + '------------\n')
+ else:
+ value = subunit.RemoteError()
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addExpectedFailure', self.test, value),
+ ('stopTest', self.test),
+ ], self.client._events)
+
+ def test_simple_xfail(self):
+ self.setup_python26()
+ self.simple_xfail_keyword("xfail", True)
+ self.setup_python27()
+ self.simple_xfail_keyword("xfail", False)
+ self.setup_python_ex()
+ self.simple_xfail_keyword("xfail", False)
+
+ def test_simple_xfail_colon(self):
+ self.setup_python26()
+ self.simple_xfail_keyword("xfail:", True)
+ self.setup_python27()
+ self.simple_xfail_keyword("xfail:", False)
+ self.setup_python_ex()
+ self.simple_xfail_keyword("xfail:", False)
+
+ def test_xfail_empty_message(self):
+ self.setup_python26()
+ self.empty_message(True)
+ self.setup_python27()
+ self.empty_message(False)
+ self.setup_python_ex()
+ self.empty_message(False, error_message="")
+
+ def empty_message(self, as_success, error_message="\n"):
+ self.protocol.lineReceived("xfail mcdonalds farm [\n")
+ self.protocol.lineReceived("]\n")
+ self.check_success_or_xfail(as_success, error_message)
+
+ def xfail_quoted_bracket(self, keyword, as_success):
+ # This tests it is accepted, but cannot test it is used today, because
+ # of not having a way to expose it in Python so far.
+ self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
+ self.protocol.lineReceived(" ]\n")
+ self.protocol.lineReceived("]\n")
+ self.check_success_or_xfail(as_success, "]\n")
+
+ def test_xfail_quoted_bracket(self):
+ self.setup_python26()
+ self.xfail_quoted_bracket("xfail", True)
+ self.setup_python27()
+ self.xfail_quoted_bracket("xfail", False)
+ self.setup_python_ex()
+ self.xfail_quoted_bracket("xfail", False)
+
+ def test_xfail_colon_quoted_bracket(self):
+ self.setup_python26()
+ self.xfail_quoted_bracket("xfail:", True)
+ self.setup_python27()
+ self.xfail_quoted_bracket("xfail:", False)
+ self.setup_python_ex()
+ self.xfail_quoted_bracket("xfail:", False)
+
+
+class TestTestProtocolServerAddSkip(unittest.TestCase):
+ """Tests for the skip keyword.
+
+ In Python this meets the testtools extended TestResult contract.
+ (See https://launchpad.net/testtools).
+ """
+
+ def setUp(self):
+ """Setup a test object ready to be skipped."""
+ self.client = ExtendedTestResult()
+ self.protocol = subunit.TestProtocolServer(self.client)
+ self.protocol.lineReceived("test mcdonalds farm\n")
+ self.test = self.client._events[-1][-1]
+
+ def assertSkip(self, reason):
+ details = {}
+ if reason is not None:
+ details['reason'] = Content(
+ ContentType("text", "plain"), lambda:[reason])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addSkip', self.test, details),
+ ('stopTest', self.test),
+ ], self.client._events)
+
+ def simple_skip_keyword(self, keyword):
+ self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
+ self.assertSkip(None)
+
+ def test_simple_skip(self):
+ self.simple_skip_keyword("skip")
+
+ def test_simple_skip_colon(self):
+ self.simple_skip_keyword("skip:")
+
+ def test_skip_empty_message(self):
+ self.protocol.lineReceived("skip mcdonalds farm [\n")
+ self.protocol.lineReceived("]\n")
+ self.assertSkip("")
+
+ def skip_quoted_bracket(self, keyword):
+ # This tests it is accepted, but cannot test it is used today, because
+ # of not having a way to expose it in Python so far.
+ self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
+ self.protocol.lineReceived(" ]\n")
+ self.protocol.lineReceived("]\n")
+ self.assertSkip("]\n")
+
+ def test_skip_quoted_bracket(self):
+ self.skip_quoted_bracket("skip")
+
+ def test_skip_colon_quoted_bracket(self):
+ self.skip_quoted_bracket("skip:")
+
+
class TestTestProtocolServerAddSuccess(unittest.TestCase):
def setUp(self):
- self.client = MockTestProtocolServerClient()
+ self.client = ExtendedTestResult()
self.protocol = subunit.TestProtocolServer(self.client)
self.protocol.lineReceived("test mcdonalds farm\n")
self.test = subunit.RemotedTestCase("mcdonalds farm")
def simple_success_keyword(self, keyword):
self.protocol.lineReceived("%s mcdonalds farm\n" % keyword)
- self.assertEqual(self.client.start_calls, [self.test])
- self.assertEqual(self.client.end_calls, [self.test])
- self.assertEqual(self.client.error_calls, [])
- self.assertEqual(self.client.success_calls, [self.test])
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addSuccess', self.test),
+ ('stopTest', self.test),
+ ], self.client._events)
def test_simple_success(self):
self.simple_success_keyword("failure")
@@ -520,6 +683,134 @@ class TestTestProtocolServerAddSuccess(unittest.TestCase):
def test_simple_success_colon(self):
self.simple_success_keyword("successful:")
+ def assertSuccess(self, details):
+ self.assertEqual([
+ ('startTest', self.test),
+ ('addSuccess', self.test, details),
+ ('stopTest', self.test),
+ ], self.client._events)
+
+ def test_success_empty_message(self):
+ self.protocol.lineReceived("success mcdonalds farm [\n")
+ self.protocol.lineReceived("]\n")
+ details = {}
+ details['message'] = Content(ContentType("text", "plain"),
+ lambda:[""])
+ self.assertSuccess(details)
+
+ def success_quoted_bracket(self, keyword):
+ # This tests it is accepted, but cannot test it is used today, because
+ # of not having a way to expose it in Python so far.
+ self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword)
+ self.protocol.lineReceived(" ]\n")
+ self.protocol.lineReceived("]\n")
+ details = {}
+ details['message'] = Content(ContentType("text", "plain"),
+ lambda:["]\n"])
+ self.assertSuccess(details)
+
+ def test_success_quoted_bracket(self):
+ self.success_quoted_bracket("success")
+
+ def test_success_colon_quoted_bracket(self):
+ self.success_quoted_bracket("success:")
+
+
+class TestTestProtocolServerProgress(unittest.TestCase):
+ """Test receipt of progress: directives."""
+
+ def test_progress_accepted_stdlib(self):
+ self.result = Python26TestResult()
+ 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 = ExtendedTestResult()
+ self.stream = StringIO()
+ self.protocol = subunit.TestProtocolServer(self.result,
+ stream=self.stream)
+ self.protocol.lineReceived("progress: 23")
+ self.protocol.lineReceived("progress: push")
+ self.protocol.lineReceived("progress: -2")
+ self.protocol.lineReceived("progress: pop")
+ self.protocol.lineReceived("progress: +4")
+ self.assertEqual("", self.stream.getvalue())
+ self.assertEqual([
+ ('progress', 23, subunit.PROGRESS_SET),
+ ('progress', None, subunit.PROGRESS_PUSH),
+ ('progress', -2, subunit.PROGRESS_CUR),
+ ('progress', None, subunit.PROGRESS_POP),
+ ('progress', 4, subunit.PROGRESS_CUR),
+ ], self.result._events)
+
+
+class TestTestProtocolServerStreamTags(unittest.TestCase):
+ """Test managing tags on the protocol level."""
+
+ def setUp(self):
+ self.client = ExtendedTestResult()
+ self.protocol = subunit.TestProtocolServer(self.client)
+
+ def test_initial_tags(self):
+ self.protocol.lineReceived("tags: foo bar:baz quux\n")
+ self.assertEqual([
+ ('tags', set(["foo", "bar:baz", "quux"]), set()),
+ ], self.client._events)
+
+ def test_minus_removes_tags(self):
+ self.protocol.lineReceived("tags: -bar quux\n")
+ self.assertEqual([
+ ('tags', set(["quux"]), set(["bar"])),
+ ], self.client._events)
+
+ def test_tags_do_not_get_set_on_test(self):
+ self.protocol.lineReceived("test mcdonalds farm\n")
+ test = self.client._events[0][-1]
+ self.assertEqual(None, getattr(test, 'tags', None))
+
+ def test_tags_do_not_get_set_on_global_tags(self):
+ self.protocol.lineReceived("tags: foo bar\n")
+ self.protocol.lineReceived("test mcdonalds farm\n")
+ test = self.client._events[-1][-1]
+ self.assertEqual(None, getattr(test, 'tags', None))
+
+ def test_tags_get_set_on_test_tags(self):
+ self.protocol.lineReceived("test mcdonalds farm\n")
+ test = self.client._events[-1][-1]
+ self.protocol.lineReceived("tags: foo bar\n")
+ self.protocol.lineReceived("success mcdonalds farm\n")
+ self.assertEqual(None, getattr(test, 'tags', None))
+
+
+class TestTestProtocolServerStreamTime(unittest.TestCase):
+ """Test managing time information at the protocol level."""
+
+ def test_time_accepted_stdlib(self):
+ self.result = Python26TestResult()
+ self.stream = StringIO()
+ self.protocol = subunit.TestProtocolServer(self.result,
+ stream=self.stream)
+ self.protocol.lineReceived("time: 2001-12-12 12:59:59Z\n")
+ self.assertEqual("", self.stream.getvalue())
+
+ def test_time_accepted_extended(self):
+ self.result = ExtendedTestResult()
+ self.stream = StringIO()
+ self.protocol = subunit.TestProtocolServer(self.result,
+ stream=self.stream)
+ self.protocol.lineReceived("time: 2001-12-12 12:59:59Z\n")
+ self.assertEqual("", self.stream.getvalue())
+ self.assertEqual([
+ ('time', datetime.datetime(2001, 12, 12, 12, 59, 59, 0,
+ iso8601.Utc()))
+ ], self.result._events)
+
class TestRemotedTestCase(unittest.TestCase):
@@ -529,14 +820,14 @@ class TestRemotedTestCase(unittest.TestCase):
self.assertRaises(NotImplementedError, test.tearDown)
self.assertEqual("A test description",
test.shortDescription())
- self.assertEqual("subunit.RemotedTestCase.A test description",
+ self.assertEqual("A test description",
test.id())
self.assertEqual("A test description (subunit.RemotedTestCase)", "%s" % test)
self.assertEqual("<subunit.RemotedTestCase description="
"'A test description'>", "%r" % test)
result = unittest.TestResult()
test.run(result)
- self.assertEqual([(test, "RemoteException: "
+ self.assertEqual([(test, _remote_exception_str + ": "
"Cannot run RemotedTestCases.\n\n")],
result.errors)
self.assertEqual(1, result.testsRun)
@@ -570,27 +861,43 @@ class TestExecTestCase(unittest.TestCase):
# the sample script runs three tests, one each
# that fails, errors and succeeds
+ def test_sample_method_args(self):
+ """sample-script.py foo"""
+ # sample that will run just one test.
def test_construct(self):
test = self.SampleExecTestCase("test_sample_method")
self.assertEqual(test.script,
subunit.join_dir(__file__, 'sample-script.py'))
+ def test_args(self):
+ result = unittest.TestResult()
+ test = self.SampleExecTestCase("test_sample_method_args")
+ test.run(result)
+ self.assertEqual(1, result.testsRun)
+
def test_run(self):
- runner = MockTestProtocolServerClient()
+ result = ExtendedTestResult()
test = self.SampleExecTestCase("test_sample_method")
- test.run(runner)
+ test.run(result)
mcdonald = subunit.RemotedTestCase("old mcdonald")
bing = subunit.RemotedTestCase("bing crosby")
+ bing_details = {}
+ bing_details['traceback'] = Content(ContentType("text", "x-traceback"),
+ lambda:["foo.c:53:ERROR invalid state\n"])
an_error = subunit.RemotedTestCase("an error")
- self.assertEqual(runner.error_calls,
- [(an_error, subunit.RemoteError())])
- self.assertEqual(runner.failure_calls,
- [(bing,
- subunit.RemoteError(
- "foo.c:53:ERROR invalid state\n"))])
- self.assertEqual(runner.start_calls, [mcdonald, bing, an_error])
- self.assertEqual(runner.end_calls, [mcdonald, bing, an_error])
+ error_details = {}
+ self.assertEqual([
+ ('startTest', mcdonald),
+ ('addSuccess', mcdonald),
+ ('stopTest', mcdonald),
+ ('startTest', bing),
+ ('addFailure', bing, bing_details),
+ ('stopTest', bing),
+ ('startTest', an_error),
+ ('addError', an_error, error_details),
+ ('stopTest', an_error),
+ ], result._events)
def test_debug(self):
test = self.SampleExecTestCase("test_sample_method")
@@ -689,7 +996,11 @@ class TestTestProtocolClient(unittest.TestCase):
self.io = StringIO()
self.protocol = subunit.TestProtocolClient(self.io)
self.test = TestTestProtocolClient("test_start_test")
-
+ self.sample_details = {'something':Content(
+ ContentType('text', 'plain'), lambda:['serialised\nform'])}
+ self.sample_tb_details = dict(self.sample_details)
+ self.sample_tb_details['traceback'] = TracebackContent(
+ subunit.RemoteError("boo qux"), self.test)
def test_start_test(self):
"""Test startTest on a TestProtocolClient."""
@@ -697,7 +1008,7 @@ class TestTestProtocolClient(unittest.TestCase):
self.assertEqual(self.io.getvalue(), "test: %s\n" % self.test.id())
def test_stop_test(self):
- """Test stopTest on a TestProtocolClient."""
+ # stopTest doesn't output anything.
self.protocol.stopTest(self.test)
self.assertEqual(self.io.getvalue(), "")
@@ -707,22 +1018,154 @@ class TestTestProtocolClient(unittest.TestCase):
self.assertEqual(
self.io.getvalue(), "successful: %s\n" % self.test.id())
+ def test_add_success_details(self):
+ """Test addSuccess on a TestProtocolClient with details."""
+ self.protocol.addSuccess(self.test, details=self.sample_details)
+ self.assertEqual(
+ self.io.getvalue(), "successful: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "something\n"
+ "F\r\nserialised\nform0\r\n]\n" % self.test.id())
+
def test_add_failure(self):
"""Test addFailure on a TestProtocolClient."""
- self.protocol.addFailure(self.test, subunit.RemoteError("boo"))
+ self.protocol.addFailure(
+ self.test, subunit.RemoteError("boo qux"))
+ self.assertEqual(
+ self.io.getvalue(),
+ ('failure: %s [\n' + _remote_exception_str + ': boo qux\n]\n')
+ % self.test.id())
+
+ def test_add_failure_details(self):
+ """Test addFailure on a TestProtocolClient with details."""
+ self.protocol.addFailure(
+ self.test, details=self.sample_tb_details)
self.assertEqual(
self.io.getvalue(),
- 'failure: %s [\nRemoteException: boo\n]\n' % self.test.id())
+ ("failure: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "something\n"
+ "F\r\nserialised\nform0\r\n"
+ "Content-Type: text/x-traceback;charset=utf8,language=python\n"
+ "traceback\n"
+ "1A\r\n" + _remote_exception_str + ": boo qux\n0\r\n"
+ "]\n") % self.test.id())
def test_add_error(self):
"""Test stopTest on a TestProtocolClient."""
- self.protocol.addError(self.test, subunit.RemoteError("phwoar"))
+ self.protocol.addError(
+ self.test, subunit.RemoteError("phwoar crikey"))
+ self.assertEqual(
+ self.io.getvalue(),
+ ('error: %s [\n' +
+ _remote_exception_str + ": phwoar crikey\n"
+ "]\n") % self.test.id())
+
+ def test_add_error_details(self):
+ """Test stopTest on a TestProtocolClient with details."""
+ self.protocol.addError(
+ self.test, details=self.sample_tb_details)
+ self.assertEqual(
+ self.io.getvalue(),
+ ("error: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "something\n"
+ "F\r\nserialised\nform0\r\n"
+ "Content-Type: text/x-traceback;charset=utf8,language=python\n"
+ "traceback\n"
+ "1A\r\n" + _remote_exception_str + ": boo qux\n0\r\n"
+ "]\n") % self.test.id())
+
+ def test_add_expected_failure(self):
+ """Test addExpectedFailure on a TestProtocolClient."""
+ self.protocol.addExpectedFailure(
+ self.test, subunit.RemoteError("phwoar crikey"))
self.assertEqual(
self.io.getvalue(),
- 'error: %s [\n'
- "RemoteException: phwoar\n"
+ ('xfail: %s [\n' +
+ _remote_exception_str + ": phwoar crikey\n"
+ "]\n") % self.test.id())
+
+ def test_add_expected_failure_details(self):
+ """Test addExpectedFailure on a TestProtocolClient with details."""
+ self.protocol.addExpectedFailure(
+ self.test, details=self.sample_tb_details)
+ self.assertEqual(
+ self.io.getvalue(),
+ ("xfail: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "something\n"
+ "F\r\nserialised\nform0\r\n"
+ "Content-Type: text/x-traceback;charset=utf8,language=python\n"
+ "traceback\n"
+ "1A\r\n"+ _remote_exception_str + ": boo qux\n0\r\n"
+ "]\n") % self.test.id())
+
+ def test_add_skip(self):
+ """Test addSkip on a TestProtocolClient."""
+ self.protocol.addSkip(
+ self.test, "Has it really?")
+ self.assertEqual(
+ self.io.getvalue(),
+ 'skip: %s [\nHas it really?\n]\n' % self.test.id())
+
+ def test_add_skip_details(self):
+ """Test addSkip on a TestProtocolClient with details."""
+ details = {'reason':Content(
+ ContentType('text', 'plain'), lambda:['Has it really?'])}
+ self.protocol.addSkip(
+ self.test, details=details)
+ self.assertEqual(
+ self.io.getvalue(),
+ "skip: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "reason\n"
+ "E\r\nHas it really?0\r\n"
"]\n" % self.test.id())
+ def test_progress_set(self):
+ self.protocol.progress(23, subunit.PROGRESS_SET)
+ self.assertEqual(self.io.getvalue(), 'progress: 23\n')
+
+ def test_progress_neg_cur(self):
+ self.protocol.progress(-23, subunit.PROGRESS_CUR)
+ self.assertEqual(self.io.getvalue(), 'progress: -23\n')
+
+ def test_progress_pos_cur(self):
+ self.protocol.progress(23, subunit.PROGRESS_CUR)
+ self.assertEqual(self.io.getvalue(), 'progress: +23\n')
+
+ def test_progress_pop(self):
+ self.protocol.progress(1234, subunit.PROGRESS_POP)
+ self.assertEqual(self.io.getvalue(), 'progress: pop\n')
+
+ def test_progress_push(self):
+ self.protocol.progress(1234, subunit.PROGRESS_PUSH)
+ self.assertEqual(self.io.getvalue(), 'progress: push\n')
+
+ def test_time(self):
+ # Calling time() outputs a time signal immediately.
+ self.protocol.time(
+ datetime.datetime(2009,10,11,12,13,14,15, iso8601.Utc()))
+ self.assertEqual(
+ "time: 2009-10-11 12:13:14.000015Z\n",
+ self.io.getvalue())
+
+ def test_add_unexpected_success(self):
+ """Test addUnexpectedSuccess on a TestProtocolClient."""
+ self.protocol.addUnexpectedSuccess(self.test)
+ self.assertEqual(
+ self.io.getvalue(), "successful: %s\n" % self.test.id())
+
+ def test_add_unexpected_success_details(self):
+ """Test addUnexpectedSuccess on a TestProtocolClient with details."""
+ self.protocol.addUnexpectedSuccess(self.test, details=self.sample_details)
+ self.assertEqual(
+ self.io.getvalue(), "successful: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "something\n"
+ "F\r\nserialised\nform0\r\n]\n" % self.test.id())
+
def test_suite():
loader = subunit.tests.TestUtil.TestLoader()
diff --git a/lib/subunit/python/subunit/tests/test_test_results.py b/lib/subunit/python/subunit/tests/test_test_results.py
new file mode 100644
index 0000000000..fe82c04b06
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_test_results.py
@@ -0,0 +1,199 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
+#
+# 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 datetime
+import unittest
+from StringIO import StringIO
+import os
+import sys
+
+from testtools.content_type import ContentType
+from testtools.content import Content
+
+import subunit
+import subunit.iso8601 as iso8601
+import subunit.test_results
+
+
+class LoggingDecorator(subunit.test_results.HookedTestResultDecorator):
+
+ def __init__(self, decorated):
+ self._calls = 0
+ super(LoggingDecorator, self).__init__(decorated)
+
+ def _before_event(self):
+ self._calls += 1
+
+
+class AssertBeforeTestResult(LoggingDecorator):
+ """A TestResult for checking preconditions."""
+
+ def __init__(self, decorated, test):
+ self.test = test
+ super(AssertBeforeTestResult, self).__init__(decorated)
+
+ def _before_event(self):
+ self.test.assertEqual(1, self.earlier._calls)
+ super(AssertBeforeTestResult, self)._before_event()
+
+
+class TimeCapturingResult(unittest.TestResult):
+
+ def __init__(self):
+ super(TimeCapturingResult, self).__init__()
+ self._calls = []
+
+ def time(self, a_datetime):
+ self._calls.append(a_datetime)
+
+
+class TestHookedTestResultDecorator(unittest.TestCase):
+
+ def setUp(self):
+ # An end to the chain
+ terminal = unittest.TestResult()
+ # Asserts that the call was made to self.result before asserter was
+ # called.
+ asserter = AssertBeforeTestResult(terminal, self)
+ # The result object we call, which much increase its call count.
+ self.result = LoggingDecorator(asserter)
+ asserter.earlier = self.result
+ self.decorated = asserter
+
+ def tearDown(self):
+ # The hook in self.result must have been called
+ self.assertEqual(1, self.result._calls)
+ # The hook in asserter must have been called too, otherwise the
+ # assertion about ordering won't have completed.
+ self.assertEqual(1, self.decorated._calls)
+
+ def test_startTest(self):
+ self.result.startTest(self)
+
+ def test_startTestRun(self):
+ self.result.startTestRun()
+
+ def test_stopTest(self):
+ self.result.stopTest(self)
+
+ def test_stopTestRun(self):
+ self.result.stopTestRun()
+
+ def test_addError(self):
+ self.result.addError(self, subunit.RemoteError())
+
+ def test_addError_details(self):
+ self.result.addError(self, details={})
+
+ def test_addFailure(self):
+ self.result.addFailure(self, subunit.RemoteError())
+
+ def test_addFailure_details(self):
+ self.result.addFailure(self, details={})
+
+ def test_addSuccess(self):
+ self.result.addSuccess(self)
+
+ def test_addSuccess_details(self):
+ self.result.addSuccess(self, details={})
+
+ def test_addSkip(self):
+ self.result.addSkip(self, "foo")
+
+ def test_addSkip_details(self):
+ self.result.addSkip(self, details={})
+
+ def test_addExpectedFailure(self):
+ self.result.addExpectedFailure(self, subunit.RemoteError())
+
+ def test_addExpectedFailure_details(self):
+ self.result.addExpectedFailure(self, details={})
+
+ def test_addUnexpectedSuccess(self):
+ self.result.addUnexpectedSuccess(self)
+
+ def test_addUnexpectedSuccess_details(self):
+ self.result.addUnexpectedSuccess(self, details={})
+
+ def test_progress(self):
+ self.result.progress(1, subunit.PROGRESS_SET)
+
+ def test_wasSuccessful(self):
+ self.result.wasSuccessful()
+
+ def test_shouldStop(self):
+ self.result.shouldStop
+
+ def test_stop(self):
+ self.result.stop()
+
+ def test_time(self):
+ self.result.time(None)
+
+
+class TestAutoTimingTestResultDecorator(unittest.TestCase):
+
+ def setUp(self):
+ # And end to the chain which captures time events.
+ terminal = TimeCapturingResult()
+ # The result object under test.
+ self.result = subunit.test_results.AutoTimingTestResultDecorator(
+ terminal)
+ self.decorated = terminal
+
+ def test_without_time_calls_time_is_called_and_not_None(self):
+ self.result.startTest(self)
+ self.assertEqual(1, len(self.decorated._calls))
+ self.assertNotEqual(None, self.decorated._calls[0])
+
+ def test_no_time_from_progress(self):
+ self.result.progress(1, subunit.PROGRESS_CUR)
+ self.assertEqual(0, len(self.decorated._calls))
+
+ def test_no_time_from_shouldStop(self):
+ self.decorated.stop()
+ self.result.shouldStop
+ self.assertEqual(0, len(self.decorated._calls))
+
+ def test_calling_time_inhibits_automatic_time(self):
+ # Calling time() outputs a time signal immediately and prevents
+ # automatically adding one when other methods are called.
+ time = datetime.datetime(2009,10,11,12,13,14,15, iso8601.Utc())
+ self.result.time(time)
+ self.result.startTest(self)
+ self.result.stopTest(self)
+ self.assertEqual(1, len(self.decorated._calls))
+ self.assertEqual(time, self.decorated._calls[0])
+
+ def test_calling_time_None_enables_automatic_time(self):
+ time = datetime.datetime(2009,10,11,12,13,14,15, iso8601.Utc())
+ self.result.time(time)
+ self.assertEqual(1, len(self.decorated._calls))
+ self.assertEqual(time, self.decorated._calls[0])
+ # Calling None passes the None through, in case other results care.
+ self.result.time(None)
+ self.assertEqual(2, len(self.decorated._calls))
+ self.assertEqual(None, self.decorated._calls[1])
+ # Calling other methods doesn't generate an automatic time event.
+ self.result.startTest(self)
+ self.assertEqual(3, len(self.decorated._calls))
+ self.assertNotEqual(None, self.decorated._calls[2])
+
+
+def test_suite():
+ loader = subunit.tests.TestUtil.TestLoader()
+ result = loader.loadTestsFromName(__name__)
+ return result
diff --git a/lib/subunit/tap2subunit b/lib/subunit/tap2subunit
deleted file mode 100755
index 9e335168f5..0000000000
--- a/lib/subunit/tap2subunit
+++ /dev/null
@@ -1,35 +0,0 @@
-#!/usr/bin/perl
-# Simple script that converts Perl test harness output to
-# Subunit
-# Copyright (C) 2008 Jelmer Vernooij <jelmer@samba.org>
-# Published under the GNU GPL, v3 or later
-
-my $firstline = 1;
-my $error = 0;
-while(<STDIN>) {
- if ($firstline) {
- $firstline = 0;
- next;
- }
- if (/^not ok (\d+) - (.*)$/) {
- print "test: $2\n";
- print "failure: $2\n";
- $error = 1;
- } elsif (/^ok (\d+) - (.*)$/) {
- print "test: $2\n";
- print "success: $2\n";
- } elsif (/^ok (\d+)$/) {
- print "test: $1\n";
- print "success: $1\n";
- } elsif (/^ok (\d+) # skip (.*)$/) {
- print "test: $1\n";
- print "skip: $1 [\n$2\n]\n";
- } elsif (/^not ok (\d+)$/) {
- print "test: $1\n";
- print "failure: $1\n";
- $error = 1;
- } else {
- print;
- }
-}
-exit $error;
diff --git a/lib/subunit/update.sh b/lib/subunit/update.sh
index 6ef9859c82..6d996cdfd5 100755
--- a/lib/subunit/update.sh
+++ b/lib/subunit/update.sh
@@ -5,9 +5,9 @@ TARGETDIR="`dirname $0`"
WORKDIR="`mktemp -d`"
bzr branch lp:subunit "$WORKDIR/subunit"
-for p in python filters;
+for p in python/ filters/tap2subunit;
do
- rsync -avz --delete "$WORKDIR/subunit/$p/" "$TARGETDIR/$p/"
+ rsync -avz --delete "$WORKDIR/subunit/$p" "$TARGETDIR/$p"
done
rm -rf "$WORKDIR"