diff options
Diffstat (limited to 'lib/subunit/python')
27 files changed, 0 insertions, 6426 deletions
diff --git a/lib/subunit/python/iso8601/LICENSE b/lib/subunit/python/iso8601/LICENSE deleted file mode 100644 index 5ca93dae79..0000000000 --- a/lib/subunit/python/iso8601/LICENSE +++ /dev/null @@ -1,20 +0,0 @@ -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 deleted file mode 100644 index 5ec9d45597..0000000000 --- a/lib/subunit/python/iso8601/README +++ /dev/null @@ -1,26 +0,0 @@ -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 deleted file mode 100644 index d1ed8a11a6..0000000000 --- a/lib/subunit/python/iso8601/README.subunit +++ /dev/null @@ -1,5 +0,0 @@ -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 deleted file mode 100644 index cdb61ecf6a..0000000000 --- a/lib/subunit/python/iso8601/setup.py +++ /dev/null @@ -1,58 +0,0 @@ -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 deleted file mode 100644 index ff9e2731cf..0000000000 --- a/lib/subunit/python/iso8601/test_iso8601.py +++ /dev/null @@ -1,111 +0,0 @@ -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 deleted file mode 100644 index 42dcf297e4..0000000000 --- a/lib/subunit/python/subunit/__init__.py +++ /dev/null @@ -1,1314 +0,0 @@ -# -# 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. -# - -"""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 subsequent 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 occurred 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 os -import re -import subprocess -import sys -import unittest -if sys.version_info > (3, 0): - from io import UnsupportedOperation as _UnsupportedOperation -else: - _UnsupportedOperation = AttributeError - - -from testtools import content, content_type, ExtendedToOriginalDecorator -from testtools.content import TracebackContent -from testtools.compat import _b, _u, BytesIO, StringIO -try: - from testtools.testresult.real import _StringException - RemoteException = _StringException - # For testing: different pythons have different str() implementations. - if sys.version_info > (3, 0): - _remote_exception_str = "testtools.testresult.real._StringException" - _remote_exception_str_chunked = "34\r\n" + _remote_exception_str - else: - _remote_exception_str = "_StringException" - _remote_exception_str_chunked = "1A\r\n" + _remote_exception_str -except ImportError: - raise ImportError ("testtools.testresult.real does not contain " - "_StringException, check your version.") -from testtools import testresult - -from subunit import chunked, details, iso8601, test_results - -# same format as sys.version_info: "A tuple containing the five components of -# the version number: major, minor, micro, releaselevel, and serial. All -# values except releaselevel are integers; the release level is 'alpha', -# 'beta', 'candidate', or 'final'. The version_info value corresponding to the -# Python version 2.0 is (2, 0, 0, 'final', 0)." Additionally we use a -# releaselevel of 'dev' for unreleased under-development code. -# -# If the releaselevel is 'alpha' then the major/minor/micro components are not -# established at this point, and setup.py will use a version of next-$(revno). -# If the releaselevel is 'final', then the tarball will be major.minor.micro. -# Otherwise it is major.minor.micro~$(revno). - -__version__ = (0, 0, 9, 'final', 0) - -PROGRESS_SET = 0 -PROGRESS_CUR = 1 -PROGRESS_PUSH = 2 -PROGRESS_POP = 3 - - -def test_suite(): - import subunit.tests - return subunit.tests.test_suite() - - -def join_dir(base_path, path): - """ - Returns an absolute path to C{path}, calculated relative to the parent - of C{base_path}. - - @param base_path: A path to a file or directory. - @param path: An absolute path, or a path relative to the containing - directory of C{base_path}. - - @return: An absolute path to C{path}. - """ - return os.path.join(os.path.dirname(os.path.abspath(base_path)), path) - - -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: - new_tags.add(tag) - return new_tags, gone_tags - - -class DiscardStream(object): - """A filelike object which discards what is written to it.""" - - def fileno(self): - raise _UnsupportedOperation() - - def write(self, bytes): - pass - - def read(self, len=0): - return _b('') - - -class _ParserState(object): - """State for the subunit parser.""" - - def __init__(self, parser): - self.parser = parser - self._test_sym = (_b('test'), _b('testing')) - self._colon_sym = _b(':') - self._error_sym = (_b('error'),) - self._failure_sym = (_b('failure'),) - self._progress_sym = (_b('progress'),) - self._skip_sym = _b('skip') - self._success_sym = (_b('success'), _b('successful')) - self._tags_sym = (_b('tags'),) - self._time_sym = (_b('time'),) - self._xfail_sym = (_b('xfail'),) - self._uxsuccess_sym = (_b('uxsuccess'),) - self._start_simple = _u(" [") - self._start_multipart = _u(" [ multipart") - - 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 and line.startswith(parts[0]): - cmd, rest = parts - offset = len(cmd) + 1 - cmd = cmd.rstrip(self._colon_sym) - if cmd in self._test_sym: - self.startTest(offset, line) - elif cmd in self._error_sym: - self.addError(offset, line) - elif cmd in self._failure_sym: - self.addFailure(offset, line) - elif cmd in self._progress_sym: - self.parser._handleProgress(offset, line) - elif cmd in self._skip_sym: - self.addSkip(offset, line) - elif cmd in self._success_sym: - self.addSuccess(offset, line) - elif cmd in self._tags_sym: - self.parser._handleTags(offset, line) - self.parser.subunitLineReceived(line) - elif cmd in self._time_sym: - self.parser._handleTime(offset, line) - self.parser.subunitLineReceived(line) - elif cmd in self._xfail_sym: - self.addExpectedFail(offset, line) - elif cmd in self._uxsuccess_sym: - self.addUnexpectedSuccess(offset, line) - else: - self.parser.stdOutLineReceived(line) - else: - self.parser.stdOutLineReceived(line) - - def lostConnection(self): - """Connection lost.""" - self.parser._lostConnectionInTest(_u('unknown state of ')) - - def startTest(self, offset, line): - """A test start command received.""" - self.parser.stdOutLineReceived(line) - - -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. - """ - test_name = line[offset:-1].decode('utf8') - if self.parser.current_test_description == test_name: - 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 + self._start_simple == \ - test_name: - self.parser._state = details_state - details_state.set_simple() - self.parser.subunitLineReceived(line) - elif self.parser.current_test_description + self._start_multipart == \ - test_name: - self.parser._state = details_state - details_state.set_multipart() - self.parser.subunitLineReceived(line) - else: - 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 _uxsuccess(self): - self.parser.client.addUnexpectedSuccess(self.parser._current_test) - - def addUnexpectedSuccess(self, offset, line): - """A 'uxsuccess:' directive has been read.""" - self._outcome(offset, line, self._uxsuccess, - self.parser._reading_uxsuccess_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(_u('')) - - -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 - test_name = line[offset:-1].decode('utf8') - self.parser._current_test = RemotedTestCase(test_name) - self.parser.current_test_description = test_name - 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(_u('%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 _ReadingUnexpectedSuccessDetails(_ReadingDetails): - """State for the subunit parser when reading uxsuccess details.""" - - def _report_outcome(self): - self.parser.client.addUnexpectedSuccess(self.parser._current_test, - details=self.details_parser.get_details()) - - def _outcome_label(self): - return "uxsuccess" - - -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. It should accept bytes to its write() method. - :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 - if sys.version_info > (3, 0): - stream = stream.buffer - 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) - self._reading_uxsuccess_details = _ReadingUnexpectedSuccessDetails(self) - # start with outside test. - self._state = self._outside_test - # Avoid casts on every call - self._plusminus = _b('+-') - self._push_sym = _b('push') - self._pop_sym = _b('pop') - - def _handleProgress(self, offset, line): - """Process a progress directive.""" - line = line[offset:].strip() - if line[0] in self._plusminus: - whence = PROGRESS_CUR - delta = int(line) - elif line == self._push_sym: - whence = PROGRESS_PUSH - delta = None - elif line == self._pop_sym: - whence = PROGRESS_POP - delta = None - else: - whence = PROGRESS_SET - delta = int(line) - self.client.progress(delta, whence) - - def _handleTags(self, offset, line): - """Process a tags command.""" - tags = line[offset:].decode('utf8').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: - raise TypeError(_u("Failed to parse %r, got %r") - % (line, sys.exec_info[1])) - self.client.time(event_time) - - def lineReceived(self, line): - """Call the appropriate local method for the received line.""" - self._state.lineReceived(line) - - def _lostConnectionInTest(self, state_string): - error_string = _u("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.""" - 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().""" - self._state.startTest(offset, line) - - def subunitLineReceived(self, line): - self._forward_stream.write(line) - - def stdOutLineReceived(self, line): - self._stream.write(line) - - -class TestProtocolClient(testresult.TestResult): - """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). This should accept - # bytes not strings: subunit is a byte orientated protocol. - 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): - testresult.TestResult.__init__(self) - stream = _make_stream_binary(stream) - self._stream = stream - self._progress_fmt = _b("progress: ") - self._bytes_eol = _b("\n") - self._progress_plus = _b("+") - self._progress_push = _b("push") - self._progress_pop = _b("pop") - self._empty_bytes = _b("") - self._start_simple = _b(" [\n") - self._end_simple = _b("]\n") - - 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) - if self.failfast: - self.stop() - - 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) - if self.failfast: - self.stop() - - def _addOutcome(self, outcome, test, error=None, details=None, - error_permitted=True): - """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. - :param error_permitted: If True then one and only one of error or - details must be supplied. If False then error must not be supplied - and details is still optional. """ - self._stream.write(_b("%s: " % outcome) + self._test_id(test)) - if error_permitted: - if error is None and details is None: - raise ValueError - else: - if error is not None: - raise ValueError - if error is not None: - self._stream.write(self._start_simple) - tb_content = TracebackContent(error, test) - for bytes in tb_content.iter_bytes(): - self._stream.write(bytes) - elif details is not None: - self._write_details(details) - else: - self._stream.write(_b("\n")) - if details is not None or error is not None: - self._stream.write(self._end_simple) - - 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(_b("skip: %s [\n" % test.id())) - self._stream.write(_b("%s\n" % reason)) - self._stream.write(self._end_simple) - - def addSuccess(self, test, details=None): - """Report a success in a test.""" - self._addOutcome("successful", test, details=details, error_permitted=False) - - def addUnexpectedSuccess(self, test, details=None): - """Report an unexpected success in test test. - - Details can optionally be provided: conceptually there - are two separate methods: - addError(self, test) - addError(self, test, details) - - :param details: New Testing-in-python drafted API; a dict from string - to subunit.Content objects. - """ - self._addOutcome("uxsuccess", test, details=details, - error_permitted=False) - if self.failfast: - self.stop() - - def _test_id(self, test): - result = test.id() - if type(result) is not bytes: - result = result.encode('utf8') - return result - - def startTest(self, test): - """Mark a test as starting its test run.""" - super(TestProtocolClient, self).startTest(test) - self._stream.write(_b("test: ") + self._test_id(test) + _b("\n")) - self._stream.flush() - - def stopTest(self, test): - super(TestProtocolClient, self).stopTest(test) - self._stream.flush() - - 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 = self._progress_plus - offset = _b(str(offset)) - elif whence == PROGRESS_PUSH: - prefix = self._empty_bytes - offset = self._progress_push - elif whence == PROGRESS_POP: - prefix = self._empty_bytes - offset = self._progress_pop - else: - prefix = self._empty_bytes - offset = _b(str(offset)) - self._stream.write(self._progress_fmt + prefix + offset + - self._bytes_eol) - - def tags(self, new_tags, gone_tags): - """Inform the client about tags added/removed from the stream.""" - if not new_tags and not gone_tags: - return - tags = set([tag.encode('utf8') for tag in new_tags]) - tags.update([_b("-") + tag.encode('utf8') for tag in gone_tags]) - tag_line = _b("tags: ") + _b(" ").join(tags) + _b("\n") - self._stream.write(tag_line) - - 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(_b("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(_b(" [ multipart\n")) - for name, content in sorted(details.items()): - self._stream.write(_b("Content-Type: %s/%s" % - (content.content_type.type, content.content_type.subtype))) - parameters = content.content_type.parameters - if parameters: - self._stream.write(_b(";")) - param_strs = [] - for param, value in parameters.items(): - param_strs.append("%s=%s" % (param, value)) - self._stream.write(_b(",".join(param_strs))) - self._stream.write(_b("\n%s\n" % name)) - encoder = chunked.Encoder(self._stream) - list(map(encoder.write, content.iter_bytes())) - encoder.close() - - def done(self): - """Obey the testtools result.done() interface.""" - - -def RemoteError(description=_u("")): - return (_StringException, _StringException(description), None) - - -class RemotedTestCase(unittest.TestCase): - """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: - return self.__description == other.__description - except AttributeError: - return False - - def __init__(self, description): - """Create a psuedo test case with description description.""" - self.__description = description - - def error(self, label): - raise NotImplementedError("%s on RemotedTestCases is not permitted." % - label) - - def setUp(self): - self.error("setUp") - - def tearDown(self): - self.error("tearDown") - - def shortDescription(self): - return self.__description - - def id(self): - return "%s" % (self.__description,) - - def __str__(self): - return "%s (%s)" % (self.__description, self._strclass()) - - def __repr__(self): - return "<%s description='%s'>" % \ - (self._strclass(), self.__description) - - def run(self, result=None): - if result is None: result = self.defaultTestResult() - result.startTest(self) - result.addError(self, RemoteError(_u("Cannot run RemotedTestCases.\n"))) - result.stopTest(self) - - def _strclass(self): - cls = self.__class__ - return "%s.%s" % (cls.__module__, cls.__name__) - - -class ExecTestCase(unittest.TestCase): - """A test case which runs external scripts for test fixtures.""" - - def __init__(self, methodName='runTest'): - """Create an instance of the class that will use the named test - method when executed. Raises a ValueError if the instance does - not have a method with the specified name. - """ - unittest.TestCase.__init__(self, methodName) - testMethod = getattr(self, methodName) - self.script = join_dir(sys.modules[self.__class__.__module__].__file__, - testMethod.__doc__) - - def countTestCases(self): - return 1 - - def run(self, result=None): - if result is None: result = self.defaultTestResult() - self._run(result) - - def debug(self): - """Run the test without collecting errors in a TestResult""" - self._run(testresult.TestResult()) - - def _run(self, result): - protocol = TestProtocolServer(result) - process = subprocess.Popen(self.script, shell=True, - stdout=subprocess.PIPE) - _make_stream_binary(process.stdout) - output = process.communicate()[0] - protocol.readFrom(BytesIO(output)) - - -class IsolatedTestCase(unittest.TestCase): - """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() - run_isolated(unittest.TestCase, self, result) - - -class IsolatedTestSuite(unittest.TestSuite): - """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 = testresult.TestResult() - run_isolated(unittest.TestSuite, self, result) - - -def run_isolated(klass, self, result): - """Run a test suite or case in a subprocess, using the run method on klass. - """ - c2pread, c2pwrite = os.pipe() - # fixme - error -> result - # now fork - pid = os.fork() - if pid == 0: - # Child - # Close parent's pipe ends - os.close(c2pread) - # Dup fds for child - os.dup2(c2pwrite, 1) - # Close pipe fds. - os.close(c2pwrite) - - # at this point, sys.stdin is redirected, now we want - # to filter it to escape ]'s. - ### XXX: test and write that bit. - stream = os.fdopen(1, 'wb') - result = TestProtocolClient(stream) - klass.run(self, result) - stream.flush() - sys.stderr.flush() - # exit HARD, exit NOW. - os._exit(0) - else: - # Parent - # Close child pipe ends - os.close(c2pwrite) - # hookup a protocol engine - protocol = TestProtocolServer(result) - fileobj = os.fdopen(c2pread, 'rb') - protocol.readFrom(fileobj) - os.waitpid(pid, 0) - # TODO return code evaluation. - return result - - -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 - 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|skip|todo)(?:\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.upper() == 'TODO': - result = 'xfail' - elif directive.upper() == '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=None): - """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. - """ - stream = _make_stream_binary(stream) - self._stream = stream - self._passthrough = passthrough - if forward is not None: - forward = _make_stream_binary(forward) - self._forward = forward - - def __call__(self, result=None): - return self.run(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(testresult.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.""" - testresult.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 - - -def get_default_formatter(): - """Obtain the default formatter to write to. - - :return: A file-like object. - """ - formatter = os.getenv("SUBUNIT_FORMATTER") - if formatter: - return os.popen(formatter, "w") - else: - stream = sys.stdout - if sys.version_info > (3, 0): - stream = stream.buffer - return stream - - -def read_test_list(path): - """Read a list of test ids from a file on disk. - - :param path: Path to the file - :return: Sequence of test ids - """ - f = open(path, 'rb') - try: - return [l.rstrip("\n") for l in f.readlines()] - finally: - f.close() - - -def _make_stream_binary(stream): - """Ensure that a stream will be binary safe. See _make_binary_on_windows. - - :return: A binary version of the same stream (some streams cannot be - 'fixed' but can be unwrapped). - """ - try: - fileno = stream.fileno() - except _UnsupportedOperation: - pass - else: - _make_binary_on_windows(fileno) - return _unwrap_text(stream) - -def _make_binary_on_windows(fileno): - """Win32 mangles \r\n to \n and that breaks streams. See bug lp:505078.""" - if sys.platform == "win32": - import msvcrt - msvcrt.setmode(fileno, os.O_BINARY) - - -def _unwrap_text(stream): - """Unwrap stream if it is a text stream to get the original buffer.""" - if sys.version_info > (3, 0): - try: - # Read streams - if type(stream.read(0)) is str: - return stream.buffer - except (_UnsupportedOperation, IOError): - # Cannot read from the stream: try via writes - try: - stream.write(_b('')) - except TypeError: - return stream.buffer - return stream diff --git a/lib/subunit/python/subunit/chunked.py b/lib/subunit/python/subunit/chunked.py deleted file mode 100644 index b9921291ea..0000000000 --- a/lib/subunit/python/subunit/chunked.py +++ /dev/null @@ -1,185 +0,0 @@ -# -# subunit: extensions to python unittest to get test results from subprocesses. -# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net> -# Copyright (C) 2011 Martin Pool <mbp@sourcefrog.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.""" - -from testtools.compat import _b - -empty = _b('') - -class Decoder(object): - """Decode chunked content to a byte stream.""" - - def __init__(self, output, strict=True): - """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. The output object must accept bytes objects. - - :param strict: If True (the default), the decoder will not knowingly - accept input that is not conformant to the HTTP specification. - (This does not imply that it will catch every nonconformance.) - If False, it will accept incorrect input that is still - unambiguous. - """ - self.output = output - self.buffered_bytes = [] - self.state = self._read_length - self.body_length = 0 - self.strict = strict - self._match_chars = _b("0123456789abcdefABCDEF\r\n") - self._slash_n = _b('\n') - self._slash_r = _b('\r') - self._slash_rn = _b('\r\n') - self._slash_nr = _b('\n\r') - - 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 empty.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_chars = [] - for bytes in self.buffered_bytes: - for pos in range(len(bytes)): - byte = bytes[pos:pos+1] - if byte not in self._match_chars: - break - count_chars.append(byte) - if byte == self._slash_n: - break - if not count_chars: - return - if count_chars[-1] != self._slash_n: - return - count_str = empty.join(count_chars) - if self.strict: - if count_str[-2:] != self._slash_rn: - raise ValueError("chunk header invalid: %r" % count_str) - if self._slash_r in count_str[:-2]: - raise ValueError("too many CRs in chunk header %r" % count_str) - self.body_length = int(count_str.rstrip(self._slash_nr), 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 empty - 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(_b("%X\r\n" % (buffer_size + extra_len))) - if buffer_size: - self.output.write(empty.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(_b("0\r\n")) diff --git a/lib/subunit/python/subunit/details.py b/lib/subunit/python/subunit/details.py deleted file mode 100644 index 9e5e005864..0000000000 --- a/lib/subunit/python/subunit/details.py +++ /dev/null @@ -1,119 +0,0 @@ -# -# 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 testtools import content, content_type -from testtools.compat import _b, BytesIO - -from subunit import chunked - -end_marker = _b("]\n") -quoted_marker = _b(" ]") -empty = _b('') - - -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 = _b("") - self._state = state - - def lineReceived(self, line): - if line == end_marker: - self._state.endDetails() - return - if line[0:2] == quoted_marker: - # quoted ] start - self._message += line[1:] - else: - self._message += line - - def get_details(self, style=None): - result = {} - if not style: - # We know that subunit/testtools serialise [] formatted - # tracebacks as utf8, but perhaps we need a ReplacingContent - # or something like that. - result['traceback'] = content.Content( - content_type.ContentType("text", "x-traceback", - {"charset": "utf8"}), - 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 == end_marker: - self._state.endDetails() - return - # TODO error handling - field, value = line[:-1].decode('utf8').split(' ', 1) - try: - main, sub = value.split('/') - except ValueError: - raise ValueError("Invalid MIME type %r" % value) - self._content_type = content_type.ContentType(main, sub) - self._parse_state = self._get_name - - def _get_name(self, line): - self._name = line[:-1].decode('utf8') - self._body = BytesIO() - 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 == empty, 'residue: %r' % (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/filters.py b/lib/subunit/python/subunit/filters.py deleted file mode 100644 index dc3fd8aedb..0000000000 --- a/lib/subunit/python/subunit/filters.py +++ /dev/null @@ -1,125 +0,0 @@ -# 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. -# - - -from optparse import OptionParser -import sys - -from subunit import DiscardStream, ProtocolTestCase - - -def make_options(description): - parser = OptionParser(description=description) - parser.add_option( - "--no-passthrough", action="store_true", - help="Hide all non subunit input.", default=False, - dest="no_passthrough") - parser.add_option( - "-o", "--output-to", - help="Send the output to this path rather than stdout.") - parser.add_option( - "-f", "--forward", action="store_true", default=False, - help="Forward subunit stream on stdout.") - return parser - - -def run_tests_from_stream(input_stream, result, passthrough_stream=None, - forward_stream=None): - """Run tests from a subunit input stream through 'result'. - - :param input_stream: A stream containing subunit input. - :param result: A TestResult that will receive the test events. - :param passthrough_stream: All non-subunit input received will be - sent to this stream. If not provided, uses the ``TestProtocolServer`` - default, which is ``sys.stdout``. - :param forward_stream: All subunit input received will be forwarded - to this stream. If not provided, uses the ``TestProtocolServer`` - default, which is to not forward any input. - """ - test = ProtocolTestCase( - input_stream, passthrough=passthrough_stream, - forward=forward_stream) - result.startTestRun() - test.run(result) - result.stopTestRun() - - -def filter_by_result(result_factory, output_path, passthrough, forward, - input_stream=sys.stdin): - """Filter an input stream using a test result. - - :param result_factory: A callable that when passed an output stream - returns a TestResult. It is expected that this result will output - to the given stream. - :param output_path: A path send output to. If None, output will be go - to ``sys.stdout``. - :param passthrough: If True, all non-subunit input will be sent to - ``sys.stdout``. If False, that input will be discarded. - :param forward: If True, all subunit input will be forwarded directly to - ``sys.stdout`` as well as to the ``TestResult``. - :param input_stream: The source of subunit input. Defaults to - ``sys.stdin``. - :return: A test result with the resultts of the run. - """ - if passthrough: - passthrough_stream = sys.stdout - else: - passthrough_stream = DiscardStream() - - if forward: - forward_stream = sys.stdout - else: - forward_stream = DiscardStream() - - if output_path is None: - output_to = sys.stdout - else: - output_to = file(output_path, 'wb') - - try: - result = result_factory(output_to) - run_tests_from_stream( - input_stream, result, passthrough_stream, forward_stream) - finally: - if output_path: - output_to.close() - return result - - -def run_filter_script(result_factory, description, post_run_hook=None): - """Main function for simple subunit filter scripts. - - Many subunit filter scripts take a stream of subunit input and use a - TestResult to handle the events generated by that stream. This function - wraps a lot of the boiler-plate around that by making a script with - options for handling passthrough information and stream forwarding, and - that will exit with a successful return code (i.e. 0) if the input stream - represents a successful test run. - - :param result_factory: A callable that takes an output stream and returns - a test result that outputs to that stream. - :param description: A description of the filter script. - """ - parser = make_options(description) - (options, args) = parser.parse_args() - result = filter_by_result( - result_factory, options.output_to, not options.no_passthrough, - options.forward) - if post_run_hook: - post_run_hook(result) - if result.wasSuccessful(): - sys.exit(0) - else: - sys.exit(1) diff --git a/lib/subunit/python/subunit/iso8601.py b/lib/subunit/python/subunit/iso8601.py deleted file mode 100644 index 07855d0975..0000000000 --- a/lib/subunit/python/subunit/iso8601.py +++ /dev/null @@ -1,133 +0,0 @@ -# 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 -import sys - -__all__ = ["parse_date", "ParseError"] - -# Adapted from http://delete.me.uk/2005/03/iso8601.html -ISO8601_REGEX_PATTERN = (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_PATTERN = "(?P<prefix>[+-])(?P<hours>[0-9]{2}).(?P<minutes>[0-9]{2})" -ISO8601_REGEX = re.compile(ISO8601_REGEX_PATTERN.encode('utf8')) -TIMEZONE_REGEX = re.compile(TIMEZONE_REGEX_PATTERN.encode('utf8')) - -zulu = "Z".encode('latin-1') -minus = "-".encode('latin-1') - -if sys.version_info < (3, 0): - bytes = str - - -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 == zulu: - 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 == minus: - 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, bytes): - raise ParseError("Expecting bytes %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"].decode()) * 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 deleted file mode 100644 index 3a6af89a33..0000000000 --- a/lib/subunit/python/subunit/progress_model.py +++ /dev/null @@ -1,106 +0,0 @@ -# -# 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 deleted file mode 100755 index b5ccea449d..0000000000 --- a/lib/subunit/python/subunit/run.py +++ /dev/null @@ -1,84 +0,0 @@ -#!/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, get_default_formatter -from subunit.test_results import AutoTimingTestResultDecorator -from testtools.run import ( - BUFFEROUTPUT, - CATCHBREAK, - FAILFAST, - TestProgram, - USAGE_AS_MAIN, - ) - - -class SubunitTestRunner(object): - def __init__(self, verbosity=None, failfast=None, buffer=None, stream=None): - """Create a TestToolsTestRunner. - - :param verbosity: Ignored. - :param failfast: Stop running tests at the first failure. - :param buffer: Ignored. - """ - self.failfast = failfast - self.stream = stream or sys.stdout - - def run(self, test): - "Run the given test case or test suite." - result = TestProtocolClient(self.stream) - result = AutoTimingTestResultDecorator(result) - if self.failfast is not None: - result.failfast = self.failfast - test(result) - return result - - -class SubunitTestProgram(TestProgram): - - USAGE = USAGE_AS_MAIN - - def usageExit(self, msg=None): - if msg: - print (msg) - usage = {'progName': self.progName, 'catchbreak': '', 'failfast': '', - 'buffer': ''} - if self.failfast != False: - usage['failfast'] = FAILFAST - if self.catchbreak != False: - usage['catchbreak'] = CATCHBREAK - if self.buffer != False: - usage['buffer'] = BUFFEROUTPUT - usage_text = self.USAGE % usage - usage_lines = usage_text.split('\n') - usage_lines.insert(2, "Run a test suite with a subunit reporter.") - usage_lines.insert(3, "") - print('\n'.join(usage_lines)) - sys.exit(2) - - -if __name__ == '__main__': - stream = get_default_formatter() - runner = SubunitTestRunner - SubunitTestProgram(module=None, argv=sys.argv, testRunner=runner, - stdout=sys.stdout) diff --git a/lib/subunit/python/subunit/test_results.py b/lib/subunit/python/subunit/test_results.py deleted file mode 100644 index 91c9bbdc1e..0000000000 --- a/lib/subunit/python/subunit/test_results.py +++ /dev/null @@ -1,678 +0,0 @@ -# -# 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 csv -import datetime - -import testtools -from testtools.compat import all -from testtools.content import ( - text_content, - TracebackContent, - ) - -from subunit import iso8601 - - -# 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. - """ - - # XXX: Since lp:testtools r250, this is in testtools. Once it's released, - # we should gut this and just use that. - - 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 _get_failfast(self): - return getattr(self.decorated, 'failfast', False) - - def _set_failfast(self, value): - self.decorated.failfast = value - failfast = property(_get_failfast, _set_failfast) - - 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() - - @property - def testsRun(self): - return self.decorated.testsRun - - def tags(self, new_tags, gone_tags): - return self.decorated.tags(new_tags, gone_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 TagsMixin(object): - - def __init__(self): - self._clear_tags() - - def _clear_tags(self): - self._global_tags = set(), set() - self._test_tags = None - - def _get_active_tags(self): - global_new, global_gone = self._global_tags - if self._test_tags is None: - return set(global_new) - test_new, test_gone = self._test_tags - return global_new.difference(test_gone).union(test_new) - - def _get_current_scope(self): - if self._test_tags: - return self._test_tags - return self._global_tags - - def _flush_current_scope(self, tag_receiver): - new_tags, gone_tags = self._get_current_scope() - if new_tags or gone_tags: - tag_receiver.tags(new_tags, gone_tags) - if self._test_tags: - self._test_tags = set(), set() - else: - self._global_tags = set(), set() - - def startTestRun(self): - self._clear_tags() - - def startTest(self, test): - self._test_tags = set(), set() - - def stopTest(self, test): - self._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. - """ - current_new_tags, current_gone_tags = self._get_current_scope() - current_new_tags.update(new_tags) - current_new_tags.difference_update(gone_tags) - current_gone_tags.update(gone_tags) - current_gone_tags.difference_update(new_tags) - - -class TagCollapsingDecorator(HookedTestResultDecorator, TagsMixin): - """Collapses many 'tags' calls into one where possible.""" - - def __init__(self, result): - super(TagCollapsingDecorator, self).__init__(result) - self._clear_tags() - - def _before_event(self): - self._flush_current_scope(self.decorated) - - def tags(self, new_tags, gone_tags): - TagsMixin.tags(self, new_tags, gone_tags) - - -class TimeCollapsingDecorator(HookedTestResultDecorator): - """Only pass on the first and last of a consecutive sequence of times.""" - - def __init__(self, decorated): - super(TimeCollapsingDecorator, self).__init__(decorated) - self._last_received_time = None - self._last_sent_time = None - - def _before_event(self): - if self._last_received_time is None: - return - if self._last_received_time != self._last_sent_time: - self.decorated.time(self._last_received_time) - self._last_sent_time = self._last_received_time - self._last_received_time = None - - def time(self, a_time): - # Don't upcall, because we don't want to call _before_event, it's only - # for non-time events. - if self._last_received_time is None: - self.decorated.time(a_time) - self._last_sent_time = a_time - self._last_received_time = a_time - - -def and_predicates(predicates): - """Return a predicate that is true iff all predicates are true.""" - # XXX: Should probably be in testtools to be better used by matchers. jml - return lambda *args, **kwargs: all(p(*args, **kwargs) for p in predicates) - - -def make_tag_filter(with_tags, without_tags): - """Make a callback that checks tests against tags.""" - - with_tags = with_tags and set(with_tags) or None - without_tags = without_tags and set(without_tags) or None - - def check_tags(test, outcome, err, details, tags): - if with_tags and not with_tags <= tags: - return False - if without_tags and bool(without_tags & tags): - return False - return True - - return check_tags - - -class _PredicateFilter(TestResultDecorator, TagsMixin): - - def __init__(self, result, predicate): - super(_PredicateFilter, self).__init__(result) - self._clear_tags() - self.decorated = TimeCollapsingDecorator( - TagCollapsingDecorator(self.decorated)) - self._predicate = 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 - # Calls to this result that we don't know whether to forward on yet. - self._buffered_calls = [] - - def filter_predicate(self, test, outcome, error, details): - return self._predicate( - test, outcome, error, details, self._get_active_tags()) - - def addError(self, test, err=None, details=None): - if (self.filter_predicate(test, 'error', err, details)): - self._buffered_calls.append( - ('addError', [test, err], {'details': details})) - else: - self._filtered() - - def addFailure(self, test, err=None, details=None): - if (self.filter_predicate(test, 'failure', err, details)): - self._buffered_calls.append( - ('addFailure', [test, err], {'details': details})) - else: - self._filtered() - - def addSkip(self, test, reason=None, details=None): - if (self.filter_predicate(test, 'skip', reason, details)): - self._buffered_calls.append( - ('addSkip', [test, reason], {'details': details})) - else: - self._filtered() - - def addExpectedFailure(self, test, err=None, details=None): - if self.filter_predicate(test, 'expectedfailure', err, details): - self._buffered_calls.append( - ('addExpectedFailure', [test, err], {'details': details})) - else: - self._filtered() - - def addUnexpectedSuccess(self, test, details=None): - self._buffered_calls.append( - ('addUnexpectedSuccess', [test], {'details': details})) - - def addSuccess(self, test, details=None): - if (self.filter_predicate(test, 'success', None, details)): - self._buffered_calls.append( - ('addSuccess', [test], {'details': details})) - else: - self._filtered() - - 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. - """ - TagsMixin.startTest(self, test) - self._current_test = test - self._current_test_filtered = False - self._buffered_calls.append(('startTest', [test], {})) - - 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: - for method, args, kwargs in self._buffered_calls: - getattr(self.decorated, method)(*args, **kwargs) - self.decorated.stopTest(test) - self._current_test = None - self._current_test_filtered = None - self._buffered_calls = [] - TagsMixin.stopTest(self, test) - - def tags(self, new_tags, gone_tags): - TagsMixin.tags(self, new_tags, gone_tags) - if self._current_test is not None: - self._buffered_calls.append(('tags', [new_tags, gone_tags], {})) - else: - return super(_PredicateFilter, self).tags(new_tags, gone_tags) - - def time(self, a_time): - return self.decorated.time(a_time) - - def id_to_orig_id(self, id): - if id.startswith("subunit.RemotedTestCase."): - return id[len("subunit.RemotedTestCase."):] - return id - - -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_xfail=False, - filter_predicate=None, fixup_expected_failures=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_xfail: Filter out expected failure tests. - :param filter_predicate: A callable taking (test, outcome, err, - details, tags) 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'. tags is new in 0.0.8; 0.0.7 filters - are still supported but should be updated to accept the tags - parameter for efficiency. - :param fixup_expected_failures: Set of test ids to consider known - failing. - """ - predicates = [] - if filter_error: - predicates.append( - lambda t, outcome, e, d, tags: outcome != 'error') - if filter_failure: - predicates.append( - lambda t, outcome, e, d, tags: outcome != 'failure') - if filter_success: - predicates.append( - lambda t, outcome, e, d, tags: outcome != 'success') - if filter_skip: - predicates.append( - lambda t, outcome, e, d, tags: outcome != 'skip') - if filter_xfail: - predicates.append( - lambda t, outcome, e, d, tags: outcome != 'expectedfailure') - if filter_predicate is not None: - def compat(test, outcome, error, details, tags): - # 0.0.7 and earlier did not support the 'tags' parameter. - try: - return filter_predicate( - test, outcome, error, details, tags) - except TypeError: - return filter_predicate(test, outcome, error, details) - predicates.append(compat) - predicate = and_predicates(predicates) - super(TestResultFilter, self).__init__( - _PredicateFilter(result, predicate)) - if fixup_expected_failures is None: - self._fixup_expected_failures = frozenset() - else: - self._fixup_expected_failures = fixup_expected_failures - - def addError(self, test, err=None, details=None): - if self._failure_expected(test): - self.addExpectedFailure(test, err=err, details=details) - else: - super(TestResultFilter, self).addError( - test, err=err, details=details) - - def addFailure(self, test, err=None, details=None): - if self._failure_expected(test): - self.addExpectedFailure(test, err=err, details=details) - else: - super(TestResultFilter, self).addFailure( - test, err=err, details=details) - - def addSuccess(self, test, details=None): - if self._failure_expected(test): - self.addUnexpectedSuccess(test, details=details) - else: - super(TestResultFilter, self).addSuccess(test, details=details) - - def _failure_expected(self, test): - return (test.id() in self._fixup_expected_failures) - - -class TestIdPrintingResult(testtools.TestResult): - - def __init__(self, stream, show_times=False): - """Create a FilterResult object outputting to stream.""" - super(TestIdPrintingResult, self).__init__() - self._stream = stream - self.failed_tests = 0 - self.__time = None - self.show_times = show_times - self._test = None - self._test_duration = 0 - - def addError(self, test, err): - self.failed_tests += 1 - self._test = test - - def addFailure(self, test, err): - self.failed_tests += 1 - self._test = test - - def addSuccess(self, test): - self._test = test - - def addSkip(self, test, reason=None, details=None): - self._test = test - - def addUnexpectedSuccess(self, test, details=None): - self.failed_tests += 1 - self._test = test - - def addExpectedFailure(self, test, err=None, details=None): - self._test = test - - def reportTest(self, test, duration): - if self.show_times: - seconds = duration.seconds - seconds += duration.days * 3600 * 24 - seconds += duration.microseconds / 1000000.0 - self._stream.write(test.id() + ' %0.3f\n' % seconds) - else: - self._stream.write(test.id() + '\n') - - def startTest(self, test): - self._start_time = self._time() - - def stopTest(self, test): - test_duration = self._time() - self._start_time - self.reportTest(self._test, test_duration) - - def time(self, time): - self.__time = time - - def _time(self): - return self.__time - - def wasSuccessful(self): - "Tells whether or not this result was a success" - return self.failed_tests == 0 - - -class TestByTestResult(testtools.TestResult): - """Call something every time a test completes.""" - -# XXX: In testtools since lp:testtools r249. Once that's released, just -# import that. - - def __init__(self, on_test): - """Construct a ``TestByTestResult``. - - :param on_test: A callable that take a test case, a status (one of - "success", "failure", "error", "skip", or "xfail"), a start time - (a ``datetime`` with timezone), a stop time, an iterable of tags, - and a details dict. Is called at the end of each test (i.e. on - ``stopTest``) with the accumulated values for that test. - """ - super(TestByTestResult, self).__init__() - self._on_test = on_test - - def startTest(self, test): - super(TestByTestResult, self).startTest(test) - self._start_time = self._now() - # There's no supported (i.e. tested) behaviour that relies on these - # being set, but it makes me more comfortable all the same. -- jml - self._status = None - self._details = None - self._stop_time = None - - def stopTest(self, test): - self._stop_time = self._now() - super(TestByTestResult, self).stopTest(test) - self._on_test( - test=test, - status=self._status, - start_time=self._start_time, - stop_time=self._stop_time, - # current_tags is new in testtools 0.9.13. - tags=getattr(self, 'current_tags', None), - details=self._details) - - def _err_to_details(self, test, err, details): - if details: - return details - return {'traceback': TracebackContent(err, test)} - - def addSuccess(self, test, details=None): - super(TestByTestResult, self).addSuccess(test) - self._status = 'success' - self._details = details - - def addFailure(self, test, err=None, details=None): - super(TestByTestResult, self).addFailure(test, err, details) - self._status = 'failure' - self._details = self._err_to_details(test, err, details) - - def addError(self, test, err=None, details=None): - super(TestByTestResult, self).addError(test, err, details) - self._status = 'error' - self._details = self._err_to_details(test, err, details) - - def addSkip(self, test, reason=None, details=None): - super(TestByTestResult, self).addSkip(test, reason, details) - self._status = 'skip' - if details is None: - details = {'reason': text_content(reason)} - elif reason: - # XXX: What if details already has 'reason' key? - details['reason'] = text_content(reason) - self._details = details - - def addExpectedFailure(self, test, err=None, details=None): - super(TestByTestResult, self).addExpectedFailure(test, err, details) - self._status = 'xfail' - self._details = self._err_to_details(test, err, details) - - def addUnexpectedSuccess(self, test, details=None): - super(TestByTestResult, self).addUnexpectedSuccess(test, details) - self._status = 'success' - self._details = details - - -class CsvResult(TestByTestResult): - - def __init__(self, stream): - super(CsvResult, self).__init__(self._on_test) - self._write_row = csv.writer(stream).writerow - - def _on_test(self, test, status, start_time, stop_time, tags, details): - self._write_row([test.id(), status, start_time, stop_time]) - - def startTestRun(self): - super(CsvResult, self).startTestRun() - self._write_row(['test', 'status', 'start_time', 'stop_time']) diff --git a/lib/subunit/python/subunit/tests/TestUtil.py b/lib/subunit/python/subunit/tests/TestUtil.py deleted file mode 100644 index 39d901e0a9..0000000000 --- a/lib/subunit/python/subunit/tests/TestUtil.py +++ /dev/null @@ -1,80 +0,0 @@ -# Copyright (c) 2004 Canonical Limited -# Author: Robert Collins <robert.collins@canonical.com> -# -# 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 -# - -import sys -import logging -import unittest - - -class LogCollector(logging.Handler): - def __init__(self): - logging.Handler.__init__(self) - self.records=[] - def emit(self, record): - self.records.append(record.getMessage()) - - -def makeCollectingLogger(): - """I make a logger instance that collects its logs for programmatic analysis - -> (logger, collector)""" - logger=logging.Logger("collector") - handler=LogCollector() - handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) - logger.addHandler(handler) - return logger, handler - - -def visitTests(suite, visitor): - """A foreign method for visiting the tests in a test suite.""" - for test in suite._tests: - #Abusing types to avoid monkey patching unittest.TestCase. - # Maybe that would be better? - try: - test.visit(visitor) - except AttributeError: - if isinstance(test, unittest.TestCase): - visitor.visitCase(test) - elif isinstance(test, unittest.TestSuite): - visitor.visitSuite(test) - visitTests(test, visitor) - else: - print ("unvisitable non-unittest.TestCase element %r (%r)" % (test, test.__class__)) - - -class TestSuite(unittest.TestSuite): - """I am an extended TestSuite with a visitor interface. - This is primarily to allow filtering of tests - and suites or - more in the future. An iterator of just tests wouldn't scale...""" - - def visit(self, visitor): - """visit the composite. Visiting is depth-first. - current callbacks are visitSuite and visitCase.""" - visitor.visitSuite(self) - visitTests(self, visitor) - - -class TestLoader(unittest.TestLoader): - """Custome TestLoader to set the right TestSuite class.""" - suiteClass = TestSuite - -class TestVisitor(object): - """A visitor for Tests""" - def visitSuite(self, aTestSuite): - pass - def visitCase(self, aTestCase): - pass diff --git a/lib/subunit/python/subunit/tests/__init__.py b/lib/subunit/python/subunit/tests/__init__.py deleted file mode 100644 index e0e1eb1b04..0000000000 --- a/lib/subunit/python/subunit/tests/__init__.py +++ /dev/null @@ -1,43 +0,0 @@ -# -# 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 subunit.tests import ( - TestUtil, - test_chunked, - test_details, - test_progress_model, - test_run, - 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()) - result.addTest(test_run.test_suite()) - return result diff --git a/lib/subunit/python/subunit/tests/sample-script.py b/lib/subunit/python/subunit/tests/sample-script.py deleted file mode 100755 index 91838f6d6f..0000000000 --- a/lib/subunit/python/subunit/tests/sample-script.py +++ /dev/null @@ -1,21 +0,0 @@ -#!/usr/bin/env python -import sys -if sys.platform == "win32": - import msvcrt, os - msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY) -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") -print("failure bing crosby [") -print("foo.c:53:ERROR invalid state") -print("]") -print("test an error") -print("error an error") -sys.exit(0) diff --git a/lib/subunit/python/subunit/tests/sample-two-script.py b/lib/subunit/python/subunit/tests/sample-two-script.py deleted file mode 100755 index fc73dfc409..0000000000 --- a/lib/subunit/python/subunit/tests/sample-two-script.py +++ /dev/null @@ -1,7 +0,0 @@ -#!/usr/bin/env python -import sys -print("test old mcdonald") -print("success old mcdonald") -print("test bing crosby") -print("success bing crosby") -sys.exit(0) diff --git a/lib/subunit/python/subunit/tests/test_chunked.py b/lib/subunit/python/subunit/tests/test_chunked.py deleted file mode 100644 index e0742f1af3..0000000000 --- a/lib/subunit/python/subunit/tests/test_chunked.py +++ /dev/null @@ -1,152 +0,0 @@ -# -# subunit: extensions to python unittest to get test results from subprocesses. -# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net> -# Copyright (C) 2011 Martin Pool <mbp@sourcefrog.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 - -from testtools.compat import _b, BytesIO - -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 = BytesIO() - 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(_b('2\r\na'))) - self.assertRaises(ValueError, self.decoder.close) - - def test_close_body_buffered_data_errors(self): - self.assertEqual(None, self.decoder.write(_b('2\r'))) - self.assertRaises(ValueError, self.decoder.close) - - def test_close_after_finished_stream_safe(self): - self.assertEqual(None, self.decoder.write(_b('2\r\nab'))) - self.assertEqual(_b(''), self.decoder.write(_b('0\r\n'))) - self.decoder.close() - - def test_decode_nothing(self): - self.assertEqual(_b(''), self.decoder.write(_b('0\r\n'))) - self.assertEqual(_b(''), self.output.getvalue()) - - def test_decode_serialised_form(self): - self.assertEqual(None, self.decoder.write(_b("F\r\n"))) - self.assertEqual(None, self.decoder.write(_b("serialised\n"))) - self.assertEqual(_b(''), self.decoder.write(_b("form0\r\n"))) - - def test_decode_short(self): - self.assertEqual(_b(''), self.decoder.write(_b('3\r\nabc0\r\n'))) - self.assertEqual(_b('abc'), self.output.getvalue()) - - def test_decode_combines_short(self): - self.assertEqual(_b(''), self.decoder.write(_b('6\r\nabcdef0\r\n'))) - self.assertEqual(_b('abcdef'), self.output.getvalue()) - - def test_decode_excess_bytes_from_write(self): - self.assertEqual(_b('1234'), self.decoder.write(_b('3\r\nabc0\r\n1234'))) - self.assertEqual(_b('abc'), self.output.getvalue()) - - def test_decode_write_after_finished_errors(self): - self.assertEqual(_b('1234'), self.decoder.write(_b('3\r\nabc0\r\n1234'))) - self.assertRaises(ValueError, self.decoder.write, _b('')) - - def test_decode_hex(self): - self.assertEqual(_b(''), self.decoder.write(_b('A\r\n12345678900\r\n'))) - self.assertEqual(_b('1234567890'), self.output.getvalue()) - - def test_decode_long_ranges(self): - self.assertEqual(None, self.decoder.write(_b('10000\r\n'))) - self.assertEqual(None, self.decoder.write(_b('1' * 65536))) - self.assertEqual(None, self.decoder.write(_b('10000\r\n'))) - self.assertEqual(None, self.decoder.write(_b('2' * 65536))) - self.assertEqual(_b(''), self.decoder.write(_b('0\r\n'))) - self.assertEqual(_b('1' * 65536 + '2' * 65536), self.output.getvalue()) - - def test_decode_newline_nonstrict(self): - """Tolerate chunk markers with no CR character.""" - # From <http://pad.lv/505078> - self.decoder = subunit.chunked.Decoder(self.output, strict=False) - self.assertEqual(None, self.decoder.write(_b('a\n'))) - self.assertEqual(None, self.decoder.write(_b('abcdeabcde'))) - self.assertEqual(_b(''), self.decoder.write(_b('0\n'))) - self.assertEqual(_b('abcdeabcde'), self.output.getvalue()) - - def test_decode_strict_newline_only(self): - """Reject chunk markers with no CR character in strict mode.""" - # From <http://pad.lv/505078> - self.assertRaises(ValueError, - self.decoder.write, _b('a\n')) - - def test_decode_strict_multiple_crs(self): - self.assertRaises(ValueError, - self.decoder.write, _b('a\r\r\n')) - - def test_decode_short_header(self): - self.assertRaises(ValueError, - self.decoder.write, _b('\n')) - - -class TestEncode(unittest.TestCase): - - def setUp(self): - unittest.TestCase.setUp(self) - self.output = BytesIO() - self.encoder = subunit.chunked.Encoder(self.output) - - def test_encode_nothing(self): - self.encoder.close() - self.assertEqual(_b('0\r\n'), self.output.getvalue()) - - def test_encode_empty(self): - self.encoder.write(_b('')) - self.encoder.close() - self.assertEqual(_b('0\r\n'), self.output.getvalue()) - - def test_encode_short(self): - self.encoder.write(_b('abc')) - self.encoder.close() - self.assertEqual(_b('3\r\nabc0\r\n'), self.output.getvalue()) - - def test_encode_combines_short(self): - self.encoder.write(_b('abc')) - self.encoder.write(_b('def')) - self.encoder.close() - self.assertEqual(_b('6\r\nabcdef0\r\n'), self.output.getvalue()) - - def test_encode_over_9_is_in_hex(self): - self.encoder.write(_b('1234567890')) - self.encoder.close() - self.assertEqual(_b('A\r\n12345678900\r\n'), self.output.getvalue()) - - def test_encode_long_ranges_not_combined(self): - self.encoder.write(_b('1' * 65536)) - self.encoder.write(_b('2' * 65536)) - self.encoder.close() - self.assertEqual(_b('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 deleted file mode 100644 index 746aa041e5..0000000000 --- a/lib/subunit/python/subunit/tests/test_details.py +++ /dev/null @@ -1,112 +0,0 @@ -# -# 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. -# - -import unittest - -from testtools.compat import _b, StringIO - -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(_b("foo\n")) - parser.lineReceived(_b("bar\n")) - self.assertEqual(_b("foo\nbar\n"), parser._message) - - def test_lineReceived_escaped_bracket(self): - parser = details.SimpleDetailsParser(None) - parser.lineReceived(_b("foo\n")) - parser.lineReceived(_b(" ]are\n")) - parser.lineReceived(_b("bar\n")) - self.assertEqual(_b("foo\n]are\nbar\n"), parser._message) - - def test_get_message(self): - parser = details.SimpleDetailsParser(None) - self.assertEqual(_b(""), parser.get_message()) - - def test_get_details(self): - parser = details.SimpleDetailsParser(None) - traceback = "" - expected = {} - expected['traceback'] = content.Content( - content_type.ContentType("text", "x-traceback", - {'charset': 'utf8'}), - lambda:[_b("")]) - found = parser.get_details() - self.assertEqual(expected.keys(), found.keys()) - self.assertEqual(expected['traceback'].content_type, - found['traceback'].content_type) - self.assertEqual(_b('').join(expected['traceback'].iter_bytes()), - _b('').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:[_b("")]) - 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:[_b("")]) - 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(_b("Content-Type: text/plain\n")) - parser.lineReceived(_b("something\n")) - parser.lineReceived(_b("F\r\n")) - parser.lineReceived(_b("serialised\n")) - parser.lineReceived(_b("form0\r\n")) - expected = {} - expected['something'] = content.Content( - content_type.ContentType("text", "plain"), - lambda:[_b("serialised\nform")]) - found = parser.get_details() - self.assertEqual(expected.keys(), found.keys()) - self.assertEqual(expected['something'].content_type, - found['something'].content_type) - self.assertEqual(_b('').join(expected['something'].iter_bytes()), - _b('').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 deleted file mode 100644 index 76200c6107..0000000000 --- a/lib/subunit/python/subunit/tests/test_progress_model.py +++ /dev/null @@ -1,118 +0,0 @@ -# -# 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_run.py b/lib/subunit/python/subunit/tests/test_run.py deleted file mode 100644 index 10519ed086..0000000000 --- a/lib/subunit/python/subunit/tests/test_run.py +++ /dev/null @@ -1,52 +0,0 @@ -# -# subunit: extensions to python unittest to get test results from subprocesses. -# Copyright (C) 2011 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 testtools.compat import BytesIO -import unittest - -from testtools import PlaceHolder - -import subunit -from subunit.run import SubunitTestRunner - - -def test_suite(): - loader = subunit.tests.TestUtil.TestLoader() - result = loader.loadTestsFromName(__name__) - return result - - -class TimeCollectingTestResult(unittest.TestResult): - - def __init__(self, *args, **kwargs): - super(TimeCollectingTestResult, self).__init__(*args, **kwargs) - self.time_called = [] - - def time(self, a_time): - self.time_called.append(a_time) - - -class TestSubunitTestRunner(unittest.TestCase): - - def test_includes_timing_output(self): - io = BytesIO() - runner = SubunitTestRunner(stream=io) - test = PlaceHolder('name') - runner.run(test) - client = TimeCollectingTestResult() - io.seek(0) - subunit.TestProtocolServer(client).readFrom(io) - self.assertTrue(len(client.time_called) > 0) diff --git a/lib/subunit/python/subunit/tests/test_subunit_filter.py b/lib/subunit/python/subunit/tests/test_subunit_filter.py deleted file mode 100644 index 33b924824d..0000000000 --- a/lib/subunit/python/subunit/tests/test_subunit_filter.py +++ /dev/null @@ -1,370 +0,0 @@ -# -# 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.""" - -from datetime import datetime -import os -import subprocess -import sys -from subunit import iso8601 -import unittest - -from testtools import TestCase -from testtools.compat import _b, BytesIO -from testtools.testresult.doubles import ExtendedTestResult - -import subunit -from subunit.test_results import make_tag_filter, TestResultFilter - - -class TestTestResultFilter(TestCase): - """Test for TestResultFilter, a TestResult object which filters tests.""" - - # 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. - example_subunit_stream = _b("""\ -tags: global -test passed -success passed -test failed -tags: local -failure failed -test error -error error [ -error details -] -test skipped -skip skipped -test todo -xfail todo -""") - - def run_tests(self, result_filter, input_stream=None): - """Run tests through the given filter. - - :param result_filter: A filtering TestResult object. - :param input_stream: Bytes of subunit stream data. If not provided, - uses TestTestResultFilter.example_subunit_stream. - """ - if input_stream is None: - input_stream = self.example_subunit_stream - test = subunit.ProtocolTestCase(BytesIO(input_stream)) - test.run(result_filter) - - def test_default(self): - """The default is to exclude success and include everything else.""" - filtered_result = unittest.TestResult() - result_filter = TestResultFilter(filtered_result) - self.run_tests(result_filter) - # skips are seen as success by default python TestResult. - self.assertEqual(['error'], - [error[0].id() for error in filtered_result.errors]) - self.assertEqual(['failed'], - [failure[0].id() for failure in - filtered_result.failures]) - self.assertEqual(4, filtered_result.testsRun) - - def test_tag_filter(self): - tag_filter = make_tag_filter(['global'], ['local']) - result = ExtendedTestResult() - result_filter = TestResultFilter( - result, filter_success=False, filter_predicate=tag_filter) - self.run_tests(result_filter) - tests_included = [ - event[1] for event in result._events if event[0] == 'startTest'] - tests_expected = list(map( - subunit.RemotedTestCase, - ['passed', 'error', 'skipped', 'todo'])) - self.assertEquals(tests_expected, tests_included) - - def test_tags_tracked_correctly(self): - tag_filter = make_tag_filter(['a'], []) - result = ExtendedTestResult() - result_filter = TestResultFilter( - result, filter_success=False, filter_predicate=tag_filter) - input_stream = _b( - "test: foo\n" - "tags: a\n" - "successful: foo\n" - "test: bar\n" - "successful: bar\n") - self.run_tests(result_filter, input_stream) - foo = subunit.RemotedTestCase('foo') - self.assertEquals( - [('startTest', foo), - ('tags', set(['a']), set()), - ('addSuccess', foo), - ('stopTest', foo), - ], - result._events) - - def test_exclude_errors(self): - filtered_result = unittest.TestResult() - result_filter = TestResultFilter(filtered_result, filter_error=True) - self.run_tests(result_filter) - # skips are seen as errors by default python TestResult. - self.assertEqual([], filtered_result.errors) - self.assertEqual(['failed'], - [failure[0].id() for failure in - filtered_result.failures]) - self.assertEqual(3, filtered_result.testsRun) - - def test_fixup_expected_failures(self): - filtered_result = unittest.TestResult() - result_filter = TestResultFilter(filtered_result, - fixup_expected_failures=set(["failed"])) - self.run_tests(result_filter) - self.assertEqual(['failed', 'todo'], - [failure[0].id() for failure in filtered_result.expectedFailures]) - self.assertEqual([], filtered_result.failures) - self.assertEqual(4, filtered_result.testsRun) - - def test_fixup_expected_errors(self): - filtered_result = unittest.TestResult() - result_filter = TestResultFilter(filtered_result, - fixup_expected_failures=set(["error"])) - self.run_tests(result_filter) - self.assertEqual(['error', 'todo'], - [failure[0].id() for failure in filtered_result.expectedFailures]) - self.assertEqual([], filtered_result.errors) - self.assertEqual(4, filtered_result.testsRun) - - def test_fixup_unexpected_success(self): - filtered_result = unittest.TestResult() - result_filter = TestResultFilter(filtered_result, filter_success=False, - fixup_expected_failures=set(["passed"])) - self.run_tests(result_filter) - self.assertEqual(['passed'], - [passed.id() for passed in filtered_result.unexpectedSuccesses]) - self.assertEqual(5, filtered_result.testsRun) - - def test_exclude_failure(self): - filtered_result = unittest.TestResult() - result_filter = TestResultFilter(filtered_result, filter_failure=True) - self.run_tests(result_filter) - self.assertEqual(['error'], - [error[0].id() for error in filtered_result.errors]) - self.assertEqual([], - [failure[0].id() for failure in - filtered_result.failures]) - self.assertEqual(3, filtered_result.testsRun) - - def test_exclude_skips(self): - filtered_result = subunit.TestResultStats(None) - result_filter = TestResultFilter(filtered_result, filter_skip=True) - self.run_tests(result_filter) - self.assertEqual(0, filtered_result.skipped_tests) - self.assertEqual(2, filtered_result.failed_tests) - self.assertEqual(3, filtered_result.testsRun) - - def test_include_success(self): - """Successes can be included if requested.""" - filtered_result = unittest.TestResult() - result_filter = TestResultFilter(filtered_result, - filter_success=False) - self.run_tests(result_filter) - self.assertEqual(['error'], - [error[0].id() for error in filtered_result.errors]) - self.assertEqual(['failed'], - [failure[0].id() for failure in - filtered_result.failures]) - self.assertEqual(5, filtered_result.testsRun) - - def test_filter_predicate(self): - """You can filter by predicate callbacks""" - # 0.0.7 and earlier did not support the 'tags' parameter, so we need - # to test that we still support behaviour without it. - filtered_result = unittest.TestResult() - def filter_cb(test, outcome, err, details): - return outcome == 'success' - result_filter = TestResultFilter(filtered_result, - filter_predicate=filter_cb, - filter_success=False) - self.run_tests(result_filter) - # Only success should pass - self.assertEqual(1, filtered_result.testsRun) - - def test_filter_predicate_with_tags(self): - """You can filter by predicate callbacks that accept tags""" - filtered_result = unittest.TestResult() - def filter_cb(test, outcome, err, details, tags): - return outcome == 'success' - result_filter = TestResultFilter(filtered_result, - filter_predicate=filter_cb, - filter_success=False) - self.run_tests(result_filter) - # Only success should pass - self.assertEqual(1, filtered_result.testsRun) - - def test_time_ordering_preserved(self): - # Passing a subunit stream through TestResultFilter preserves the - # relative ordering of 'time' directives and any other subunit - # directives that are still included. - date_a = datetime(year=2000, month=1, day=1, tzinfo=iso8601.UTC) - date_b = datetime(year=2000, month=1, day=2, tzinfo=iso8601.UTC) - date_c = datetime(year=2000, month=1, day=3, tzinfo=iso8601.UTC) - subunit_stream = _b('\n'.join([ - "time: %s", - "test: foo", - "time: %s", - "error: foo", - "time: %s", - ""]) % (date_a, date_b, date_c)) - result = ExtendedTestResult() - result_filter = TestResultFilter(result) - self.run_tests(result_filter, subunit_stream) - foo = subunit.RemotedTestCase('foo') - self.maxDiff = None - self.assertEqual( - [('time', date_a), - ('time', date_b), - ('startTest', foo), - ('addError', foo, {}), - ('stopTest', foo), - ('time', date_c)], result._events) - - def test_time_passes_through_filtered_tests(self): - # Passing a subunit stream through TestResultFilter preserves 'time' - # directives even if a specific test is filtered out. - date_a = datetime(year=2000, month=1, day=1, tzinfo=iso8601.UTC) - date_b = datetime(year=2000, month=1, day=2, tzinfo=iso8601.UTC) - date_c = datetime(year=2000, month=1, day=3, tzinfo=iso8601.UTC) - subunit_stream = _b('\n'.join([ - "time: %s", - "test: foo", - "time: %s", - "success: foo", - "time: %s", - ""]) % (date_a, date_b, date_c)) - result = ExtendedTestResult() - result_filter = TestResultFilter(result) - result_filter.startTestRun() - self.run_tests(result_filter, subunit_stream) - result_filter.stopTestRun() - foo = subunit.RemotedTestCase('foo') - self.maxDiff = None - self.assertEqual( - [('startTestRun',), - ('time', date_a), - ('time', date_c), - ('stopTestRun',),], result._events) - - def test_skip_preserved(self): - subunit_stream = _b('\n'.join([ - "test: foo", - "skip: foo", - ""])) - result = ExtendedTestResult() - result_filter = TestResultFilter(result) - self.run_tests(result_filter, subunit_stream) - foo = subunit.RemotedTestCase('foo') - self.assertEquals( - [('startTest', foo), - ('addSkip', foo, {}), - ('stopTest', foo), ], result._events) - - if sys.version_info < (2, 7): - # These tests require Python >=2.7. - del test_fixup_expected_failures, test_fixup_expected_errors, test_fixup_unexpected_success - - -class TestFilterCommand(TestCase): - - example_subunit_stream = _b("""\ -tags: global -test passed -success passed -test failed -tags: local -failure failed -test error -error error [ -error details -] -test skipped -skip skipped -test todo -xfail todo -""") - - def run_command(self, args, stream): - root = os.path.dirname( - os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) - script_path = os.path.join(root, 'filters', 'subunit-filter') - command = [sys.executable, script_path] + list(args) - ps = subprocess.Popen( - command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - out, err = ps.communicate(stream) - if ps.returncode != 0: - raise RuntimeError("%s failed: %s" % (command, err)) - return out - - def to_events(self, stream): - test = subunit.ProtocolTestCase(BytesIO(stream)) - result = ExtendedTestResult() - test.run(result) - return result._events - - def test_default(self): - output = self.run_command([], _b( - "test: foo\n" - "skip: foo\n" - )) - events = self.to_events(output) - foo = subunit.RemotedTestCase('foo') - self.assertEqual( - [('startTest', foo), - ('addSkip', foo, {}), - ('stopTest', foo)], - events) - - def test_tags(self): - output = self.run_command(['-s', '--with-tag', 'a'], _b( - "tags: a\n" - "test: foo\n" - "success: foo\n" - "tags: -a\n" - "test: bar\n" - "success: bar\n" - "test: baz\n" - "tags: a\n" - "success: baz\n" - )) - events = self.to_events(output) - foo = subunit.RemotedTestCase('foo') - baz = subunit.RemotedTestCase('baz') - self.assertEqual( - [('tags', set(['a']), set()), - ('startTest', foo), - ('addSuccess', foo), - ('stopTest', foo), - ('tags', set(), set(['a'])), - ('startTest', baz), - ('tags', set(['a']), set()), - ('addSuccess', baz), - ('stopTest', baz), - ], - events) - - -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 deleted file mode 100644 index 6fd3301060..0000000000 --- a/lib/subunit/python/subunit/tests/test_subunit_stats.py +++ /dev/null @@ -1,84 +0,0 @@ -# -# 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 testtools.compat import _b, BytesIO, 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 = BytesIO() - 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(_b("""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 deleted file mode 100644 index c98506a737..0000000000 --- a/lib/subunit/python/subunit/tests/test_subunit_tags.py +++ /dev/null @@ -1,69 +0,0 @@ -# -# 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 testtools.compat 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 deleted file mode 100644 index 11bc1916b3..0000000000 --- a/lib/subunit/python/subunit/tests/test_tap2subunit.py +++ /dev/null @@ -1,445 +0,0 @@ -# -# 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 testtools.compat import StringIO - -import subunit - - -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_skip_number_comment_lowercase(self): - self.tap.write("ok 1 # skip no samba environment available, skipping compilation\n") - self.tap.seek(0) - result = subunit.TAP2SubUnit(self.tap, self.subunit) - self.assertEqual(0, result) - self.assertEqual([ - "test test 1", - "skip test 1 [", - "no samba environment available, skipping compilation", - "]" - ], - self.subunit.getvalue().splitlines()) - - 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 deleted file mode 100644 index 7831ba16cd..0000000000 --- a/lib/subunit/python/subunit/tests/test_test_protocol.py +++ /dev/null @@ -1,1337 +0,0 @@ -# -# 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. -# - -import datetime -import unittest -import os - -from testtools import PlaceHolder, skipIf, TestCase, TestResult -from testtools.compat import _b, _u, BytesIO -from testtools.content import Content, TracebackContent, text_content -from testtools.content_type import ContentType -try: - from testtools.testresult.doubles import ( - Python26TestResult, - Python27TestResult, - ExtendedTestResult, - ) -except ImportError: - from testtools.tests.helpers import ( - Python26TestResult, - Python27TestResult, - ExtendedTestResult, - ) - -import subunit -from subunit import _remote_exception_str, _remote_exception_str_chunked -import subunit.iso8601 as iso8601 - - -def details_to_str(details): - return TestResult()._err_details_to_string(None, details=details) - - -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 = BytesIO() - protocol = subunit.TestProtocolServer(client, forward_stream=out) - pipe = BytesIO(_b("test old mcdonald\n" - "success old mcdonald\n")) - protocol.readFrom(pipe) - self.assertEqual(client.testsRun, 1) - self.assertEqual(pipe.getvalue(), out.getvalue()) - - def test_not_command(self): - client = unittest.TestResult() - out = BytesIO() - protocol = subunit.TestProtocolServer(client, - stream=subunit.DiscardStream(), forward_stream=out) - pipe = BytesIO(_b("success old mcdonald\n")) - protocol.readFrom(pipe) - self.assertEqual(client.testsRun, 0) - self.assertEqual(_b(""), out.getvalue()) - - -class TestTestProtocolServerPipe(unittest.TestCase): - - def test_story(self): - client = unittest.TestResult() - protocol = subunit.TestProtocolServer(client) - traceback = "foo.c:53:ERROR invalid state\n" - pipe = BytesIO(_b("test old mcdonald\n" - "success old mcdonald\n" - "test bing crosby\n" - "failure bing crosby [\n" - + traceback + - "]\n" - "test an error\n" - "error an error\n")) - protocol.readFrom(pipe) - bing = subunit.RemotedTestCase("bing crosby") - an_error = subunit.RemotedTestCase("an error") - self.assertEqual(client.errors, - [(an_error, _remote_exception_str + '\n')]) - self.assertEqual( - client.failures, - [(bing, _remote_exception_str + ": " - + details_to_str({'traceback': text_content(traceback)}) + "\n")]) - self.assertEqual(client.testsRun, 3) - - def test_non_test_characters_forwarded_immediately(self): - pass - - -class TestTestProtocolServerStartTest(unittest.TestCase): - - def setUp(self): - self.client = Python26TestResult() - self.stream = BytesIO() - self.protocol = subunit.TestProtocolServer(self.client, self.stream) - - def test_start_test(self): - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.assertEqual(self.client._events, - [('startTest', subunit.RemotedTestCase("old mcdonald"))]) - - def test_start_testing(self): - self.protocol.lineReceived(_b("testing old mcdonald\n")) - self.assertEqual(self.client._events, - [('startTest', subunit.RemotedTestCase("old mcdonald"))]) - - def test_start_test_colon(self): - self.protocol.lineReceived(_b("test: old mcdonald\n")) - self.assertEqual(self.client._events, - [('startTest', subunit.RemotedTestCase("old mcdonald"))]) - - def test_indented_test_colon_ignored(self): - ignored_line = _b(" test: old mcdonald\n") - self.protocol.lineReceived(ignored_line) - self.assertEqual([], self.client._events) - self.assertEqual(self.stream.getvalue(), ignored_line) - - def test_start_testing_colon(self): - self.protocol.lineReceived(_b("testing: old mcdonald\n")) - self.assertEqual(self.client._events, - [('startTest', subunit.RemotedTestCase("old mcdonald"))]) - - -class TestTestProtocolServerPassThrough(unittest.TestCase): - - def setUp(self): - self.stdout = BytesIO() - self.test = subunit.RemotedTestCase("old mcdonald") - self.client = ExtendedTestResult() - self.protocol = subunit.TestProtocolServer(self.client, self.stdout) - - def keywords_before_test(self): - self.protocol.lineReceived(_b("failure a\n")) - self.protocol.lineReceived(_b("failure: a\n")) - self.protocol.lineReceived(_b("error a\n")) - self.protocol.lineReceived(_b("error: a\n")) - self.protocol.lineReceived(_b("success a\n")) - self.protocol.lineReceived(_b("success: a\n")) - self.protocol.lineReceived(_b("successful a\n")) - self.protocol.lineReceived(_b("successful: a\n")) - self.protocol.lineReceived(_b("]\n")) - self.assertEqual(self.stdout.getvalue(), _b("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")) - - def test_keywords_before_test(self): - self.keywords_before_test() - self.assertEqual(self.client._events, []) - - def test_keywords_after_error(self): - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("error old mcdonald\n")) - self.keywords_before_test() - self.assertEqual([ - ('startTest', self.test), - ('addError', self.test, {}), - ('stopTest', self.test), - ], self.client._events) - - def test_keywords_after_failure(self): - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("failure old mcdonald\n")) - self.keywords_before_test() - self.assertEqual(self.client._events, [ - ('startTest', self.test), - ('addFailure', self.test, {}), - ('stopTest', self.test), - ]) - - def test_keywords_after_success(self): - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("success old mcdonald\n")) - self.keywords_before_test() - self.assertEqual([ - ('startTest', self.test), - ('addSuccess', self.test), - ('stopTest', self.test), - ], self.client._events) - - def test_keywords_after_test(self): - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("failure a\n")) - self.protocol.lineReceived(_b("failure: a\n")) - self.protocol.lineReceived(_b("error a\n")) - self.protocol.lineReceived(_b("error: a\n")) - self.protocol.lineReceived(_b("success a\n")) - self.protocol.lineReceived(_b("success: a\n")) - self.protocol.lineReceived(_b("successful a\n")) - self.protocol.lineReceived(_b("successful: a\n")) - self.protocol.lineReceived(_b("]\n")) - self.protocol.lineReceived(_b("failure old mcdonald\n")) - self.assertEqual(self.stdout.getvalue(), _b("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, {}), - ('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(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("failure: old mcdonald [\n")) - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("failure a\n")) - self.protocol.lineReceived(_b("failure: a\n")) - self.protocol.lineReceived(_b("error a\n")) - self.protocol.lineReceived(_b("error: a\n")) - self.protocol.lineReceived(_b("success a\n")) - self.protocol.lineReceived(_b("success: a\n")) - self.protocol.lineReceived(_b("successful a\n")) - self.protocol.lineReceived(_b("successful: a\n")) - self.protocol.lineReceived(_b(" ]\n")) - self.protocol.lineReceived(_b("]\n")) - self.assertEqual(self.stdout.getvalue(), _b("")) - details = {} - details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), - lambda:[_b( - "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 - should be passed through to sys.stdout. - """ - bytes = _b("randombytes\n") - self.protocol.lineReceived(bytes) - self.assertEqual(self.stdout.getvalue(), bytes) - - -class TestTestProtocolServerLostConnection(unittest.TestCase): - - def setUp(self): - 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._events) - - def test_lost_connection_after_start(self): - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.protocol.lostConnection() - failure = subunit.RemoteError( - _u("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(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("error old mcdonald\n")) - self.protocol.lostConnection() - self.assertEqual([ - ('startTest', self.test), - ('addError', self.test, subunit.RemoteError(_u(""))), - ('stopTest', self.test), - ], self.client._events) - - def do_connection_lost(self, outcome, opening): - self.protocol.lineReceived(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("%s old mcdonald %s" % (outcome, opening))) - self.protocol.lostConnection() - failure = subunit.RemoteError( - _u("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(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("failure old mcdonald\n")) - self.protocol.lostConnection() - self.assertEqual([ - ('startTest', self.test), - ('addFailure', self.test, subunit.RemoteError(_u(""))), - ('stopTest', self.test), - ], self.client._events) - - def test_lost_connection_during_failure(self): - 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(_b("test old mcdonald\n")) - self.protocol.lineReceived(_b("success old mcdonald\n")) - self.protocol.lostConnection() - 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") - - def test_lost_connection_during_uxsuccess(self): - self.do_connection_lost("uxsuccess", "[\n") - - def test_lost_connection_during_uxsuccess_details(self): - self.do_connection_lost("uxsuccess", "[ multipart\n") - - -class TestInTestMultipart(unittest.TestCase): - - def setUp(self): - self.client = ExtendedTestResult() - self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived(_b("test mcdonalds farm\n")) - self.test = subunit.RemotedTestCase(_u("mcdonalds farm")) - - def test__outcome_sets_details_parser(self): - self.protocol._reading_success_details.details_parser = None - self.protocol._state._outcome(0, _b("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 = ExtendedTestResult() - self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived(_b("test mcdonalds farm\n")) - self.test = subunit.RemotedTestCase("mcdonalds farm") - - def simple_error_keyword(self, keyword): - self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) - 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") - - def test_simple_error_colon(self): - self.simple_error_keyword("error:") - - def test_error_empty_message(self): - self.protocol.lineReceived(_b("error mcdonalds farm [\n")) - self.protocol.lineReceived(_b("]\n")) - details = {} - details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:[_b("")]) - self.assertEqual([ - ('startTest', self.test), - ('addError', self.test, details), - ('stopTest', self.test), - ], self.client._events) - - def error_quoted_bracket(self, keyword): - self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) - self.protocol.lineReceived(_b(" ]\n")) - self.protocol.lineReceived(_b("]\n")) - details = {} - details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:[_b("]\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") - - def test_error_colon_quoted_bracket(self): - self.error_quoted_bracket("error:") - - -class TestTestProtocolServerAddFailure(unittest.TestCase): - - def setUp(self): - self.client = ExtendedTestResult() - self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived(_b("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(_b("%s mcdonalds farm\n" % keyword)) - details = {} - self.assertFailure(details) - - def test_simple_failure(self): - self.simple_failure_keyword("failure") - - def test_simple_failure_colon(self): - self.simple_failure_keyword("failure:") - - def test_failure_empty_message(self): - self.protocol.lineReceived(_b("failure mcdonalds farm [\n")) - self.protocol.lineReceived(_b("]\n")) - details = {} - details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:[_b("")]) - self.assertFailure(details) - - def failure_quoted_bracket(self, keyword): - self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) - self.protocol.lineReceived(_b(" ]\n")) - self.protocol.lineReceived(_b("]\n")) - details = {} - details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:[_b("]\n")]) - self.assertFailure(details) - - def test_failure_quoted_bracket(self): - self.failure_quoted_bracket("failure") - - def test_failure_colon_quoted_bracket(self): - 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(_b("test mcdonalds farm\n")) - self.test = self.client._events[-1][-1] - - def simple_xfail_keyword(self, keyword, as_success): - self.protocol.lineReceived(_b("%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", {'charset': 'utf8'}), - lambda:[_b(error_message)]) - if isinstance(self.client, ExtendedTestResult): - value = details - else: - if error_message is not None: - value = subunit.RemoteError(details_to_str(details)) - 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(_b("xfail mcdonalds farm [\n")) - self.protocol.lineReceived(_b("]\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(_b("%s mcdonalds farm [\n" % keyword)) - self.protocol.lineReceived(_b(" ]\n")) - self.protocol.lineReceived(_b("]\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 TestTestProtocolServerAddunexpectedSuccess(TestCase): - """Tests for the uxsuccess keyword.""" - - 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(_b("test mcdonalds farm\n")) - self.test = self.client._events[-1][-1] - - def simple_uxsuccess_keyword(self, keyword, as_fail): - self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) - self.check_fail_or_uxsuccess(as_fail) - - def check_fail_or_uxsuccess(self, as_fail, error_message=None): - details = {} - if error_message is not None: - details['traceback'] = Content( - ContentType("text", "x-traceback", {'charset': 'utf8'}), - lambda:[_b(error_message)]) - if isinstance(self.client, ExtendedTestResult): - value = details - else: - value = None - if as_fail: - self.client._events[1] = self.client._events[1][:2] - # The value is generated within the extended to original decorator: - # todo use the testtools matcher to check on this. - self.assertEqual([ - ('startTest', self.test), - ('addFailure', self.test), - ('stopTest', self.test), - ], self.client._events) - elif value: - self.assertEqual([ - ('startTest', self.test), - ('addUnexpectedSuccess', self.test, value), - ('stopTest', self.test), - ], self.client._events) - else: - self.assertEqual([ - ('startTest', self.test), - ('addUnexpectedSuccess', self.test), - ('stopTest', self.test), - ], self.client._events) - - def test_simple_uxsuccess(self): - self.setup_python26() - self.simple_uxsuccess_keyword("uxsuccess", True) - self.setup_python27() - self.simple_uxsuccess_keyword("uxsuccess", False) - self.setup_python_ex() - self.simple_uxsuccess_keyword("uxsuccess", False) - - def test_simple_uxsuccess_colon(self): - self.setup_python26() - self.simple_uxsuccess_keyword("uxsuccess:", True) - self.setup_python27() - self.simple_uxsuccess_keyword("uxsuccess:", False) - self.setup_python_ex() - self.simple_uxsuccess_keyword("uxsuccess:", False) - - def test_uxsuccess_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_fail, error_message="\n"): - self.protocol.lineReceived(_b("uxsuccess mcdonalds farm [\n")) - self.protocol.lineReceived(_b("]\n")) - self.check_fail_or_uxsuccess(as_fail, error_message) - - def uxsuccess_quoted_bracket(self, keyword, as_fail): - self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) - self.protocol.lineReceived(_b(" ]\n")) - self.protocol.lineReceived(_b("]\n")) - self.check_fail_or_uxsuccess(as_fail, "]\n") - - def test_uxsuccess_quoted_bracket(self): - self.setup_python26() - self.uxsuccess_quoted_bracket("uxsuccess", True) - self.setup_python27() - self.uxsuccess_quoted_bracket("uxsuccess", False) - self.setup_python_ex() - self.uxsuccess_quoted_bracket("uxsuccess", False) - - def test_uxsuccess_colon_quoted_bracket(self): - self.setup_python26() - self.uxsuccess_quoted_bracket("uxsuccess:", True) - self.setup_python27() - self.uxsuccess_quoted_bracket("uxsuccess:", False) - self.setup_python_ex() - self.uxsuccess_quoted_bracket("uxsuccess:", 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(_b("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(_b("%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(_b("skip mcdonalds farm [\n")) - self.protocol.lineReceived(_b("]\n")) - self.assertSkip(_b("")) - - 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(_b("%s mcdonalds farm [\n" % keyword)) - self.protocol.lineReceived(_b(" ]\n")) - self.protocol.lineReceived(_b("]\n")) - self.assertSkip(_b("]\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 = ExtendedTestResult() - self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived(_b("test mcdonalds farm\n")) - self.test = subunit.RemotedTestCase("mcdonalds farm") - - def simple_success_keyword(self, keyword): - self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) - self.assertEqual([ - ('startTest', self.test), - ('addSuccess', self.test), - ('stopTest', self.test), - ], self.client._events) - - def test_simple_success(self): - self.simple_success_keyword("successful") - - 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(_b("success mcdonalds farm [\n")) - self.protocol.lineReceived(_b("]\n")) - details = {} - details['message'] = Content(ContentType("text", "plain"), - lambda:[_b("")]) - 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(_b("%s mcdonalds farm [\n" % keyword)) - self.protocol.lineReceived(_b(" ]\n")) - self.protocol.lineReceived(_b("]\n")) - details = {} - details['message'] = Content(ContentType("text", "plain"), - lambda:[_b("]\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 = BytesIO() - self.protocol = subunit.TestProtocolServer(self.result, - stream=self.stream) - self.protocol.lineReceived(_b("progress: 23")) - self.protocol.lineReceived(_b("progress: -2")) - self.protocol.lineReceived(_b("progress: +4")) - self.assertEqual(_b(""), self.stream.getvalue()) - - def test_progress_accepted_extended(self): - # With a progress capable TestResult, progress events are emitted. - self.result = ExtendedTestResult() - self.stream = BytesIO() - self.protocol = subunit.TestProtocolServer(self.result, - stream=self.stream) - self.protocol.lineReceived(_b("progress: 23")) - self.protocol.lineReceived(_b("progress: push")) - self.protocol.lineReceived(_b("progress: -2")) - self.protocol.lineReceived(_b("progress: pop")) - self.protocol.lineReceived(_b("progress: +4")) - self.assertEqual(_b(""), 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(_b("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(_b("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(_b("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(_b("tags: foo bar\n")) - self.protocol.lineReceived(_b("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(_b("test mcdonalds farm\n")) - test = self.client._events[-1][-1] - self.protocol.lineReceived(_b("tags: foo bar\n")) - self.protocol.lineReceived(_b("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 = BytesIO() - self.protocol = subunit.TestProtocolServer(self.result, - stream=self.stream) - self.protocol.lineReceived(_b("time: 2001-12-12 12:59:59Z\n")) - self.assertEqual(_b(""), self.stream.getvalue()) - - def test_time_accepted_extended(self): - self.result = ExtendedTestResult() - self.stream = BytesIO() - self.protocol = subunit.TestProtocolServer(self.result, - stream=self.stream) - self.protocol.lineReceived(_b("time: 2001-12-12 12:59:59Z\n")) - self.assertEqual(_b(""), self.stream.getvalue()) - self.assertEqual([ - ('time', datetime.datetime(2001, 12, 12, 12, 59, 59, 0, - iso8601.Utc())) - ], self.result._events) - - -class TestRemotedTestCase(unittest.TestCase): - - def test_simple(self): - test = subunit.RemotedTestCase("A test description") - self.assertRaises(NotImplementedError, test.setUp) - self.assertRaises(NotImplementedError, test.tearDown) - self.assertEqual("A test description", - test.shortDescription()) - 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, _remote_exception_str + ": " - "Cannot run RemotedTestCases.\n\n")], - result.errors) - self.assertEqual(1, result.testsRun) - another_test = subunit.RemotedTestCase("A test description") - self.assertEqual(test, another_test) - different_test = subunit.RemotedTestCase("ofo") - self.assertNotEqual(test, different_test) - self.assertNotEqual(another_test, different_test) - - -class TestRemoteError(unittest.TestCase): - - def test_eq(self): - error = subunit.RemoteError(_u("Something went wrong")) - another_error = subunit.RemoteError(_u("Something went wrong")) - different_error = subunit.RemoteError(_u("boo!")) - self.assertEqual(error, another_error) - self.assertNotEqual(error, different_error) - self.assertNotEqual(different_error, another_error) - - def test_empty_constructor(self): - self.assertEqual(subunit.RemoteError(), subunit.RemoteError(_u(""))) - - -class TestExecTestCase(unittest.TestCase): - - class SampleExecTestCase(subunit.ExecTestCase): - - def test_sample_method(self): - """sample-script.py""" - # 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): - result = ExtendedTestResult() - test = self.SampleExecTestCase("test_sample_method") - test.run(result) - mcdonald = subunit.RemotedTestCase("old mcdonald") - bing = subunit.RemotedTestCase("bing crosby") - bing_details = {} - bing_details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:[_b("foo.c:53:ERROR invalid state\n")]) - an_error = subunit.RemotedTestCase("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") - test.debug() - - def test_count_test_cases(self): - """TODO run the child process and count responses to determine the count.""" - - def test_join_dir(self): - sibling = subunit.join_dir(__file__, 'foo') - filedir = os.path.abspath(os.path.dirname(__file__)) - expected = os.path.join(filedir, 'foo') - self.assertEqual(sibling, expected) - - -class DoExecTestCase(subunit.ExecTestCase): - - def test_working_script(self): - """sample-two-script.py""" - - -class TestIsolatedTestCase(TestCase): - - class SampleIsolatedTestCase(subunit.IsolatedTestCase): - - SETUP = False - TEARDOWN = False - TEST = False - - def setUp(self): - TestIsolatedTestCase.SampleIsolatedTestCase.SETUP = True - - def tearDown(self): - TestIsolatedTestCase.SampleIsolatedTestCase.TEARDOWN = True - - def test_sets_global_state(self): - TestIsolatedTestCase.SampleIsolatedTestCase.TEST = True - - - def test_construct(self): - self.SampleIsolatedTestCase("test_sets_global_state") - - @skipIf(os.name != "posix", "Need a posix system for forking tests") - def test_run(self): - result = unittest.TestResult() - test = self.SampleIsolatedTestCase("test_sets_global_state") - test.run(result) - self.assertEqual(result.testsRun, 1) - self.assertEqual(self.SampleIsolatedTestCase.SETUP, False) - self.assertEqual(self.SampleIsolatedTestCase.TEARDOWN, False) - self.assertEqual(self.SampleIsolatedTestCase.TEST, False) - - def test_debug(self): - pass - #test = self.SampleExecTestCase("test_sample_method") - #test.debug() - - -class TestIsolatedTestSuite(TestCase): - - class SampleTestToIsolate(unittest.TestCase): - - SETUP = False - TEARDOWN = False - TEST = False - - def setUp(self): - TestIsolatedTestSuite.SampleTestToIsolate.SETUP = True - - def tearDown(self): - TestIsolatedTestSuite.SampleTestToIsolate.TEARDOWN = True - - def test_sets_global_state(self): - TestIsolatedTestSuite.SampleTestToIsolate.TEST = True - - - def test_construct(self): - subunit.IsolatedTestSuite() - - @skipIf(os.name != "posix", "Need a posix system for forking tests") - def test_run(self): - result = unittest.TestResult() - suite = subunit.IsolatedTestSuite() - sub_suite = unittest.TestSuite() - sub_suite.addTest(self.SampleTestToIsolate("test_sets_global_state")) - sub_suite.addTest(self.SampleTestToIsolate("test_sets_global_state")) - suite.addTest(sub_suite) - suite.addTest(self.SampleTestToIsolate("test_sets_global_state")) - suite.run(result) - self.assertEqual(result.testsRun, 3) - self.assertEqual(self.SampleTestToIsolate.SETUP, False) - self.assertEqual(self.SampleTestToIsolate.TEARDOWN, False) - self.assertEqual(self.SampleTestToIsolate.TEST, False) - - -class TestTestProtocolClient(unittest.TestCase): - - def setUp(self): - self.io = BytesIO() - self.protocol = subunit.TestProtocolClient(self.io) - self.unicode_test = PlaceHolder(_u('\u2603')) - self.test = TestTestProtocolClient("test_start_test") - self.sample_details = {'something':Content( - ContentType('text', 'plain'), lambda:[_b('serialised\nform')])} - self.sample_tb_details = dict(self.sample_details) - self.sample_tb_details['traceback'] = TracebackContent( - subunit.RemoteError(_u("boo qux")), self.test) - - def test_start_test(self): - """Test startTest on a TestProtocolClient.""" - self.protocol.startTest(self.test) - self.assertEqual(self.io.getvalue(), _b("test: %s\n" % self.test.id())) - - def test_start_test_unicode_id(self): - """Test startTest on a TestProtocolClient.""" - self.protocol.startTest(self.unicode_test) - expected = _b("test: ") + _u('\u2603').encode('utf8') + _b("\n") - self.assertEqual(expected, self.io.getvalue()) - - def test_stop_test(self): - # stopTest doesn't output anything. - self.protocol.stopTest(self.test) - self.assertEqual(self.io.getvalue(), _b("")) - - def test_add_success(self): - """Test addSuccess on a TestProtocolClient.""" - self.protocol.addSuccess(self.test) - self.assertEqual( - self.io.getvalue(), _b("successful: %s\n" % self.test.id())) - - def test_add_outcome_unicode_id(self): - """Test addSuccess on a TestProtocolClient.""" - self.protocol.addSuccess(self.unicode_test) - expected = _b("successful: ") + _u('\u2603').encode('utf8') + _b("\n") - self.assertEqual(expected, self.io.getvalue()) - - 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(), _b("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(_u("boo qux"))) - self.assertEqual( - self.io.getvalue(), - _b(('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(), - _b(("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" + _remote_exception_str_chunked + ": 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(_u("phwoar crikey"))) - self.assertEqual( - self.io.getvalue(), - _b(('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(), - _b(("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" + _remote_exception_str_chunked + ": 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(_u("phwoar crikey"))) - self.assertEqual( - self.io.getvalue(), - _b(('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(), - _b(("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" + _remote_exception_str_chunked + ": 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(), - _b('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:[_b('Has it really?')])} - self.protocol.addSkip(self.test, details=details) - self.assertEqual( - self.io.getvalue(), - _b("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(), _b('progress: 23\n')) - - def test_progress_neg_cur(self): - self.protocol.progress(-23, subunit.PROGRESS_CUR) - self.assertEqual(self.io.getvalue(), _b('progress: -23\n')) - - def test_progress_pos_cur(self): - self.protocol.progress(23, subunit.PROGRESS_CUR) - self.assertEqual(self.io.getvalue(), _b('progress: +23\n')) - - def test_progress_pop(self): - self.protocol.progress(1234, subunit.PROGRESS_POP) - self.assertEqual(self.io.getvalue(), _b('progress: pop\n')) - - def test_progress_push(self): - self.protocol.progress(1234, subunit.PROGRESS_PUSH) - self.assertEqual(self.io.getvalue(), _b('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( - _b("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(), _b("uxsuccess: %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(), _b("uxsuccess: %s [ multipart\n" - "Content-Type: text/plain\n" - "something\n" - "F\r\nserialised\nform0\r\n]\n" % self.test.id())) - - def test_tags_empty(self): - self.protocol.tags(set(), set()) - self.assertEqual(_b(""), self.io.getvalue()) - - def test_tags_add(self): - self.protocol.tags(set(['foo']), set()) - self.assertEqual(_b("tags: foo\n"), self.io.getvalue()) - - def test_tags_both(self): - self.protocol.tags(set(['quux']), set(['bar'])) - self.assertEqual(_b("tags: quux -bar\n"), self.io.getvalue()) - - def test_tags_gone(self): - self.protocol.tags(set(), set(['bar'])) - self.assertEqual(_b("tags: -bar\n"), self.io.getvalue()) - - -def test_suite(): - loader = subunit.tests.TestUtil.TestLoader() - result = loader.loadTestsFromName(__name__) - return result diff --git a/lib/subunit/python/subunit/tests/test_test_results.py b/lib/subunit/python/subunit/tests/test_test_results.py deleted file mode 100644 index ff74b9a818..0000000000 --- a/lib/subunit/python/subunit/tests/test_test_results.py +++ /dev/null @@ -1,572 +0,0 @@ -# -# 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 csv -import datetime -import sys -import unittest - -from testtools import TestCase -from testtools.compat import StringIO -from testtools.content import ( - text_content, - TracebackContent, - ) -from testtools.testresult.doubles import ExtendedTestResult - -import subunit -import subunit.iso8601 as iso8601 -import subunit.test_results - -import testtools - - -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 = [] - self.failfast = False - - 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_set_failfast_True(self): - self.assertFalse(self.decorated.failfast) - self.result.failfast = True - self.assertTrue(self.decorated.failfast) - - -class TestTagCollapsingDecorator(TestCase): - - def test_tags_collapsed_outside_of_tests(self): - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TagCollapsingDecorator(result) - tag_collapser.tags(set(['a']), set()) - tag_collapser.tags(set(['b']), set()) - tag_collapser.startTest(self) - self.assertEquals( - [('tags', set(['a', 'b']), set([])), - ('startTest', self), - ], result._events) - - def test_tags_collapsed_outside_of_tests_are_flushed(self): - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TagCollapsingDecorator(result) - tag_collapser.startTestRun() - tag_collapser.tags(set(['a']), set()) - tag_collapser.tags(set(['b']), set()) - tag_collapser.startTest(self) - tag_collapser.addSuccess(self) - tag_collapser.stopTest(self) - tag_collapser.stopTestRun() - self.assertEquals( - [('startTestRun',), - ('tags', set(['a', 'b']), set([])), - ('startTest', self), - ('addSuccess', self), - ('stopTest', self), - ('stopTestRun',), - ], result._events) - - def test_tags_forwarded_after_tests(self): - test = subunit.RemotedTestCase('foo') - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TagCollapsingDecorator(result) - tag_collapser.startTestRun() - tag_collapser.startTest(test) - tag_collapser.addSuccess(test) - tag_collapser.stopTest(test) - tag_collapser.tags(set(['a']), set(['b'])) - tag_collapser.stopTestRun() - self.assertEqual( - [('startTestRun',), - ('startTest', test), - ('addSuccess', test), - ('stopTest', test), - ('tags', set(['a']), set(['b'])), - ('stopTestRun',), - ], - result._events) - - def test_tags_collapsed_inside_of_tests(self): - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TagCollapsingDecorator(result) - test = subunit.RemotedTestCase('foo') - tag_collapser.startTest(test) - tag_collapser.tags(set(['a']), set()) - tag_collapser.tags(set(['b']), set(['a'])) - tag_collapser.tags(set(['c']), set()) - tag_collapser.stopTest(test) - self.assertEquals( - [('startTest', test), - ('tags', set(['b', 'c']), set(['a'])), - ('stopTest', test)], - result._events) - - def test_tags_collapsed_inside_of_tests_different_ordering(self): - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TagCollapsingDecorator(result) - test = subunit.RemotedTestCase('foo') - tag_collapser.startTest(test) - tag_collapser.tags(set(), set(['a'])) - tag_collapser.tags(set(['a', 'b']), set()) - tag_collapser.tags(set(['c']), set()) - tag_collapser.stopTest(test) - self.assertEquals( - [('startTest', test), - ('tags', set(['a', 'b', 'c']), set()), - ('stopTest', test)], - result._events) - - def test_tags_sent_before_result(self): - # Because addSuccess and friends tend to send subunit output - # immediately, and because 'tags:' before a result line means - # something different to 'tags:' after a result line, we need to be - # sure that tags are emitted before 'addSuccess' (or whatever). - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TagCollapsingDecorator(result) - test = subunit.RemotedTestCase('foo') - tag_collapser.startTest(test) - tag_collapser.tags(set(['a']), set()) - tag_collapser.addSuccess(test) - tag_collapser.stopTest(test) - self.assertEquals( - [('startTest', test), - ('tags', set(['a']), set()), - ('addSuccess', test), - ('stopTest', test)], - result._events) - - -class TestTimeCollapsingDecorator(TestCase): - - def make_time(self): - # Heh heh. - return datetime.datetime( - 2000, 1, self.getUniqueInteger(), tzinfo=iso8601.UTC) - - def test_initial_time_forwarded(self): - # We always forward the first time event we see. - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TimeCollapsingDecorator(result) - a_time = self.make_time() - tag_collapser.time(a_time) - self.assertEquals([('time', a_time)], result._events) - - def test_time_collapsed_to_first_and_last(self): - # If there are many consecutive time events, only the first and last - # are sent through. - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TimeCollapsingDecorator(result) - times = [self.make_time() for i in range(5)] - for a_time in times: - tag_collapser.time(a_time) - tag_collapser.startTest(subunit.RemotedTestCase('foo')) - self.assertEquals( - [('time', times[0]), ('time', times[-1])], result._events[:-1]) - - def test_only_one_time_sent(self): - # If we receive a single time event followed by a non-time event, we - # send exactly one time event. - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TimeCollapsingDecorator(result) - a_time = self.make_time() - tag_collapser.time(a_time) - tag_collapser.startTest(subunit.RemotedTestCase('foo')) - self.assertEquals([('time', a_time)], result._events[:-1]) - - def test_duplicate_times_not_sent(self): - # Many time events with the exact same time are collapsed into one - # time event. - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TimeCollapsingDecorator(result) - a_time = self.make_time() - for i in range(5): - tag_collapser.time(a_time) - tag_collapser.startTest(subunit.RemotedTestCase('foo')) - self.assertEquals([('time', a_time)], result._events[:-1]) - - def test_no_times_inserted(self): - result = ExtendedTestResult() - tag_collapser = subunit.test_results.TimeCollapsingDecorator(result) - a_time = self.make_time() - tag_collapser.time(a_time) - foo = subunit.RemotedTestCase('foo') - tag_collapser.startTest(foo) - tag_collapser.addSuccess(foo) - tag_collapser.stopTest(foo) - self.assertEquals( - [('time', a_time), - ('startTest', foo), - ('addSuccess', foo), - ('stopTest', foo)], result._events) - - -class TestByTestResultTests(testtools.TestCase): - - def setUp(self): - super(TestByTestResultTests, self).setUp() - self.log = [] - self.result = subunit.test_results.TestByTestResult(self.on_test) - if sys.version_info >= (3, 0): - self.result._now = iter(range(5)).__next__ - else: - self.result._now = iter(range(5)).next - - def assertCalled(self, **kwargs): - defaults = { - 'test': self, - 'tags': set(), - 'details': None, - 'start_time': 0, - 'stop_time': 1, - } - defaults.update(kwargs) - self.assertEqual([defaults], self.log) - - def on_test(self, **kwargs): - self.log.append(kwargs) - - def test_no_tests_nothing_reported(self): - self.result.startTestRun() - self.result.stopTestRun() - self.assertEqual([], self.log) - - def test_add_success(self): - self.result.startTest(self) - self.result.addSuccess(self) - self.result.stopTest(self) - self.assertCalled(status='success') - - def test_add_success_details(self): - self.result.startTest(self) - details = {'foo': 'bar'} - self.result.addSuccess(self, details=details) - self.result.stopTest(self) - self.assertCalled(status='success', details=details) - - def test_tags(self): - if not getattr(self.result, 'tags', None): - self.skipTest("No tags in testtools") - self.result.tags(['foo'], []) - self.result.startTest(self) - self.result.addSuccess(self) - self.result.stopTest(self) - self.assertCalled(status='success', tags=set(['foo'])) - - def test_add_error(self): - self.result.startTest(self) - try: - 1/0 - except ZeroDivisionError: - error = sys.exc_info() - self.result.addError(self, error) - self.result.stopTest(self) - self.assertCalled( - status='error', - details={'traceback': TracebackContent(error, self)}) - - def test_add_error_details(self): - self.result.startTest(self) - details = {"foo": text_content("bar")} - self.result.addError(self, details=details) - self.result.stopTest(self) - self.assertCalled(status='error', details=details) - - def test_add_failure(self): - self.result.startTest(self) - try: - self.fail("intentional failure") - except self.failureException: - failure = sys.exc_info() - self.result.addFailure(self, failure) - self.result.stopTest(self) - self.assertCalled( - status='failure', - details={'traceback': TracebackContent(failure, self)}) - - def test_add_failure_details(self): - self.result.startTest(self) - details = {"foo": text_content("bar")} - self.result.addFailure(self, details=details) - self.result.stopTest(self) - self.assertCalled(status='failure', details=details) - - def test_add_xfail(self): - self.result.startTest(self) - try: - 1/0 - except ZeroDivisionError: - error = sys.exc_info() - self.result.addExpectedFailure(self, error) - self.result.stopTest(self) - self.assertCalled( - status='xfail', - details={'traceback': TracebackContent(error, self)}) - - def test_add_xfail_details(self): - self.result.startTest(self) - details = {"foo": text_content("bar")} - self.result.addExpectedFailure(self, details=details) - self.result.stopTest(self) - self.assertCalled(status='xfail', details=details) - - def test_add_unexpected_success(self): - self.result.startTest(self) - details = {'foo': 'bar'} - self.result.addUnexpectedSuccess(self, details=details) - self.result.stopTest(self) - self.assertCalled(status='success', details=details) - - def test_add_skip_reason(self): - self.result.startTest(self) - reason = self.getUniqueString() - self.result.addSkip(self, reason) - self.result.stopTest(self) - self.assertCalled( - status='skip', details={'reason': text_content(reason)}) - - def test_add_skip_details(self): - self.result.startTest(self) - details = {'foo': 'bar'} - self.result.addSkip(self, details=details) - self.result.stopTest(self) - self.assertCalled(status='skip', details=details) - - def test_twice(self): - self.result.startTest(self) - self.result.addSuccess(self, details={'foo': 'bar'}) - self.result.stopTest(self) - self.result.startTest(self) - self.result.addSuccess(self) - self.result.stopTest(self) - self.assertEqual( - [{'test': self, - 'status': 'success', - 'start_time': 0, - 'stop_time': 1, - 'tags': set(), - 'details': {'foo': 'bar'}}, - {'test': self, - 'status': 'success', - 'start_time': 2, - 'stop_time': 3, - 'tags': set(), - 'details': None}, - ], - self.log) - - -class TestCsvResult(testtools.TestCase): - - def parse_stream(self, stream): - stream.seek(0) - reader = csv.reader(stream) - return list(reader) - - def test_csv_output(self): - stream = StringIO() - result = subunit.test_results.CsvResult(stream) - if sys.version_info >= (3, 0): - result._now = iter(range(5)).__next__ - else: - result._now = iter(range(5)).next - result.startTestRun() - result.startTest(self) - result.addSuccess(self) - result.stopTest(self) - result.stopTestRun() - self.assertEqual( - [['test', 'status', 'start_time', 'stop_time'], - [self.id(), 'success', '0', '1'], - ], - self.parse_stream(stream)) - - def test_just_header_when_no_tests(self): - stream = StringIO() - result = subunit.test_results.CsvResult(stream) - result.startTestRun() - result.stopTestRun() - self.assertEqual( - [['test', 'status', 'start_time', 'stop_time']], - self.parse_stream(stream)) - - def test_no_output_before_events(self): - stream = StringIO() - subunit.test_results.CsvResult(stream) - self.assertEqual([], self.parse_stream(stream)) - - -def test_suite(): - loader = subunit.tests.TestUtil.TestLoader() - result = loader.loadTestsFromName(__name__) - return result |