[PATCH 02/17] subunit: Update to new upstream snapshot.
Jelmer Vernooij
jelmer at samba.org
Sat Nov 1 13:21:54 MDT 2014
Change-Id: I5da01c29fee4998df04ab8333230139377ec91c0
Signed-Off-By: Jelmer Vernooij <jelmer at samba.org>
---
lib/subunit/.gitignore | 56 ++
lib/subunit/INSTALL | 5 +-
lib/subunit/MANIFEST.in | 2 +-
lib/subunit/Makefile.am | 24 +-
lib/subunit/NEWS | 203 +++++++
lib/subunit/README | 281 +++++++++-
lib/subunit/all_tests.py | 36 ++
lib/subunit/configure.ac | 3 +-
lib/subunit/filters/subunit-1to2 | 42 ++
lib/subunit/filters/subunit-2to1 | 47 ++
lib/subunit/filters/subunit-filter | 25 +-
lib/subunit/filters/subunit-ls | 32 +-
lib/subunit/filters/subunit-notify | 6 +-
lib/subunit/filters/subunit-output | 23 +
lib/subunit/filters/subunit-stats | 29 +-
lib/subunit/filters/subunit-tags | 1 +
lib/subunit/filters/subunit2csv | 5 +-
lib/subunit/filters/subunit2gtk | 77 +--
lib/subunit/filters/subunit2junitxml | 7 +-
lib/subunit/filters/subunit2pyunit | 23 +-
lib/subunit/perl/Makefile.PL.in | 4 +-
lib/subunit/python/subunit/__init__.py | 150 +++---
lib/subunit/python/subunit/_output.py | 203 +++++++
lib/subunit/python/subunit/filters.py | 111 +++-
lib/subunit/python/subunit/run.py | 79 ++-
lib/subunit/python/subunit/test_results.py | 64 ++-
lib/subunit/python/subunit/tests/TestUtil.py | 80 ---
lib/subunit/python/subunit/tests/__init__.py | 52 +-
lib/subunit/python/subunit/tests/test_chunked.py | 6 -
lib/subunit/python/subunit/tests/test_details.py | 6 -
lib/subunit/python/subunit/tests/test_filters.py | 35 ++
.../python/subunit/tests/test_output_filter.py | 596 +++++++++++++++++++++
.../python/subunit/tests/test_progress_model.py | 6 -
lib/subunit/python/subunit/tests/test_run.py | 82 ++-
.../python/subunit/tests/test_subunit_filter.py | 110 ++--
.../python/subunit/tests/test_subunit_stats.py | 6 -
.../python/subunit/tests/test_subunit_tags.py | 90 ++--
.../python/subunit/tests/test_tap2subunit.py | 378 ++++++-------
.../python/subunit/tests/test_test_protocol.py | 75 ++-
.../python/subunit/tests/test_test_protocol2.py | 436 +++++++++++++++
.../python/subunit/tests/test_test_results.py | 6 -
lib/subunit/python/subunit/v2.py | 495 +++++++++++++++++
lib/subunit/runtests.py | 138 -----
lib/subunit/setup.py | 22 +-
44 files changed, 3289 insertions(+), 868 deletions(-)
create mode 100644 lib/subunit/.gitignore
create mode 100644 lib/subunit/all_tests.py
create mode 100755 lib/subunit/filters/subunit-1to2
create mode 100755 lib/subunit/filters/subunit-2to1
create mode 100644 lib/subunit/filters/subunit-output
create mode 100644 lib/subunit/python/subunit/_output.py
delete mode 100644 lib/subunit/python/subunit/tests/TestUtil.py
create mode 100644 lib/subunit/python/subunit/tests/test_filters.py
create mode 100644 lib/subunit/python/subunit/tests/test_output_filter.py
create mode 100644 lib/subunit/python/subunit/tests/test_test_protocol2.py
create mode 100644 lib/subunit/python/subunit/v2.py
delete mode 100755 lib/subunit/runtests.py
diff --git a/lib/subunit/.gitignore b/lib/subunit/.gitignore
new file mode 100644
index 0000000..30f733a
--- /dev/null
+++ b/lib/subunit/.gitignore
@@ -0,0 +1,56 @@
+/c/lib/child.os
+/c/tests/test_child
+.sconsign
+.sconsign.dblite
+/m4/**
+aclocal.m4
+compile
+config.guess
+config.h.in
+config.sub
+configure
+depcomp
+install-sh
+ltmain.sh
+missing
+autom4te.cache
+Makefile.in
+py-compile
+.deps
+.dirstamp
+.libs
+*.lo
+libsubunit.la
+libcppunit_subunit.la
+libtool
+stamp-h1
+libsubunit.pc
+libcppunit_subunit.pc
+config.log
+config.status
+Makefile
+config.h
+debian/files
+debian/libsubunit0
+debian/libsubunit-dev
+debian/subunit
+debian/python-subunit
+debian/*.log
+debian/*.debhelper
+debian/tmp
+debian/*.substvars
+/perl/blib
+/perl/pm_to_blib
+subunit-*.tar.gz
+subunit-*.tar.gz.asc
+perl/Makefile.PL
+/.testrepository
+__pycache__
+perl/MYMETA.yml
+/build/
+/dist/
+/*.egg-info/
+*.pyc
+*~
+.*.swp
+.*.swo
diff --git a/lib/subunit/INSTALL b/lib/subunit/INSTALL
index eeea734..29052eb 100644
--- a/lib/subunit/INSTALL
+++ b/lib/subunit/INSTALL
@@ -14,9 +14,12 @@ Dependencies
* Python for the filters
* 'testtools' (On Debian and Ubuntu systems the 'python-testtools' package,
the testtools package on pypi, or https://launchpad.net/testtools) for
- the extended test API which permits attachments. Version 0.9.23 or newer is
+ the extended test API which permits attachments. Version 0.9.30 or newer is
required. Of particular note, http://testtools.python-hosting.com/ is not
the testtools you want.
+* 'testscenarios' (On Debian and Ubuntu systems the 'python-testscenarios'
+ package, the 'testscenarios' package on pypi, or
+ https://launchpad.net/testscenarios) for running some of the python unit tests.
* A C compiler for the C bindings
* Perl for the Perl tools (including subunit-diff)
* Check to run the subunit test suite.
diff --git a/lib/subunit/MANIFEST.in b/lib/subunit/MANIFEST.in
index eb98981..4f521dc 100644
--- a/lib/subunit/MANIFEST.in
+++ b/lib/subunit/MANIFEST.in
@@ -1,4 +1,4 @@
-exclude .bzrignore
+exclude .gitignore
exclude aclocal.m4
prune autom4te.cache
prune c
diff --git a/lib/subunit/Makefile.am b/lib/subunit/Makefile.am
index da16020..e8f018e 100644
--- a/lib/subunit/Makefile.am
+++ b/lib/subunit/Makefile.am
@@ -6,6 +6,7 @@ EXTRA_DIST = \
Makefile.am \
NEWS \
README \
+ all_tests.py \
c++/README \
c/README \
c/check-subunit-0.9.3.patch \
@@ -20,21 +21,22 @@ EXTRA_DIST = \
python/iso8601/README.subunit \
python/iso8601/setup.py \
python/iso8601/test_iso8601.py \
- python/subunit/tests/TestUtil.py \
python/subunit/tests/__init__.py \
python/subunit/tests/sample-script.py \
python/subunit/tests/sample-two-script.py \
python/subunit/tests/test_chunked.py \
python/subunit/tests/test_details.py \
+ python/subunit/tests/test_filters.py \
+ python/subunit/tests/test_output_filter.py \
python/subunit/tests/test_progress_model.py \
- python/subunit/tests/test_subunit_filter.py \
python/subunit/tests/test_run.py \
+ python/subunit/tests/test_subunit_filter.py \
python/subunit/tests/test_subunit_stats.py \
python/subunit/tests/test_subunit_tags.py \
python/subunit/tests/test_tap2subunit.py \
python/subunit/tests/test_test_protocol.py \
+ python/subunit/tests/test_test_protocol2.py \
python/subunit/tests/test_test_results.py \
- runtests.py \
setup.py \
shell/README \
shell/share/subunit.sh \
@@ -47,9 +49,12 @@ ACLOCAL_AMFLAGS = -I m4
include_subunitdir = $(includedir)/subunit
dist_bin_SCRIPTS = \
+ filters/subunit-1to2 \
+ filters/subunit-2to1 \
filters/subunit-filter \
filters/subunit-ls \
filters/subunit-notify \
+ filters/subunit-output \
filters/subunit-stats \
filters/subunit-tags \
filters/subunit2csv \
@@ -58,8 +63,7 @@ dist_bin_SCRIPTS = \
filters/subunit2pyunit \
filters/tap2subunit
-TESTS_ENVIRONMENT = SHELL_SHARE='$(top_srcdir)/shell/share/' PYTHONPATH='$(abs_top_srcdir)/python':${PYTHONPATH}
-TESTS = runtests.py $(check_PROGRAMS)
+TESTS = $(check_PROGRAMS)
## install libsubunit.pc
pcdatadir = $(libdir)/pkgconfig
@@ -75,7 +79,9 @@ pkgpython_PYTHON = \
python/subunit/iso8601.py \
python/subunit/progress_model.py \
python/subunit/run.py \
- python/subunit/test_results.py
+ python/subunit/v2.py \
+ python/subunit/test_results.py \
+ python/subunit/_output.py
lib_LTLIBRARIES = libsubunit.la
lib_LTLIBRARIES += libcppunit_subunit.la
@@ -87,9 +93,6 @@ include_subunit_HEADERS = \
check_PROGRAMS = \
c/tests/test_child
-check_SCRIPTS = \
- runtests.py
-
libsubunit_la_SOURCES = \
c/lib/child.c \
c/include/subunit/child.h
@@ -108,6 +111,9 @@ all-local: perl/Makefile
check-local: perl/Makefile
$(MAKE) -C perl check
+ SHELL_SHARE='$(top_srcdir)/shell/share/' \
+ PYTHONPATH='$(abs_top_srcdir)/python':'$(abs_top_srcdir)':${PYTHONPATH} \
+ $(PYTHON) -m testtools.run all_tests.test_suite
clean-local:
find . -type f -name "*.pyc" -exec rm {} ';'
diff --git a/lib/subunit/NEWS b/lib/subunit/NEWS
index 081dc5d..119d2ee 100644
--- a/lib/subunit/NEWS
+++ b/lib/subunit/NEWS
@@ -5,6 +5,209 @@ subunit release notes
NEXT (In development)
---------------------
+0.0.21
+------
+
+BUGFIXES
+~~~~~~~~
+
+* Brown bag bugfix - 0.0.20's setup.py referenced cvs not csv.
+ (Robert Collins, #1361924)
+
+0.0.20
+------
+
+BUGFIXES
+~~~~~~~~
+
+* subunit2csv is now installed when using pip.
+ (Robert Collins, #1279669)
+
+* testscenarios is now a test dependency, not an install dependency.
+ (Arfrever Frehtes Taifersar Arahesis, #1292757)
+
+* The python-subunit tarball can now have setup run from the current
+ directory. (Robert Collins, #1361857)
+
+0.0.19
+------
+
+IMPROVEMENTS
+~~~~~~~~~~~~
+
+* ``subunit.run`` in Python will now exit 0 as long as the test stream has
+ been generated correctly - this has always been the intent but API friction
+ with testtools had prevented it working.
+ (Robert Collins)
+
+0.0.18
+------
+
+IMPROVEMENTS
+~~~~~~~~~~~~
+
+* Fix compatibility with testtools 0.9.35 which dropped the 'all' compat
+ symbol. This breaks support for Python versions lower than 2.6.
+ (Robert Collins, #1274056)
+
+0.0.17
+------
+
+IMPROVEMENTS
+~~~~~~~~~~~~
+
+* Add ``subunit-output`` tool that can generate a Subunit v2 bytestream from
+ arguments passed on the command line. (Thomi Richards, #1252084)
+
+0.0.16
+------
+
+BUG FIXES
+~~~~~~~~~
+
+* Perl files should now honour perl system config.
+ (Benedikt Morbach, #1233198)
+
+* Python 3.1 and 3.2 have an inconsistent memoryview implementation which
+ required a workaround for NUL byte detection. (Robert Collins, #1216246)
+
+* The test suite was failing 6 tests due to testtools changing it's output
+ formatting of exceptions. (Robert Collins)
+
+* V2 parser errors now set appropriate mime types for the encapsulated packet
+ data and the error message. (Robert Collins)
+
+* When tests fail to import ``python subunit.run -l ...`` will now write a
+ subunit file attachment listing the failed imports and exit 2, rather than
+ listing the stub objects from the importer and exiting 0.
+ (Robert Collins, #1245672)
+
+IMPROVEMENTS
+~~~~~~~~~~~~
+
+* Most filters will now accept a file path argument instead of only reading
+ from stdin. (Robert Collins, #409206)
+
+0.0.15
+------
+
+BUG FIXES
+~~~~~~~~~
+
+* Clients of subunit did not expect memoryview objects in StreamResult events.
+ (Robert Collins)
+
+* Memoryview and struct were mutually incompatible in 2.7.3 and 3.2.
+ (Robert Collins, #1216163)
+
+0.0.14
+------
+
+BUG FIXES
+~~~~~~~~~
+
+* Memoryview detection was broken and thus it's use was never really tested.
+ (Robert Collins, 1216101)
+
+* TestProtocol2's tag tests were set sort order dependent.
+ (Robert Collins, #1025392)
+
+* TestTestProtocols' test_tags_both was set sort order dependent.
+ (Robert Collins, #1025392)
+
+* TestTestProtocols' test_*_details were dictionary sort order dependent.
+ (Robert Collins, #1025392)
+
+* TestSubUnitTags's test_add_tag was also se sort order dependent.
+ (Robert Collins, #1025392)
+
+0.0.13
+------
+
+IMPROVEMENTS
+~~~~~~~~~~~~
+
+* subunit should now build with automake 1.11 again. (Robert Collins)
+
+* `subunit-stats` no longer outputs encapsulated stdout as subunit.
+ (Robert Collins, #1171987)
+
+* The logic for `subunit.run` is now importable via python -
+ `subunit.run.main`. (Robert Collins, #606770)
+
+BUG FIXES
+~~~~~~~~~
+
+* Removed GPL files that were (C) non Subunit Developers - they are
+ incompatible for binary distribution, which affects redistributors.
+ (Robert Collins, #1185591)
+
+0.0.12
+------
+
+BUG FIXES
+~~~~~~~~~
+
+* Subunit v2 packets with both file content and route code were not being
+ parsed correctly - they would incorrectly emit a parser error, due to trying
+ to parse the route code length from the first byes of the file content.
+ (Robert Collins, 1172815)
+
+0.0.11
+------
+
+v2 protocol draft included in this release. The v2 protocol trades off human
+readability for a massive improvement in robustness, the ability to represent
+concurrent tests in a single stream, cheaper parsing, and that provides
+significantly better in-line debugging support and structured forwarding
+of non-test data (such as stdout or stdin data).
+
+This change includes two new filters (subunit-1to2 and subunit-2to1). Use
+these filters to convert old streams to v2 and convert v2 streams to v1.
+
+All the other filters now only parse and emit v2 streams. V2 is still in
+draft format, so if you want to delay and wait for v2 to be finalised, you
+should use subunit-2to1 before any serialisation steps take place.
+With the ability to encapsulate multiple non-test streams, another significant
+cange is that filters which emit subunit now encapsulate any non-subunit they
+encounter, labelling it 'stdout'. This permits multiplexing such streams and
+detangling the stdout streams from each input.
+
+The subunit libraries (Python etc) have not changed their behaviour: they
+still emit v1 from their existing API calls. New API's are being added
+and applications should migrate once their language has those API's available.
+
+IMPROVEMENTS
+~~~~~~~~~~~~
+
+* ``subunit.run`` now replaces sys.stdout to ensure that stdout is unbuffered
+ - without this pdb output is not reliably visible when stdout is a pipe
+ as it usually is. (Robert Collins)
+
+* v2 protocol draft included in this release. (Python implementation only so
+ far). (Robert Collins)
+
+* Two new Python classes -- ``StreamResultToBytes`` and
+ ``ByteStreamToStreamResult`` handle v2 generation and parsing.
+ (Robert Collins)
+
+0.0.10
+------
+
+BUG FIXES
+~~~~~~~~~
+
+* make_stream_binary is now public for reuse. (Robert Collins)
+
+* NAME was not defined in the protocol BNF. (Robert Collins)
+
+* UnsupportedOperation is available in the Python2.6 io library, so ask
+ forgiveness rather than permission for obtaining it. (Robert Collins)
+
+* Streams with no fileno() attribute are now supported, but they are not
+ checked for being in binary mode: be sure to take care of that if using
+ the library yourself. (Robert Collins)
+
0.0.9
-----
diff --git a/lib/subunit/README b/lib/subunit/README
index 47a9734..dab8be7 100644
--- a/lib/subunit/README
+++ b/lib/subunit/README
@@ -1,12 +1,12 @@
subunit: A streaming protocol for test results
- Copyright (C) 2005-2009 Robert Collins <robertc at robertcollins.net>
+ Copyright (C) 2005-2013 Robert Collins <robertc at 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
@@ -21,9 +21,26 @@
Subunit
-------
-Subunit is a streaming protocol for test results. The protocol is human
-readable and easily generated and parsed. By design all the components of
-the protocol conceptually fit into the xUnit TestCase->TestResult interaction.
+Subunit is a streaming protocol for test results.
+
+There are two major revisions of the protocol. Version 1 was trivially human
+readable but had significant defects as far as highly parallel testing was
+concerned - it had no room for doing discovery and execution in parallel,
+required substantial buffering when multiplexing and was fragile - a corrupt
+byte could cause an entire stream to be misparsed. Version 1.1 added
+encapsulation of binary streams which mitigated some of the issues but the
+core remained.
+
+Version 2 shares many of the good characteristics of Version 1 - it can be
+embedded into a regular text stream (e.g. from a build system) and it still
+models xUnit style test execution. It also fixes many of the issues with
+Version 1 - Version 2 can be multiplexed without excessive buffering (in
+time or space), it has a well defined recovery mechanism for dealing with
+corrupted streams (e.g. where two processes write to the same stream
+concurrently, or where the stream generator suffers a bug).
+
+More details on both protocol version s can be found in the 'Protocol' section
+of this document.
Subunit comes with command line filters to process a subunit stream and
language bindings for python, C, C++ and shell. Bindings are easy to write
@@ -32,11 +49,12 @@ for other languages.
A number of useful things can be done easily with subunit:
* Test aggregation: Tests run separately can be combined and then
reported/displayed together. For instance, tests from different languages
- can be shown as a seamless whole.
+ can be shown as a seamless whole, and tests running on multiple machines
+ can be aggregated into a single stream through a multiplexer.
* Test archiving: A test run may be recorded and replayed later.
* Test isolation: Tests that may crash or otherwise interact badly with each
- other can be run separately and then aggregated, rather than interfering
- with each other.
+ other can be run seperately and then aggregated, rather than interfering
+ with each other or requiring an adhoc test->runner reporting protocol.
* Grid testing: subunit can act as the necessary serialisation and
deserialiation to get test runs on distributed machines to be reported in
real time.
@@ -68,20 +86,20 @@ Subunit has excellent Python support: most of the filters and tools are written
in python and there are facilities for using Subunit to increase test isolation
seamlessly within a test suite.
-One simple way to run an existing python test suite and have it output subunit
-is the module ``subunit.run``::
+The most common way is to run an existing python test suite and have it output
+subunit via the ``subunit.run`` module::
$ python -m subunit.run mypackage.tests.test_suite
-
+
For more information on the Python support Subunit offers , please see
-``pydoc subunit``, or the source in ``python/subunit/__init__.py``
+``pydoc subunit``, or the source in ``python/subunit/``
C
=
-Subunit has C bindings to emit the protocol, and comes with a patch for 'check'
-which has been nominally accepted by the 'check' developers. See 'c/README' for
-more details.
+Subunit has C bindings to emit the protocol. The 'check' C unit testing project
+has included subunit support in their project for some years now. See
+'c/README' for more details.
C++
===
@@ -92,9 +110,13 @@ CPPUnit is included in the Subunit distribution. See 'c++/README' for details.
shell
=====
-Similar to C, the shell bindings consist of simple functions to output protocol
-elements, and a patch for adding subunit output to the 'ShUnit' shell test
-runner. See 'shell/README' for details.
+There are two sets of shell tools. There are filters, which accept a subunit
+stream on stdin and output processed data (or a transformed stream) on stdout.
+
+Then there are unittest facilities similar to those for C : shell bindings
+consisting of simple functions to output protocol elements, and a patch for
+adding subunit output to the 'ShUnit' shell test runner. See 'shell/README' for
+details.
Filter recipes
--------------
@@ -104,9 +126,225 @@ To ignore some failing tests whose root cause is already known::
subunit-filter --without 'AttributeError.*flavor'
+The xUnit test model
+--------------------
+
+Subunit implements a slightly modified xUnit test model. The stock standard
+model is that there are tests, which have an id(), can be run, and when run
+start, emit an outcome (like success or failure) and then finish.
+
+Subunit extends this with the idea of test enumeration (find out about tests
+a runner has without running them), tags (allow users to describe tests in
+ways the test framework doesn't apply any semantic value to), file attachments
+(allow arbitrary data to make analysing a failure easy) and timestamps.
+
The protocol
------------
+Version 2, or v2 is new and still under development, but is intended to
+supercede version 1 in the very near future. Subunit's bundled tools accept
+only version 2 and only emit version 2, but the new filters subunit-1to2 and
+subunit-2to1 can be used to interoperate with older third party libraries.
+
+Version 2
+=========
+
+Version 2 is a binary protocol consisting of independent packets that can be
+embedded in the output from tools like make - as long as each packet has no
+other bytes mixed in with it (which 'make -j N>1' has a tendency of doing).
+Version 2 is currently in draft form, and early adopters should be willing
+to either discard stored results (if protocol changes are made), or bulk
+convert them back to v1 and then to a newer edition of v2.
+
+The protocol synchronises at the start of the stream, after a packet, or
+after any 0x0A byte. That is, a subunit v2 packet starts after a newline or
+directly after the end of the prior packet.
+
+Subunit is intended to be transported over a reliable streaming protocol such
+as TCP. As such it does not concern itself with out of order delivery of
+packets. However, because of the possibility of corruption due to either
+bugs in the sender, or due to mixed up data from concurrent writes to the same
+fd when being embedded, subunit strives to recover reasonably gracefully from
+damaged data.
+
+A key design goal for Subunit version 2 is to allow processing and multiplexing
+without forcing buffering for semantic correctness, as buffering tends to hide
+hung or otherwise misbehaving tests. That said, limited time based buffering
+for network efficiency is a good idea - this is ultimately implementator
+choice. Line buffering is also discouraged for subunit streams, as dropping
+into a debugger or other tool may require interactive traffic even if line
+buffering would not otherwise be a problem.
+
+In version two there are two conceptual events - a test status event and a file
+attachment event. Events may have timestamps, and the path of multiplexers that
+an event is routed through is recorded to permit sending actions back to the
+source (such as new tests to run or stdin for driving debuggers and other
+interactive input). Test status events are used to enumerate tests, to report
+tests and test helpers as they run. Tests may have tags, used to allow
+tunnelling extra meanings through subunit without requiring parsing of
+arbitrary file attachments. Things that are not standalone tests get marked
+as such by setting the 'Runnable' flag to false. (For instance, individual
+assertions in TAP are not runnable tests, only the top level TAP test script
+is runnable).
+
+File attachments are used to provide rich detail about the nature of a failure.
+File attachments can also be used to encapsulate stdout and stderr both during
+and outside tests.
+
+Most numbers are stored in network byte order - Most Significant Byte first
+encoded using a variation of http://www.dlugosz.com/ZIP2/VLI.html. The first
+byte's top 2 high order bits encode the total number of octets in the number.
+This encoding can encode values from 0 to 2**30-1, enough to encode a
+nanosecond. Numbers that are not variable length encoded are still stored in
+MSB order.
+
+ prefix octets max max
++-------+--------+---------+------------+
+| 00 | 1 | 2**6-1 | 63 |
+| 01 | 2 | 2**14-1 | 16383 |
+| 10 | 3 | 2**22-1 | 4194303 |
+| 11 | 4 | 2**30-1 | 1073741823 |
++-------+--------+---------+------------+
+
+All variable length elements of the packet are stored with a length prefix
+number allowing them to be skipped over for consumers that don't need to
+interpret them.
+
+UTF-8 strings are with no terminating NUL and should not have any embedded NULs
+(implementations SHOULD validate any such strings that they process and take
+some remedial action (such as discarding the packet as corrupt).
+
+In short the structure of a packet is:
+PACKET := SIGNATURE FLAGS PACKET_LENGTH TIMESTAMP? TESTID? TAGS? MIME?
+ FILECONTENT? ROUTING_CODE? CRC32
+
+In more detail...
+
+Packets are identified by a single byte signature - 0xB3, which is never legal
+in a UTF-8 stream as the first byte of a character. 0xB3 starts with the first
+bit set and the second not, which is the UTF-8 signature for a continuation
+byte. 0xB3 was chosen as 0x73 ('s' in ASCII') with the top two bits replaced by
+the 1 and 0 for a continuation byte.
+
+If subunit packets are being embedded in a non-UTF-8 text stream, where 0x73 is
+a legal character, consider either recoding the text to UTF-8, or using
+subunit's 'file' packets to embed the text stream in subunit, rather than the
+other way around.
+
+Following the signature byte comes a 16-bit flags field, which includes a
+4-bit version field - if the version is not 0x2 then the packet cannot be
+read. It is recommended to signal an error at this point (e.g. by emitting
+a synthetic error packet and returning to the top level loop to look for
+new packets, or exiting with an error). If recovery is desired, treat the
+packet signature as an opaque byte and scan for a new synchronisation point.
+NB: Subunit V1 and V2 packets may legitimately included 0xB3 internally,
+as they are an 8-bit safe container format, so recovery from this situation
+may involve an arbitrary number of false positives until an actual packet
+is encountered : and even then it may still be false, failing after passing
+the version check due to coincidence.
+
+Flags are stored in network byte order too.
++-------------------------+------------------------+
+| High byte | Low byte |
+| 15 14 13 12 11 10 9 8 | 7 6 5 4 3 2 1 0 |
+| VERSION |feature bits| |
++------------+------------+------------------------+
+
+Valid version values are:
+0x2 - version 2
+
+Feature bits:
+Bit 11 - mask 0x0800 - Test id present.
+Bit 10 - mask 0x0400 - Routing code present.
+Bit 9 - mask 0x0200 - Timestamp present.
+Bit 8 - mask 0x0100 - Test is 'runnable'.
+Bit 7 - mask 0x0080 - Tags are present.
+Bit 6 - mask 0x0040 - File content is present.
+Bit 5 - mask 0x0020 - File MIME type is present.
+Bit 4 - mask 0x0010 - EOF marker.
+Bit 3 - mask 0x0008 - Must be zero in version 2.
+
+Test status gets three bits:
+Bit 2 | Bit 1 | Bit 0 - mask 0x0007 - A test status enum lookup:
+000 - undefined / no test
+001 - Enumeration / existence
+002 - In progress
+003 - Success
+004 - Unexpected Success
+005 - Skipped
+006 - Failed
+007 - Expected failure
+
+After the flags field is a number field giving the length in bytes for the
+entire packet including the signature and the checksum. This length must
+be less than 4MiB - 4194303 bytes. The encoding can obviously record a larger
+number but one of the goals is to avoid requiring large buffers, or causing
+large latency in the packet forward/processing pipeline. Larger file
+attachments can be communicated in multiple packets, and the overhead in such a
+4MiB packet is approximately 0.2%.
+
+The rest of the packet is a series of optional features as specified by the set
+feature bits in the flags field. When absent they are entirely absent.
+
+Forwarding and multiplexing of packets can be done without interpreting the
+remainder of the packet until the routing code and checksum (which are both at
+the end of the packet). Additionally, routers can often avoid copying or moving
+the bulk of the packet, as long as the routing code size increase doesn't force
+the length encoding to take up a new byte (which will only happen to packets
+less than or equal to 16KiB in length) - large packets are very efficient to
+route.
+
+Timestamp when present is a 32 bit unsigned integer for secnods, and a variable
+length number for nanoseconds, representing UTC time since Unix Epoch in
+seconds and nanoseconds.
+
+Test id when present is a UTF-8 string. The test id should uniquely identify
+runnable tests such that they can be selected individually. For tests and other
+actions which cannot be individually run (such as test
+fixtures/layers/subtests) uniqueness is not required (though being human
+meaningful is highly recommended).
+
+Tags when present is a length prefixed vector of UTF-8 strings, one per tag.
+There are no restrictions on tag content (other than the restrictions on UTF-8
+strings in subunit in general). Tags have no ordering.
+
+When a MIME type is present, it defines the MIME type for the file across all
+packets same file (routing code + testid + name uniquely identifies a file,
+reset when EOF is flagged). If a file never has a MIME type set, it should be
+treated as application/octet-stream.
+
+File content when present is a UTF-8 string for the name followed by the length
+in bytes of the content, and then the content octets.
+
+If present routing code is a UTF-8 string. The routing code is used to
+determine which test backend a test was running on when doing data analysis,
+and to route stdin to the test process if interaction is required.
+
+Multiplexers SHOULD add a routing code if none is present, and prefix any
+existing routing code with a routing code ('/' separated) if one is already
+present. For example, a multiplexer might label each stream it is multiplexing
+with a simple ordinal ('0', '1' etc), and given an incoming packet with route
+code '3' from stream '0' would adjust the route code when forwarding the packet
+to be '0/3'.
+
+Following the end of the packet is a CRC-32 checksum of the contents of the
+packet including the signature.
+
+Example packets
+~~~~~~~~~~~~~~~
+
+Trivial test "foo" enumeration packet, with test id, runnable set,
+status=enumeration. Spaces below are to visually break up signature / flags /
+length / testid / crc32
+
+b3 2901 0c 03666f6f 08555f1b
+
+
+Version 1 (and 1.1)
+===================
+
+Version 1 (and 1.1) are mostly human readable protocols.
+
Sample subunit wire contents
----------------------------
@@ -163,6 +401,7 @@ tags: [-]TAG ...
time: YYYY-MM-DD HH:MM:SSZ
LABEL: UTF8*
+NAME: UTF8*
DETAILS ::= BRACKETED | MULTIPART
BRACKETED ::= '[' CR UTF8-lines ']' CR
MULTIPART ::= '[ multipart' CR PART* ']' CR
@@ -223,7 +462,9 @@ Releases
========
* Update versions in configure.ac and python/subunit/__init__.py.
-* Make PyPI and regular tarball releases. Upload the regular one to LP, the
- PyPI one to PyPI.
+* Update NEWS.
+* Do a make distcheck, which will update Makefile etc.
+* Do a PyPI release: PYTHONPATH=../../python python ../../setup.py sdist upload -s
+* Upload the regular one to LP.
* Push a tagged commit.
diff --git a/lib/subunit/all_tests.py b/lib/subunit/all_tests.py
new file mode 100644
index 0000000..23fd65d
--- /dev/null
+++ b/lib/subunit/all_tests.py
@@ -0,0 +1,36 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Robert Collins <robertc at 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
+
+
+class ShellTests(subunit.ExecTestCase):
+
+ def test_sourcing(self):
+ """./shell/tests/test_source_library.sh"""
+
+ def test_functions(self):
+ """./shell/tests/test_function_output.sh"""
+
+
+def test_suite():
+ result = unittest.TestSuite()
+ result.addTest(subunit.test_suite())
+ result.addTest(ShellTests('test_sourcing'))
+ result.addTest(ShellTests('test_functions'))
+ return result
diff --git a/lib/subunit/configure.ac b/lib/subunit/configure.ac
index cf21d55..ef1a048 100644
--- a/lib/subunit/configure.ac
+++ b/lib/subunit/configure.ac
@@ -1,6 +1,6 @@
m4_define([SUBUNIT_MAJOR_VERSION], [0])
m4_define([SUBUNIT_MINOR_VERSION], [0])
-m4_define([SUBUNIT_MICRO_VERSION], [9])
+m4_define([SUBUNIT_MICRO_VERSION], [21])
m4_define([SUBUNIT_VERSION],
m4_defn([SUBUNIT_MAJOR_VERSION]).m4_defn([SUBUNIT_MINOR_VERSION]).m4_defn([SUBUNIT_MICRO_VERSION]))
AC_PREREQ([2.59])
@@ -19,6 +19,7 @@ AC_SUBST([SUBUNIT_VERSION])
AC_USE_SYSTEM_EXTENSIONS
AC_PROG_CC
AC_PROG_CXX
+m4_ifdef([AM_PROG_AR], [AM_PROG_AR])
AM_PROG_CC_C_O
AC_PROG_INSTALL
AC_PROG_LN_S
diff --git a/lib/subunit/filters/subunit-1to2 b/lib/subunit/filters/subunit-1to2
new file mode 100755
index 0000000..9725820
--- /dev/null
+++ b/lib/subunit/filters/subunit-1to2
@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Robert Collins <robertc at 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.
+#
+
+"""Convert a version 1 subunit stream to version 2 stream."""
+
+from optparse import OptionParser
+import sys
+
+from testtools import ExtendedToStreamDecorator
+
+from subunit import StreamResultToBytes
+from subunit.filters import find_stream, run_tests_from_stream
+
+
+def make_options(description):
+ parser = OptionParser(description=__doc__)
+ return parser
+
+
+def main():
+ parser = make_options(__doc__)
+ (options, args) = parser.parse_args()
+ run_tests_from_stream(find_stream(sys.stdin, args),
+ ExtendedToStreamDecorator(StreamResultToBytes(sys.stdout)))
+ sys.exit(0)
+
+
+if __name__ == '__main__':
+ main()
diff --git a/lib/subunit/filters/subunit-2to1 b/lib/subunit/filters/subunit-2to1
new file mode 100755
index 0000000..0072307
--- /dev/null
+++ b/lib/subunit/filters/subunit-2to1
@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Robert Collins <robertc at 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.
+#
+
+"""Convert a version 2 subunit stream to a version 1 stream."""
+
+from optparse import OptionParser
+import sys
+
+from testtools import StreamToExtendedDecorator
+
+from subunit import ByteStreamToStreamResult, TestProtocolClient
+from subunit.filters import find_stream, run_tests_from_stream
+
+
+def make_options(description):
+ parser = OptionParser(description=__doc__)
+ return parser
+
+
+def main():
+ parser = make_options(__doc__)
+ (options, args) = parser.parse_args()
+ case = ByteStreamToStreamResult(
+ find_stream(sys.stdin, args), non_subunit_name='stdout')
+ result = StreamToExtendedDecorator(TestProtocolClient(sys.stdout))
+ # What about stdout chunks?
+ result.startTestRun()
+ case.run(result)
+ result.stopTestRun()
+ sys.exit(0)
+
+
+if __name__ == '__main__':
+ main()
diff --git a/lib/subunit/filters/subunit-filter b/lib/subunit/filters/subunit-filter
index 6a1ecc9..e9e2bb0 100755
--- a/lib/subunit/filters/subunit-filter
+++ b/lib/subunit/filters/subunit-filter
@@ -1,6 +1,6 @@
#!/usr/bin/env python
# subunit: extensions to python unittest to get test results from subprocesses.
-# Copyright (C) 2008 Robert Collins <robertc at robertcollins.net>
+# Copyright (C) 200-2013 Robert Collins <robertc at robertcollins.net>
# (C) 2009 Martin Pool
#
# Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
@@ -30,13 +30,15 @@ from optparse import OptionParser
import sys
import re
+from testtools import ExtendedToStreamDecorator, StreamToExtendedDecorator
+
from subunit import (
DiscardStream,
ProtocolTestCase,
- TestProtocolClient,
+ StreamResultToBytes,
read_test_list,
)
-from subunit.filters import filter_by_result
+from subunit.filters import filter_by_result, find_stream
from subunit.test_results import (
and_predicates,
make_tag_filter,
@@ -55,9 +57,11 @@ def make_options(description):
parser.add_option("-f", "--no-failure", action="store_true",
help="exclude failures", dest="failure")
parser.add_option("--passthrough", action="store_false",
- help="Show all non subunit input.", default=False, dest="no_passthrough")
+ help="Forward non-subunit input as 'stdout'.", default=False,
+ dest="no_passthrough")
parser.add_option("--no-passthrough", action="store_true",
- help="Hide all non subunit input.", default=False, dest="no_passthrough")
+ help="Discard all non subunit input.", default=False,
+ dest="no_passthrough")
parser.add_option("-s", "--success", action="store_false",
help="include successes", dest="success")
parser.add_option("--no-success", action="store_true",
@@ -126,15 +130,16 @@ def _make_result(output, options, predicate):
fixup_expected_failures = set()
for path in options.fixup_expected_failures or ():
fixup_expected_failures.update(read_test_list(path))
- return TestResultFilter(
- TestProtocolClient(output),
+ return StreamToExtendedDecorator(TestResultFilter(
+ ExtendedToStreamDecorator(
+ StreamResultToBytes(output)),
filter_error=options.error,
filter_failure=options.failure,
filter_success=options.success,
filter_skip=options.skip,
filter_xfail=options.xfail,
filter_predicate=predicate,
- fixup_expected_failures=fixup_expected_failures)
+ fixup_expected_failures=fixup_expected_failures))
def main():
@@ -150,7 +155,9 @@ def main():
lambda output_to: _make_result(sys.stdout, options, filter_predicate),
output_path=None,
passthrough=(not options.no_passthrough),
- forward=False)
+ forward=False,
+ protocol_version=2,
+ input_stream=find_stream(sys.stdin, args))
sys.exit(0)
diff --git a/lib/subunit/filters/subunit-ls b/lib/subunit/filters/subunit-ls
index 82db4c3..8c6a1e7 100755
--- a/lib/subunit/filters/subunit-ls
+++ b/lib/subunit/filters/subunit-ls
@@ -19,9 +19,14 @@
from optparse import OptionParser
import sys
-from subunit import DiscardStream, ProtocolTestCase
+from testtools import (
+ CopyStreamResult, StreamToExtendedDecorator, StreamResultRouter,
+ StreamSummary)
+
+from subunit import ByteStreamToStreamResult
+from subunit.filters import find_stream, run_tests_from_stream
from subunit.test_results import (
- AutoTimingTestResultDecorator,
+ CatFiles,
TestIdPrintingResult,
)
@@ -30,18 +35,25 @@ parser = OptionParser(description=__doc__)
parser.add_option("--times", action="store_true",
help="list the time each test took (requires a timestamped stream)",
default=False)
+parser.add_option("--exists", action="store_true",
+ help="list tests that are reported as existing (as well as ran)",
+ default=False)
parser.add_option("--no-passthrough", action="store_true",
help="Hide all non subunit input.", default=False, dest="no_passthrough")
(options, args) = parser.parse_args()
-result = AutoTimingTestResultDecorator(
- TestIdPrintingResult(sys.stdout, options.times))
-if options.no_passthrough:
- passthrough_stream = DiscardStream()
-else:
- passthrough_stream = None
-test = ProtocolTestCase(sys.stdin, passthrough=passthrough_stream)
+test = ByteStreamToStreamResult(
+ find_stream(sys.stdin, args), non_subunit_name="stdout")
+result = TestIdPrintingResult(sys.stdout, options.times, options.exists)
+if not options.no_passthrough:
+ result = StreamResultRouter(result)
+ cat = CatFiles(sys.stdout)
+ result.add_rule(cat, 'test_id', test_id=None)
+summary = StreamSummary()
+result = CopyStreamResult([result, summary])
+result.startTestRun()
test.run(result)
-if result.wasSuccessful():
+result.stopTestRun()
+if summary.wasSuccessful():
exit_code = 0
else:
exit_code = 1
diff --git a/lib/subunit/filters/subunit-notify b/lib/subunit/filters/subunit-notify
index 8cce2d1..bc833da 100755
--- a/lib/subunit/filters/subunit-notify
+++ b/lib/subunit/filters/subunit-notify
@@ -19,6 +19,7 @@
import pygtk
pygtk.require('2.0')
import pynotify
+from testtools import StreamToExtendedDecorator
from subunit import TestResultStats
from subunit.filters import run_filter_script
@@ -28,6 +29,7 @@ if not pynotify.init("Subunit-notify"):
def notify_of_result(result):
+ result = result.decorated
if result.failed_tests > 0:
summary = "Test run failed"
else:
@@ -41,4 +43,6 @@ def notify_of_result(result):
nw.show()
-run_filter_script(TestResultStats, __doc__, notify_of_result)
+run_filter_script(
+ lambda output:StreamToExtendedDecorator(TestResultStats(output)),
+ __doc__, notify_of_result, protocol_version=2)
diff --git a/lib/subunit/filters/subunit-output b/lib/subunit/filters/subunit-output
new file mode 100644
index 0000000..61e5d11
--- /dev/null
+++ b/lib/subunit/filters/subunit-output
@@ -0,0 +1,23 @@
+#!/usr/bin/env python
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Subunit Contributors
+#
+# Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
+# license at the users choice. A copy of both licenses are available in the
+# project source as Apache-2.0 and BSD. You may not use this file except in
+# compliance with one of these two licences.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# license you chose for the specific language governing permissions and
+# limitations under that license.
+
+
+"""A command-line tool to generate a subunit result byte-stream."""
+
+from subunit._output import output_main
+
+
+if __name__ == '__main__':
+ exit(output_main())
diff --git a/lib/subunit/filters/subunit-stats b/lib/subunit/filters/subunit-stats
index 4734988..79733b0 100755
--- a/lib/subunit/filters/subunit-stats
+++ b/lib/subunit/filters/subunit-stats
@@ -16,26 +16,17 @@
"""Filter a subunit stream to get aggregate statistics."""
-from optparse import OptionParser
import sys
-import unittest
-from subunit import DiscardStream, ProtocolTestCase, TestResultStats
+from testtools import StreamToExtendedDecorator
+
+from subunit import TestResultStats
+from subunit.filters import run_filter_script
+
-parser = OptionParser(description=__doc__)
-parser.add_option("--no-passthrough", action="store_true",
- help="Hide all non subunit input.", default=False, dest="no_passthrough")
-(options, args) = parser.parse_args()
result = TestResultStats(sys.stdout)
-if options.no_passthrough:
- passthrough_stream = DiscardStream()
-else:
- passthrough_stream = None
-test = ProtocolTestCase(sys.stdin, passthrough=passthrough_stream)
-test.run(result)
-result.formatStats()
-if result.wasSuccessful():
- exit_code = 0
-else:
- exit_code = 1
-sys.exit(exit_code)
+def show_stats(r):
+ r.decorated.formatStats()
+run_filter_script(
+ lambda output:StreamToExtendedDecorator(result),
+ __doc__, show_stats, protocol_version=2, passthrough_subunit=False)
diff --git a/lib/subunit/filters/subunit-tags b/lib/subunit/filters/subunit-tags
index edbbfce..1022492 100755
--- a/lib/subunit/filters/subunit-tags
+++ b/lib/subunit/filters/subunit-tags
@@ -23,4 +23,5 @@ subunit-tags foo -bar -> adds foo and removes bar
import sys
from subunit import tag_stream
+
sys.exit(tag_stream(sys.stdin, sys.stdout, sys.argv[1:]))
diff --git a/lib/subunit/filters/subunit2csv b/lib/subunit/filters/subunit2csv
index 14620ff..4adf5cd 100755
--- a/lib/subunit/filters/subunit2csv
+++ b/lib/subunit/filters/subunit2csv
@@ -16,8 +16,11 @@
"""Turn a subunit stream into a CSV"""
+from testtools import StreamToExtendedDecorator
+
from subunit.filters import run_filter_script
from subunit.test_results import CsvResult
-run_filter_script(CsvResult, __doc__)
+run_filter_script(lambda output:StreamToExtendedDecorator(CsvResult(output)),
+ __doc__, protocol_version=2)
diff --git a/lib/subunit/filters/subunit2gtk b/lib/subunit/filters/subunit2gtk
index c2cb2de..78b4309 100755
--- a/lib/subunit/filters/subunit2gtk
+++ b/lib/subunit/filters/subunit2gtk
@@ -46,17 +46,20 @@
"""Display a subunit stream in a gtk progress window."""
import sys
+import threading
import unittest
import pygtk
pygtk.require('2.0')
import gtk, gtk.gdk, gobject
+from testtools import StreamToExtendedDecorator
+
from subunit import (
PROGRESS_POP,
PROGRESS_PUSH,
PROGRESS_SET,
- TestProtocolServer,
+ ByteStreamToStreamResult,
)
from subunit.progress_model import ProgressModel
@@ -139,6 +142,9 @@ class GTKTestResult(unittest.TestResult):
def stopTest(self, test):
super(GTKTestResult, self).stopTest(test)
+ gobject.idle_add(self._stopTest)
+
+ def _stopTest(self):
self.progress_model.advance()
if self.progress_model.width() == 0:
self.pbar.pulse()
@@ -153,26 +159,26 @@ class GTKTestResult(unittest.TestResult):
super(GTKTestResult, self).stopTestRun()
except AttributeError:
pass
- self.pbar.set_text('Finished')
+ gobject.idle_add(self.pbar.set_text, 'Finished')
def addError(self, test, err):
super(GTKTestResult, self).addError(test, err)
- self.update_counts()
+ gobject.idle_add(self.update_counts)
def addFailure(self, test, err):
super(GTKTestResult, self).addFailure(test, err)
- self.update_counts()
+ gobject.idle_add(self.update_counts)
def addSuccess(self, test):
super(GTKTestResult, self).addSuccess(test)
- self.update_counts()
+ gobject.idle_add(self.update_counts)
def addSkip(self, test, reason):
# addSkip is new in Python 2.7/3.1
addSkip = getattr(super(GTKTestResult, self), 'addSkip', None)
if callable(addSkip):
addSkip(test, reason)
- self.update_counts()
+ gobject.idle_add(self.update_counts)
def addExpectedFailure(self, test, err):
# addExpectedFailure is new in Python 2.7/3.1
@@ -180,7 +186,7 @@ class GTKTestResult(unittest.TestResult):
'addExpectedFailure', None)
if callable(addExpectedFailure):
addExpectedFailure(test, err)
- self.update_counts()
+ gobject.idle_add(self.update_counts)
def addUnexpectedSuccess(self, test):
# addUnexpectedSuccess is new in Python 2.7/3.1
@@ -188,7 +194,7 @@ class GTKTestResult(unittest.TestResult):
'addUnexpectedSuccess', None)
if callable(addUnexpectedSuccess):
addUnexpectedSuccess(test)
- self.update_counts()
+ gobject.idle_add(self.update_counts)
def progress(self, offset, whence):
if whence == PROGRESS_PUSH:
@@ -212,47 +218,22 @@ class GTKTestResult(unittest.TestResult):
self.ok_label.set_text(str(self.testsRun - bad))
self.not_ok_label.set_text(str(bad))
-
-class GIOProtocolTestCase(object):
-
- def __init__(self, stream, result, on_finish):
- self.stream = stream
- self.schedule_read()
- self.hup_id = gobject.io_add_watch(stream, gobject.IO_HUP, self.hup)
- self.protocol = TestProtocolServer(result)
- self.on_finish = on_finish
-
- def read(self, source, condition, all=False):
- #NB: \o/ actually blocks
- line = source.readline()
- if not line:
- self.protocol.lostConnection()
- self.on_finish()
- return False
- self.protocol.lineReceived(line)
- # schedule more IO shortly - if we say we're willing to do it
- # immediately we starve things.
- if not all:
- source_id = gobject.timeout_add(1, self.schedule_read)
- return False
- else:
- return True
-
- def schedule_read(self):
- self.read_id = gobject.io_add_watch(self.stream, gobject.IO_IN, self.read)
-
- def hup(self, source, condition):
- while self.read(source, condition, all=True): pass
- self.protocol.lostConnection()
- gobject.source_remove(self.read_id)
- self.on_finish()
- return False
-
-
-result = GTKTestResult()
-test = GIOProtocolTestCase(sys.stdin, result, result.stopTestRun)
+gobject.threads_init()
+result = StreamToExtendedDecorator(GTKTestResult())
+test = ByteStreamToStreamResult(sys.stdin, non_subunit_name='stdout')
+# Get setup
+while gtk.events_pending():
+ gtk.main_iteration()
+# Start IO
+def run_and_finish():
+ test.run(result)
+ result.stopTestRun()
+t = threading.Thread(target=run_and_finish)
+t.daemon = True
+result.startTestRun()
+t.start()
gtk.main()
-if result.wasSuccessful():
+if result.decorated.wasSuccessful():
exit_code = 0
else:
exit_code = 1
diff --git a/lib/subunit/filters/subunit2junitxml b/lib/subunit/filters/subunit2junitxml
index d568c71..8e827d5 100755
--- a/lib/subunit/filters/subunit2junitxml
+++ b/lib/subunit/filters/subunit2junitxml
@@ -18,6 +18,9 @@
import sys
+
+from testtools import StreamToExtendedDecorator
+
from subunit.filters import run_filter_script
try:
@@ -28,4 +31,6 @@ except ImportError:
raise
-run_filter_script(JUnitXmlResult, __doc__)
+run_filter_script(
+ lambda output:StreamToExtendedDecorator(JUnitXmlResult(output)), __doc__,
+ protocol_version=2)
diff --git a/lib/subunit/filters/subunit2pyunit b/lib/subunit/filters/subunit2pyunit
index 83a23d1..d10ceea 100755
--- a/lib/subunit/filters/subunit2pyunit
+++ b/lib/subunit/filters/subunit2pyunit
@@ -16,11 +16,16 @@
"""Display a subunit stream through python's unittest test runner."""
+from operator import methodcaller
from optparse import OptionParser
import sys
import unittest
-from subunit import DiscardStream, ProtocolTestCase, TestProtocolServer
+from testtools import StreamToExtendedDecorator, DecorateTestCaseResult, StreamResultRouter
+
+from subunit import ByteStreamToStreamResult
+from subunit.filters import find_stream
+from subunit.test_results import CatFiles
parser = OptionParser(description=__doc__)
parser.add_option("--no-passthrough", action="store_true",
@@ -29,11 +34,17 @@ parser.add_option("--progress", action="store_true",
help="Use bzrlib's test reporter (requires bzrlib)",
default=False)
(options, args) = parser.parse_args()
-if options.no_passthrough:
- passthrough_stream = DiscardStream()
-else:
- passthrough_stream = None
-test = ProtocolTestCase(sys.stdin, passthrough=passthrough_stream)
+test = ByteStreamToStreamResult(
+ find_stream(sys.stdin, args), non_subunit_name='stdout')
+def wrap_result(result):
+ result = StreamToExtendedDecorator(result)
+ if not options.no_passthrough:
+ result = StreamResultRouter(result)
+ result.add_rule(CatFiles(sys.stdout), 'test_id', test_id=None)
+ return result
+test = DecorateTestCaseResult(test, wrap_result,
+ before_run=methodcaller('startTestRun'),
+ after_run=methodcaller('stopTestRun'))
if options.progress:
from bzrlib.tests import TextTestRunner
from bzrlib import ui
diff --git a/lib/subunit/perl/Makefile.PL.in b/lib/subunit/perl/Makefile.PL.in
index cf5e6c4..90a6a5e 100755
--- a/lib/subunit/perl/Makefile.PL.in
+++ b/lib/subunit/perl/Makefile.PL.in
@@ -1,6 +1,6 @@
use ExtUtils::MakeMaker;
WriteMakefile(
- 'INSTALL_BASE' => '@prefix@',
+ 'PREFIX' => '@prefix@',
'NAME' => 'Subunit',
'VERSION' => '@SUBUNIT_VERSION@',
'test' => { 'TESTS' => 'tests/*.pl' },
@@ -12,7 +12,7 @@ sub MY::postamble {
check: # test
uninstall_distcheck:
- rm -fr $(DESTINSTALLARCHLIB)
+ find $(DESTDIR)$(INSTALLSITEARCH) -type f -exec rm {} \;
rm MYMETA.yml
VPATH = @srcdir@
diff --git a/lib/subunit/python/subunit/__init__.py b/lib/subunit/python/subunit/__init__.py
index 42dcf29..ca873b3 100644
--- a/lib/subunit/python/subunit/__init__.py
+++ b/lib/subunit/python/subunit/__init__.py
@@ -121,31 +121,25 @@ import re
import subprocess
import sys
import unittest
-if sys.version_info > (3, 0):
+try:
from io import UnsupportedOperation as _UnsupportedOperation
-else:
+except ImportError:
_UnsupportedOperation = AttributeError
-
+from extras import safe_hasattr
from testtools import content, content_type, ExtendedToOriginalDecorator
from testtools.content import TracebackContent
from testtools.compat import _b, _u, BytesIO, StringIO
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 testtools import testresult, CopyStreamResult
from subunit import chunked, details, iso8601, test_results
+from subunit.v2 import ByteStreamToStreamResult, StreamResultToBytes
# same format as sys.version_info: "A tuple containing the five components of
# the version number: major, minor, micro, releaselevel, and serial. All
@@ -159,7 +153,7 @@ from subunit import chunked, details, iso8601, test_results
# 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)
+__version__ = (0, 0, 21, 'final', 0)
PROGRESS_SET = 0
PROGRESS_CUR = 1
@@ -624,7 +618,7 @@ class TestProtocolClient(testresult.TestResult):
def __init__(self, stream):
testresult.TestResult.__init__(self)
- stream = _make_stream_binary(stream)
+ stream = make_stream_binary(stream)
self._stream = stream
self._progress_fmt = _b("progress: ")
self._bytes_eol = _b("\n")
@@ -921,7 +915,7 @@ class ExecTestCase(unittest.TestCase):
protocol = TestProtocolServer(result)
process = subprocess.Popen(self.script, shell=True,
stdout=subprocess.PIPE)
- _make_stream_binary(process.stdout)
+ make_stream_binary(process.stdout)
output = process.communicate()[0]
protocol.readFrom(BytesIO(output))
@@ -992,44 +986,51 @@ def run_isolated(klass, self, result):
return result
-def TAP2SubUnit(tap, subunit):
+def TAP2SubUnit(tap, output_stream):
"""Filter a TAP pipe into a subunit pipe.
- :param tap: A tap pipe/stream/file object.
+ This should be invoked once per TAP script, as TAP scripts get
+ mapped to a single runnable case with multiple components.
+
+ :param tap: A tap pipe/stream/file object - should emit unicode strings.
:param subunit: A pipe/stream/file object to write subunit results to.
:return: The exit code to exit with.
"""
+ output = StreamResultToBytes(output_stream)
+ UTF8_TEXT = 'text/plain; charset=UTF8'
BEFORE_PLAN = 0
AFTER_PLAN = 1
SKIP_STREAM = 2
state = BEFORE_PLAN
plan_start = 1
plan_stop = 0
- def _skipped_test(subunit, plan_start):
- # Some tests were skipped.
- subunit.write('test test %d\n' % plan_start)
- subunit.write('error test %d [\n' % plan_start)
- subunit.write('test missing from TAP output\n')
- subunit.write(']\n')
- return plan_start + 1
# Test data for the next test to emit
test_name = None
log = []
result = None
+ def missing_test(plan_start):
+ output.status(test_id='test %d' % plan_start,
+ test_status='fail', runnable=False,
+ mime_type=UTF8_TEXT, eof=True, file_name="tap meta",
+ file_bytes=b"test missing from TAP output")
def _emit_test():
"write out a test"
if test_name is None:
return
- subunit.write("test %s\n" % test_name)
- if not log:
- subunit.write("%s %s\n" % (result, test_name))
- else:
- subunit.write("%s %s [\n" % (result, test_name))
if log:
- for line in log:
- subunit.write("%s\n" % line)
- subunit.write("]\n")
+ log_bytes = b'\n'.join(log_line.encode('utf8') for log_line in log)
+ mime_type = UTF8_TEXT
+ file_name = 'tap comment'
+ eof = True
+ else:
+ log_bytes = None
+ mime_type = None
+ file_name = None
+ eof = True
del log[:]
+ output.status(test_id=test_name, test_status=result,
+ file_bytes=log_bytes, mime_type=mime_type, eof=eof,
+ file_name=file_name, runnable=False)
for line in tap:
if state == BEFORE_PLAN:
match = re.match("(\d+)\.\.(\d+)\s*(?:\#\s+(.*))?\n", line)
@@ -1040,10 +1041,9 @@ def TAP2SubUnit(tap, subunit):
if plan_start > plan_stop and plan_stop == 0:
# skipped file
state = SKIP_STREAM
- subunit.write("test file skip\n")
- subunit.write("skip file skip [\n")
- subunit.write("%s\n" % comment)
- subunit.write("]\n")
+ output.status(test_id='file skip', test_status='skip',
+ file_bytes=comment.encode('utf8'), eof=True,
+ file_name='tap comment')
continue
# not a plan line, or have seen one before
match = re.match("(ok|not ok)(?:\s+(\d+)?)?(?:\s+([^#]*[^#\s]+)\s*)?(?:\s+#\s+(TODO|SKIP|skip|todo)(?:\s+(.*))?)?\n", line)
@@ -1054,7 +1054,7 @@ def TAP2SubUnit(tap, subunit):
if status == 'ok':
result = 'success'
else:
- result = "failure"
+ result = "fail"
if description is None:
description = ''
else:
@@ -1069,7 +1069,8 @@ def TAP2SubUnit(tap, subunit):
if number is not None:
number = int(number)
while plan_start < number:
- plan_start = _skipped_test(subunit, plan_start)
+ missing_test(plan_start)
+ plan_start += 1
test_name = "test %d%s" % (plan_start, description)
plan_start += 1
continue
@@ -1082,18 +1083,21 @@ def TAP2SubUnit(tap, subunit):
extra = ' %s' % reason
_emit_test()
test_name = "Bail out!%s" % extra
- result = "error"
+ result = "fail"
state = SKIP_STREAM
continue
match = re.match("\#.*\n", line)
if match:
log.append(line[:-1])
continue
- subunit.write(line)
+ # Should look at buffering status and binding this to the prior result.
+ output.status(file_bytes=line.encode('utf8'), file_name='stdout',
+ mime_type=UTF8_TEXT)
_emit_test()
while plan_start <= plan_stop:
# record missed tests
- plan_start = _skipped_test(subunit, plan_start)
+ missing_test(plan_start)
+ plan_start += 1
return 0
@@ -1121,24 +1125,21 @@ def tag_stream(original, filtered, tags):
:return: 0
"""
new_tags, gone_tags = tags_to_new_gone(tags)
- def write_tags(new_tags, gone_tags):
- if new_tags or gone_tags:
- filtered.write("tags: " + ' '.join(new_tags))
- if gone_tags:
- for tag in gone_tags:
- filtered.write("-" + tag)
- filtered.write("\n")
- write_tags(new_tags, gone_tags)
- # TODO: use the protocol parser and thus don't mangle test comments.
- for line in original:
- if line.startswith("tags:"):
- line_tags = line[5:].split()
- line_new, line_gone = tags_to_new_gone(line_tags)
- line_new = line_new - gone_tags
- line_gone = line_gone - new_tags
- write_tags(line_new, line_gone)
- else:
- filtered.write(line)
+ source = ByteStreamToStreamResult(original, non_subunit_name='stdout')
+ class Tagger(CopyStreamResult):
+ def status(self, **kwargs):
+ tags = kwargs.get('test_tags')
+ if not tags:
+ tags = set()
+ tags.update(new_tags)
+ tags.difference_update(gone_tags)
+ if tags:
+ kwargs['test_tags'] = tags
+ else:
+ kwargs['test_tags'] = None
+ super(Tagger, self).status(**kwargs)
+ output = Tagger([StreamResultToBytes(filtered)])
+ source.run(output)
return 0
@@ -1177,11 +1178,11 @@ class ProtocolTestCase(object):
:param forward: A stream to pass subunit input on to. If not supplied
subunit input is not forwarded.
"""
- stream = _make_stream_binary(stream)
+ stream = make_stream_binary(stream)
self._stream = stream
self._passthrough = passthrough
if forward is not None:
- forward = _make_stream_binary(forward)
+ forward = make_stream_binary(forward)
self._forward = forward
def __call__(self, result=None):
@@ -1260,7 +1261,8 @@ def get_default_formatter():
else:
stream = sys.stdout
if sys.version_info > (3, 0):
- stream = stream.buffer
+ if safe_hasattr(stream, 'buffer'):
+ stream = stream.buffer
return stream
@@ -1277,7 +1279,7 @@ def read_test_list(path):
f.close()
-def _make_stream_binary(stream):
+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
@@ -1285,12 +1287,13 @@ def _make_stream_binary(stream):
"""
try:
fileno = stream.fileno()
- except _UnsupportedOperation:
+ except (_UnsupportedOperation, AttributeError):
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":
@@ -1301,14 +1304,17 @@ def _make_binary_on_windows(fileno):
def _unwrap_text(stream):
"""Unwrap stream if it is a text stream to get the original buffer."""
if sys.version_info > (3, 0):
+ unicode_type = str
+ else:
+ unicode_type = unicode
+ try:
+ # Read streams
+ if type(stream.read(0)) is unicode_type:
+ return stream.buffer
+ except (_UnsupportedOperation, IOError):
+ # Cannot read from the stream: try via writes
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
+ stream.write(_b(''))
+ except TypeError:
+ return stream.buffer
return stream
diff --git a/lib/subunit/python/subunit/_output.py b/lib/subunit/python/subunit/_output.py
new file mode 100644
index 0000000..aa92646
--- /dev/null
+++ b/lib/subunit/python/subunit/_output.py
@@ -0,0 +1,203 @@
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Subunit Contributors
+#
+# 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
+from functools import partial
+from optparse import (
+ OptionGroup,
+ OptionParser,
+ OptionValueError,
+)
+import sys
+
+from subunit import make_stream_binary
+from subunit.iso8601 import UTC
+from subunit.v2 import StreamResultToBytes
+
+
+_FINAL_ACTIONS = frozenset([
+ 'exists',
+ 'fail',
+ 'skip',
+ 'success',
+ 'uxsuccess',
+ 'xfail',
+])
+_ALL_ACTIONS = _FINAL_ACTIONS.union(['inprogress'])
+_CHUNK_SIZE=3670016 # 3.5 MiB
+
+
+def output_main():
+ args = parse_arguments()
+ output = StreamResultToBytes(sys.stdout)
+ generate_stream_results(args, output)
+ return 0
+
+
+def parse_arguments(args=None, ParserClass=OptionParser):
+ """Parse arguments from the command line.
+
+ If specified, args must be a list of strings, similar to sys.argv[1:].
+
+ ParserClass may be specified to override the class we use to parse the
+ command-line arguments. This is useful for testing.
+ """
+ parser = ParserClass(
+ prog="subunit-output",
+ description="A tool to generate a subunit v2 result byte-stream",
+ usage="subunit-output [-h] [status TEST_ID] [options]",
+ )
+ parser.set_default('tags', None)
+ parser.set_default('test_id', None)
+
+ status_commands = OptionGroup(
+ parser,
+ "Status Commands",
+ "These options report the status of a test. TEST_ID must be a string "
+ "that uniquely identifies the test."
+ )
+ for action_name in _ALL_ACTIONS:
+ status_commands.add_option(
+ "--%s" % action_name,
+ nargs=1,
+ action="callback",
+ callback=set_status_cb,
+ callback_args=(action_name,),
+ dest="action",
+ metavar="TEST_ID",
+ help="Report a test status."
+ )
+ parser.add_option_group(status_commands)
+
+ file_commands = OptionGroup(
+ parser,
+ "File Options",
+ "These options control attaching data to a result stream. They can "
+ "either be specified with a status command, in which case the file "
+ "is attached to the test status, or by themselves, in which case "
+ "the file is attached to the stream (and not associated with any "
+ "test id)."
+ )
+ file_commands.add_option(
+ "--attach-file",
+ help="Attach a file to the result stream for this test. If '-' is "
+ "specified, stdin will be read instead. In this case, the file "
+ "name will be set to 'stdin' (but can still be overridden with "
+ "the --file-name option)."
+ )
+ file_commands.add_option(
+ "--file-name",
+ help="The name to give this file attachment. If not specified, the "
+ "name of the file on disk will be used, or 'stdin' in the case "
+ "where '-' was passed to the '--attach-file' argument. This option"
+ " may only be specified when '--attach-file' is specified.",
+ )
+ file_commands.add_option(
+ "--mimetype",
+ help="The mime type to send with this file. This is only used if the "
+ "--attach-file argument is used. This argument is optional. If it "
+ "is not specified, the file will be sent without a mime type. This "
+ "option may only be specified when '--attach-file' is specified.",
+ default=None
+ )
+ parser.add_option_group(file_commands)
+
+ parser.add_option(
+ "--tag",
+ help="Specifies a tag. May be used multiple times",
+ action="append",
+ dest="tags",
+ default=[]
+ )
+
+ (options, args) = parser.parse_args(args)
+ if options.mimetype and not options.attach_file:
+ parser.error("Cannot specify --mimetype without --attach-file")
+ if options.file_name and not options.attach_file:
+ parser.error("Cannot specify --file-name without --attach-file")
+ if options.attach_file:
+ if options.attach_file == '-':
+ if not options.file_name:
+ options.file_name = 'stdin'
+ options.attach_file = make_stream_binary(sys.stdin)
+ else:
+ try:
+ options.attach_file = open(options.attach_file, 'rb')
+ except IOError as e:
+ parser.error("Cannot open %s (%s)" % (options.attach_file, e.strerror))
+
+ return options
+
+
+def set_status_cb(option, opt_str, value, parser, status_name):
+ if getattr(parser.values, "action", None) is not None:
+ raise OptionValueError("argument %s: Only one status may be specified at once." % opt_str)
+
+ if len(parser.rargs) == 0:
+ raise OptionValueError("argument %s: must specify a single TEST_ID." % opt_str)
+ parser.values.action = status_name
+ parser.values.test_id = parser.rargs.pop(0)
+
+
+def generate_stream_results(args, output_writer):
+ output_writer.startTestRun()
+
+ if args.attach_file:
+ reader = partial(args.attach_file.read, _CHUNK_SIZE)
+ this_file_hunk = reader()
+ next_file_hunk = reader()
+
+ is_first_packet = True
+ is_last_packet = False
+ while not is_last_packet:
+ write_status = output_writer.status
+
+ if is_first_packet:
+ if args.attach_file:
+ if args.mimetype:
+ write_status = partial(write_status, mime_type=args.mimetype)
+ if args.tags:
+ write_status = partial(write_status, test_tags=set(args.tags))
+ write_status = partial(write_status, timestamp=create_timestamp())
+ if args.action not in _FINAL_ACTIONS:
+ write_status = partial(write_status, test_status=args.action)
+ is_first_packet = False
+
+ if args.attach_file:
+ filename = args.file_name or args.attach_file.name
+ write_status = partial(write_status, file_name=filename, file_bytes=this_file_hunk)
+ if next_file_hunk == b'':
+ write_status = partial(write_status, eof=True)
+ is_last_packet = True
+ else:
+ this_file_hunk = next_file_hunk
+ next_file_hunk = reader()
+ else:
+ is_last_packet = True
+
+ if args.test_id:
+ write_status = partial(write_status, test_id=args.test_id)
+
+ if is_last_packet:
+ if args.action in _FINAL_ACTIONS:
+ write_status = partial(write_status, test_status=args.action)
+
+ write_status()
+
+ output_writer.stopTestRun()
+
+
+def create_timestamp():
+ return datetime.datetime.now(UTC)
diff --git a/lib/subunit/python/subunit/filters.py b/lib/subunit/python/subunit/filters.py
index dc3fd8a..0a0a185 100644
--- a/lib/subunit/python/subunit/filters.py
+++ b/lib/subunit/python/subunit/filters.py
@@ -17,7 +17,14 @@
from optparse import OptionParser
import sys
-from subunit import DiscardStream, ProtocolTestCase
+from extras import safe_hasattr
+from testtools import CopyStreamResult, StreamResult, StreamResultRouter
+
+from subunit import (
+ DiscardStream, ProtocolTestCase, ByteStreamToStreamResult,
+ StreamResultToBytes,
+ )
+from subunit.test_results import CatFiles
def make_options(description):
@@ -31,33 +38,76 @@ def make_options(description):
help="Send the output to this path rather than stdout.")
parser.add_option(
"-f", "--forward", action="store_true", default=False,
- help="Forward subunit stream on stdout.")
+ help="Forward subunit stream on stdout. When set, received "
+ "non-subunit output will be encapsulated in subunit.")
return parser
def run_tests_from_stream(input_stream, result, passthrough_stream=None,
- forward_stream=None):
+ forward_stream=None, protocol_version=1, passthrough_subunit=True):
"""Run tests from a subunit input stream through 'result'.
+ Non-test events - top level file attachments - are expected to be
+ dropped by v2 StreamResults at the present time (as all the analysis code
+ is in ExtendedTestResult API's), so to implement passthrough_stream they
+ are diverted and copied directly when that is set.
+
:param input_stream: A stream containing subunit input.
:param result: A TestResult that will receive the test events.
+ NB: This should be an ExtendedTestResult for v1 and a StreamResult for
+ v2.
:param passthrough_stream: All non-subunit input received will be
sent to this stream. If not provided, uses the ``TestProtocolServer``
default, which is ``sys.stdout``.
:param forward_stream: All subunit input received will be forwarded
- to this stream. If not provided, uses the ``TestProtocolServer``
- default, which is to not forward any input.
+ to this stream. If not provided, uses the ``TestProtocolServer``
+ default, which is to not forward any input. Do not set this when
+ transforming the stream - items would be double-reported.
+ :param protocol_version: What version of the subunit protocol to expect.
+ :param passthrough_subunit: If True, passthrough should be as subunit
+ otherwise unwrap it. Only has effect when forward_stream is None.
+ (when forwarding as subunit non-subunit input is always turned into
+ subunit)
"""
- test = ProtocolTestCase(
- input_stream, passthrough=passthrough_stream,
- forward=forward_stream)
+ if 1==protocol_version:
+ test = ProtocolTestCase(
+ input_stream, passthrough=passthrough_stream,
+ forward=forward_stream)
+ elif 2==protocol_version:
+ # In all cases we encapsulate unknown inputs.
+ if forward_stream is not None:
+ # Send events to forward_stream as subunit.
+ forward_result = StreamResultToBytes(forward_stream)
+ # If we're passing non-subunit through, copy:
+ if passthrough_stream is None:
+ # Not passing non-test events - split them off to nothing.
+ router = StreamResultRouter(forward_result)
+ router.add_rule(StreamResult(), 'test_id', test_id=None)
+ result = CopyStreamResult([router, result])
+ else:
+ # otherwise, copy all events to forward_result
+ result = CopyStreamResult([forward_result, result])
+ elif passthrough_stream is not None:
+ if not passthrough_subunit:
+ # Route non-test events to passthrough_stream, unwrapping them for
+ # display.
+ passthrough_result = CatFiles(passthrough_stream)
+ else:
+ passthrough_result = StreamResultToBytes(passthrough_stream)
+ result = StreamResultRouter(result)
+ result.add_rule(passthrough_result, 'test_id', test_id=None)
+ test = ByteStreamToStreamResult(input_stream,
+ non_subunit_name='stdout')
+ else:
+ raise Exception("Unknown protocol version.")
result.startTestRun()
test.run(result)
result.stopTestRun()
def filter_by_result(result_factory, output_path, passthrough, forward,
- input_stream=sys.stdin):
+ input_stream=sys.stdin, protocol_version=1,
+ passthrough_subunit=True):
"""Filter an input stream using a test result.
:param result_factory: A callable that when passed an output stream
@@ -71,17 +121,24 @@ def filter_by_result(result_factory, output_path, passthrough, forward,
``sys.stdout`` as well as to the ``TestResult``.
:param input_stream: The source of subunit input. Defaults to
``sys.stdin``.
- :return: A test result with the resultts of the run.
+ :param protocol_version: The subunit protocol version to expect.
+ :param passthrough_subunit: If True, passthrough should be as subunit.
+ :return: A test result with the results of the run.
"""
if passthrough:
passthrough_stream = sys.stdout
else:
- passthrough_stream = DiscardStream()
+ if 1==protocol_version:
+ passthrough_stream = DiscardStream()
+ else:
+ passthrough_stream = None
if forward:
forward_stream = sys.stdout
- else:
+ elif 1==protocol_version:
forward_stream = DiscardStream()
+ else:
+ forward_stream = None
if output_path is None:
output_to = sys.stdout
@@ -91,14 +148,17 @@ def filter_by_result(result_factory, output_path, passthrough, forward,
try:
result = result_factory(output_to)
run_tests_from_stream(
- input_stream, result, passthrough_stream, forward_stream)
+ input_stream, result, passthrough_stream, forward_stream,
+ protocol_version=protocol_version,
+ passthrough_subunit=passthrough_subunit)
finally:
if output_path:
output_to.close()
return result
-def run_filter_script(result_factory, description, post_run_hook=None):
+def run_filter_script(result_factory, description, post_run_hook=None,
+ protocol_version=1, passthrough_subunit=True):
"""Main function for simple subunit filter scripts.
Many subunit filter scripts take a stream of subunit input and use a
@@ -111,15 +171,36 @@ def run_filter_script(result_factory, description, post_run_hook=None):
:param result_factory: A callable that takes an output stream and returns
a test result that outputs to that stream.
:param description: A description of the filter script.
+ :param protocol_version: What protocol version to consume/emit.
+ :param passthrough_subunit: If True, passthrough should be as subunit.
"""
parser = make_options(description)
(options, args) = parser.parse_args()
result = filter_by_result(
result_factory, options.output_to, not options.no_passthrough,
- options.forward)
+ options.forward, protocol_version=protocol_version,
+ passthrough_subunit=passthrough_subunit,
+ input_stream=find_stream(sys.stdin, args))
if post_run_hook:
post_run_hook(result)
+ if not safe_hasattr(result, 'wasSuccessful'):
+ result = result.decorated
if result.wasSuccessful():
sys.exit(0)
else:
sys.exit(1)
+
+
+def find_stream(stdin, argv):
+ """Find a stream to use as input for filters.
+
+ :param stdin: Standard in - used if no files are named in argv.
+ :param argv: Command line arguments after option parsing. If one file
+ is named, that is opened in read only binary mode and returned.
+ A missing file will raise an exception, as will multiple file names.
+ """
+ assert len(argv) < 2, "Too many filenames."
+ if argv:
+ return open(argv[0], 'rb')
+ else:
+ return stdin
diff --git a/lib/subunit/python/subunit/run.py b/lib/subunit/python/subunit/run.py
index b5ccea4..cf9cc01 100755
--- a/lib/subunit/python/subunit/run.py
+++ b/lib/subunit/python/subunit/run.py
@@ -20,39 +20,80 @@
$ python -m subunit.run mylib.tests.test_suite
"""
+import io
+import os
import sys
-from subunit import TestProtocolClient, get_default_formatter
+from testtools import ExtendedToStreamDecorator
+from testtools.testsuite import iterate_tests
+
+from subunit import StreamResultToBytes, get_default_formatter
from subunit.test_results import AutoTimingTestResultDecorator
from testtools.run import (
BUFFEROUTPUT,
CATCHBREAK,
FAILFAST,
+ list_test,
TestProgram,
USAGE_AS_MAIN,
)
class SubunitTestRunner(object):
- def __init__(self, verbosity=None, failfast=None, buffer=None, stream=None):
+ def __init__(self, verbosity=None, failfast=None, buffer=None, stream=None,
+ stdout=None):
"""Create a TestToolsTestRunner.
:param verbosity: Ignored.
:param failfast: Stop running tests at the first failure.
:param buffer: Ignored.
+ :param stream: Upstream unittest stream parameter.
+ :param stdout: Testtools stream parameter.
+
+ Either stream or stdout can be supplied, and stream will take
+ precedence.
"""
self.failfast = failfast
- self.stream = stream or sys.stdout
+ self.stream = stream or stdout or sys.stdout
def run(self, test):
"Run the given test case or test suite."
- result = TestProtocolClient(self.stream)
+ result, _ = self._list(test)
+ result = ExtendedToStreamDecorator(result)
result = AutoTimingTestResultDecorator(result)
if self.failfast is not None:
result.failfast = self.failfast
- test(result)
+ result.startTestRun()
+ try:
+ test(result)
+ finally:
+ result.stopTestRun()
return result
+ def list(self, test):
+ "List the test."
+ result, errors = self._list(test)
+ if errors:
+ failed_descr = '\n'.join(errors).encode('utf8')
+ result.status(file_name="import errors", runnable=False,
+ file_bytes=failed_descr, mime_type="text/plain;charset=utf8")
+ sys.exit(2)
+
+ def _list(self, test):
+ test_ids, errors = list_test(test)
+ try:
+ fileno = self.stream.fileno()
+ except:
+ fileno = None
+ if fileno is not None:
+ stream = os.fdopen(fileno, 'wb', 0)
+ else:
+ stream = self.stream
+ result = StreamResultToBytes(stream)
+ for test_id in test_ids:
+ result.status(test_id=test_id, test_status='exists')
+ return result, errors
+
class SubunitTestProgram(TestProgram):
@@ -77,8 +118,28 @@ class SubunitTestProgram(TestProgram):
sys.exit(2)
-if __name__ == '__main__':
- stream = get_default_formatter()
+def main(argv=None, stdout=None):
+ if argv is None:
+ argv = sys.argv
runner = SubunitTestRunner
- SubunitTestProgram(module=None, argv=sys.argv, testRunner=runner,
- stdout=sys.stdout)
+ # stdout is None except in unit tests.
+ if stdout is None:
+ stdout = sys.stdout
+ # XXX: This is broken code- SUBUNIT_FORMATTER is not being honoured.
+ stream = get_default_formatter()
+ # Disable the default buffering, for Python 2.x where pdb doesn't do it
+ # on non-ttys.
+ if hasattr(stdout, 'fileno'):
+ # Patch stdout to be unbuffered, so that pdb works well on 2.6/2.7.
+ binstdout = io.open(stdout.fileno(), 'wb', 0)
+ if sys.version_info[0] > 2:
+ sys.stdout = io.TextIOWrapper(binstdout, encoding=sys.stdout.encoding)
+ else:
+ sys.stdout = binstdout
+ stdout = sys.stdout
+ SubunitTestProgram(module=None, argv=argv, testRunner=runner,
+ stdout=stdout, exit=False)
+
+
+if __name__ == '__main__':
+ main()
diff --git a/lib/subunit/python/subunit/test_results.py b/lib/subunit/python/subunit/test_results.py
index 91c9bbd..b3ca968 100644
--- a/lib/subunit/python/subunit/test_results.py
+++ b/lib/subunit/python/subunit/test_results.py
@@ -20,13 +20,14 @@ import csv
import datetime
import testtools
-from testtools.compat import all
from testtools.content import (
text_content,
TracebackContent,
)
+from testtools import StreamResult
from subunit import iso8601
+import subunit
# NOT a TestResult, because we are implementing the interface, not inheriting
@@ -525,16 +526,24 @@ class TestResultFilter(TestResultDecorator):
class TestIdPrintingResult(testtools.TestResult):
+ """Print test ids to a stream.
- def __init__(self, stream, show_times=False):
+ Implements both TestResult and StreamResult, for compatibility.
+ """
+
+ def __init__(self, stream, show_times=False, show_exists=False):
"""Create a FilterResult object outputting to stream."""
super(TestIdPrintingResult, self).__init__()
self._stream = stream
+ self.show_exists = show_exists
+ self.show_times = show_times
+
+ def startTestRun(self):
self.failed_tests = 0
self.__time = None
- self.show_times = show_times
self._test = None
self._test_duration = 0
+ self._active_tests = {}
def addError(self, test, err):
self.failed_tests += 1
@@ -557,21 +566,44 @@ class TestIdPrintingResult(testtools.TestResult):
def addExpectedFailure(self, test, err=None, details=None):
self._test = test
- def reportTest(self, test, duration):
+ def reportTest(self, test_id, duration):
if self.show_times:
seconds = duration.seconds
seconds += duration.days * 3600 * 24
seconds += duration.microseconds / 1000000.0
- self._stream.write(test.id() + ' %0.3f\n' % seconds)
+ self._stream.write(test_id + ' %0.3f\n' % seconds)
else:
- self._stream.write(test.id() + '\n')
+ self._stream.write(test_id + '\n')
def startTest(self, test):
self._start_time = self._time()
+ def status(self, test_id=None, test_status=None, test_tags=None,
+ runnable=True, file_name=None, file_bytes=None, eof=False,
+ mime_type=None, route_code=None, timestamp=None):
+ if not test_id:
+ return
+ if timestamp is not None:
+ self.time(timestamp)
+ if test_status=='exists':
+ if self.show_exists:
+ self.reportTest(test_id, 0)
+ elif test_status in ('inprogress', None):
+ self._active_tests[test_id] = self._time()
+ else:
+ self._end_test(test_id)
+
+ def _end_test(self, test_id):
+ test_start = self._active_tests.pop(test_id, None)
+ if not test_start:
+ test_duration = 0
+ else:
+ test_duration = self._time() - test_start
+ self.reportTest(test_id, test_duration)
+
def stopTest(self, test):
test_duration = self._time() - self._start_time
- self.reportTest(self._test, test_duration)
+ self.reportTest(self._test.id(), test_duration)
def time(self, time):
self.__time = time
@@ -583,6 +615,10 @@ class TestIdPrintingResult(testtools.TestResult):
"Tells whether or not this result was a success"
return self.failed_tests == 0
+ def stopTestRun(self):
+ for test_id in list(self._active_tests.keys()):
+ self._end_test(test_id)
+
class TestByTestResult(testtools.TestResult):
"""Call something every time a test completes."""
@@ -676,3 +712,17 @@ class CsvResult(TestByTestResult):
def startTestRun(self):
super(CsvResult, self).startTestRun()
self._write_row(['test', 'status', 'start_time', 'stop_time'])
+
+
+class CatFiles(StreamResult):
+ """Cat file attachments received to a stream."""
+
+ def __init__(self, byte_stream):
+ self.stream = subunit.make_stream_binary(byte_stream)
+
+ def status(self, test_id=None, test_status=None, test_tags=None,
+ runnable=True, file_name=None, file_bytes=None, eof=False,
+ mime_type=None, route_code=None, timestamp=None):
+ if file_name is not None:
+ self.stream.write(file_bytes)
+ self.stream.flush()
diff --git a/lib/subunit/python/subunit/tests/TestUtil.py b/lib/subunit/python/subunit/tests/TestUtil.py
deleted file mode 100644
index 39d901e..0000000
--- a/lib/subunit/python/subunit/tests/TestUtil.py
+++ /dev/null
@@ -1,80 +0,0 @@
-# Copyright (c) 2004 Canonical Limited
-# Author: Robert Collins <robert.collins at 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
index e0e1eb1..c1c2c64 100644
--- a/lib/subunit/python/subunit/tests/__init__.py
+++ b/lib/subunit/python/subunit/tests/__init__.py
@@ -6,7 +6,7 @@
# 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
@@ -14,10 +14,29 @@
# limitations under that license.
#
+import sys
+from unittest import TestLoader
+
+from testscenarios import generate_scenarios
+
+
+# Before the test module imports to avoid circularity.
+# For testing: different pythons have different str() implementations.
+if sys.version_info > (3, 0):
+ _remote_exception_repr = "testtools.testresult.real._StringException"
+ _remote_exception_str = "Traceback (most recent call last):\ntesttools.testresult.real._StringException"
+ _remote_exception_str_chunked = "57\r\n" + _remote_exception_str + ": boo qux\n0\r\n"
+else:
+ _remote_exception_repr = "_StringException"
+ _remote_exception_str = "Traceback (most recent call last):\n_StringException"
+ _remote_exception_str_chunked = "3D\r\n" + _remote_exception_str + ": boo qux\n0\r\n"
+
+
from subunit.tests import (
- TestUtil,
test_chunked,
test_details,
+ test_filters,
+ test_output_filter,
test_progress_model,
test_run,
test_subunit_filter,
@@ -25,19 +44,26 @@ from subunit.tests import (
test_subunit_tags,
test_tap2subunit,
test_test_protocol,
+ test_test_protocol2,
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())
+ loader = TestLoader()
+ result = loader.loadTestsFromModule(test_chunked)
+ result.addTest(loader.loadTestsFromModule(test_details))
+ result.addTest(loader.loadTestsFromModule(test_filters))
+ result.addTest(loader.loadTestsFromModule(test_progress_model))
+ result.addTest(loader.loadTestsFromModule(test_test_results))
+ result.addTest(loader.loadTestsFromModule(test_test_protocol))
+ result.addTest(loader.loadTestsFromModule(test_test_protocol2))
+ result.addTest(loader.loadTestsFromModule(test_tap2subunit))
+ result.addTest(loader.loadTestsFromModule(test_subunit_filter))
+ result.addTest(loader.loadTestsFromModule(test_subunit_tags))
+ result.addTest(loader.loadTestsFromModule(test_subunit_stats))
+ result.addTest(loader.loadTestsFromModule(test_run))
+ result.addTests(
+ generate_scenarios(loader.loadTestsFromModule(test_output_filter))
+ )
return result
diff --git a/lib/subunit/python/subunit/tests/test_chunked.py b/lib/subunit/python/subunit/tests/test_chunked.py
index e0742f1..5100b32 100644
--- a/lib/subunit/python/subunit/tests/test_chunked.py
+++ b/lib/subunit/python/subunit/tests/test_chunked.py
@@ -22,12 +22,6 @@ 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):
diff --git a/lib/subunit/python/subunit/tests/test_details.py b/lib/subunit/python/subunit/tests/test_details.py
index 746aa04..8605c5a 100644
--- a/lib/subunit/python/subunit/tests/test_details.py
+++ b/lib/subunit/python/subunit/tests/test_details.py
@@ -22,12 +22,6 @@ 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):
diff --git a/lib/subunit/python/subunit/tests/test_filters.py b/lib/subunit/python/subunit/tests/test_filters.py
new file mode 100644
index 0000000..0a5e7c7
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_filters.py
@@ -0,0 +1,35 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Robert Collins <robertc at 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 sys
+from tempfile import NamedTemporaryFile
+
+from testtools import TestCase
+
+from subunit.filters import find_stream
+
+
+class TestFindStream(TestCase):
+
+ def test_no_argv(self):
+ self.assertEqual('foo', find_stream('foo', []))
+
+ def test_opens_file(self):
+ f = NamedTemporaryFile()
+ f.write(b'foo')
+ f.flush()
+ stream = find_stream('bar', [f.name])
+ self.assertEqual(b'foo', stream.read())
diff --git a/lib/subunit/python/subunit/tests/test_output_filter.py b/lib/subunit/python/subunit/tests/test_output_filter.py
new file mode 100644
index 0000000..0f61ac5
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_output_filter.py
@@ -0,0 +1,596 @@
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Subunit Contributors
+#
+# 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
+from functools import partial
+from io import BytesIO, StringIO, TextIOWrapper
+import optparse
+import sys
+from tempfile import NamedTemporaryFile
+
+from contextlib import contextmanager
+from testtools import TestCase
+from testtools.compat import _u
+from testtools.matchers import (
+ Equals,
+ Matcher,
+ MatchesAny,
+ MatchesListwise,
+ Mismatch,
+ raises,
+)
+from testtools.testresult.doubles import StreamResult
+
+from subunit.iso8601 import UTC
+from subunit.v2 import StreamResultToBytes, ByteStreamToStreamResult
+from subunit._output import (
+ _ALL_ACTIONS,
+ _FINAL_ACTIONS,
+ generate_stream_results,
+ parse_arguments,
+)
+import subunit._output as _o
+
+
+class SafeOptionParser(optparse.OptionParser):
+ """An ArgumentParser class that doesn't call sys.exit."""
+
+ def exit(self, status=0, message=""):
+ raise RuntimeError(message)
+
+ def error(self, message):
+ raise RuntimeError(message)
+
+
+safe_parse_arguments = partial(parse_arguments, ParserClass=SafeOptionParser)
+
+
+class TestStatusArgParserTests(TestCase):
+
+ scenarios = [
+ (cmd, dict(command=cmd, option='--' + cmd)) for cmd in _ALL_ACTIONS
+ ]
+
+ def test_can_parse_all_commands_with_test_id(self):
+ test_id = self.getUniqueString()
+ args = safe_parse_arguments(args=[self.option, test_id])
+
+ self.assertThat(args.action, Equals(self.command))
+ self.assertThat(args.test_id, Equals(test_id))
+
+ def test_all_commands_parse_file_attachment(self):
+ with NamedTemporaryFile() as tmp_file:
+ args = safe_parse_arguments(
+ args=[self.option, 'foo', '--attach-file', tmp_file.name]
+ )
+ self.assertThat(args.attach_file.name, Equals(tmp_file.name))
+
+ def test_all_commands_accept_mimetype_argument(self):
+ with NamedTemporaryFile() as tmp_file:
+ args = safe_parse_arguments(
+ args=[self.option, 'foo', '--attach-file', tmp_file.name, '--mimetype', "text/plain"]
+ )
+ self.assertThat(args.mimetype, Equals("text/plain"))
+
+ def test_all_commands_accept_file_name_argument(self):
+ with NamedTemporaryFile() as tmp_file:
+ args = safe_parse_arguments(
+ args=[self.option, 'foo', '--attach-file', tmp_file.name, '--file-name', "foo"]
+ )
+ self.assertThat(args.file_name, Equals("foo"))
+
+ def test_all_commands_accept_tags_argument(self):
+ args = safe_parse_arguments(
+ args=[self.option, 'foo', '--tag', "foo", "--tag", "bar", "--tag", "baz"]
+ )
+ self.assertThat(args.tags, Equals(["foo", "bar", "baz"]))
+
+ def test_attach_file_with_hyphen_opens_stdin(self):
+ self.patch(_o.sys, 'stdin', TextIOWrapper(BytesIO(b"Hello")))
+ args = safe_parse_arguments(
+ args=[self.option, "foo", "--attach-file", "-"]
+ )
+
+ self.assertThat(args.attach_file.read(), Equals(b"Hello"))
+
+ def test_attach_file_with_hyphen_sets_filename_to_stdin(self):
+ args = safe_parse_arguments(
+ args=[self.option, "foo", "--attach-file", "-"]
+ )
+
+ self.assertThat(args.file_name, Equals("stdin"))
+
+ def test_can_override_stdin_filename(self):
+ args = safe_parse_arguments(
+ args=[self.option, "foo", "--attach-file", "-", '--file-name', 'foo']
+ )
+
+ self.assertThat(args.file_name, Equals("foo"))
+
+ def test_requires_test_id(self):
+ fn = lambda: safe_parse_arguments(args=[self.option])
+ self.assertThat(
+ fn,
+ raises(RuntimeError('argument %s: must specify a single TEST_ID.' % self.option))
+ )
+
+
+class ArgParserTests(TestCase):
+
+ def test_can_parse_attach_file_without_test_id(self):
+ with NamedTemporaryFile() as tmp_file:
+ args = safe_parse_arguments(
+ args=["--attach-file", tmp_file.name]
+ )
+ self.assertThat(args.attach_file.name, Equals(tmp_file.name))
+
+ def test_can_run_without_args(self):
+ args = safe_parse_arguments([])
+
+ def test_cannot_specify_more_than_one_status_command(self):
+ fn = lambda: safe_parse_arguments(['--fail', 'foo', '--skip', 'bar'])
+ self.assertThat(
+ fn,
+ raises(RuntimeError('argument --skip: Only one status may be specified at once.'))
+ )
+
+ def test_cannot_specify_mimetype_without_attach_file(self):
+ fn = lambda: safe_parse_arguments(['--mimetype', 'foo'])
+ self.assertThat(
+ fn,
+ raises(RuntimeError('Cannot specify --mimetype without --attach-file'))
+ )
+
+ def test_cannot_specify_filename_without_attach_file(self):
+ fn = lambda: safe_parse_arguments(['--file-name', 'foo'])
+ self.assertThat(
+ fn,
+ raises(RuntimeError('Cannot specify --file-name without --attach-file'))
+ )
+
+ def test_can_specify_tags_without_status_command(self):
+ args = safe_parse_arguments(['--tag', 'foo'])
+ self.assertEqual(['foo'], args.tags)
+
+ def test_must_specify_tags_with_tags_options(self):
+ fn = lambda: safe_parse_arguments(['--fail', 'foo', '--tag'])
+ self.assertThat(
+ fn,
+ MatchesAny(
+ raises(RuntimeError('--tag option requires 1 argument')),
+ raises(RuntimeError('--tag option requires an argument')),
+ )
+ )
+
+def get_result_for(commands):
+ """Get a result object from *commands.
+
+ Runs the 'generate_stream_results' function from subunit._output after
+ parsing *commands as if they were specified on the command line. The
+ resulting bytestream is then converted back into a result object and
+ returned.
+ """
+ result = StreamResult()
+ args = safe_parse_arguments(commands)
+ generate_stream_results(args, result)
+ return result
+
+
+ at contextmanager
+def temp_file_contents(data):
+ """Create a temporary file on disk containing 'data'."""
+ with NamedTemporaryFile() as f:
+ f.write(data)
+ f.seek(0)
+ yield f
+
+
+class StatusStreamResultTests(TestCase):
+
+ scenarios = [
+ (s, dict(status=s, option='--' + s)) for s in _ALL_ACTIONS
+ ]
+
+ _dummy_timestamp = datetime.datetime(2013, 1, 1, 0, 0, 0, 0, UTC)
+
+ def setUp(self):
+ super(StatusStreamResultTests, self).setUp()
+ self.patch(_o, 'create_timestamp', lambda: self._dummy_timestamp)
+ self.test_id = self.getUniqueString()
+
+ def test_only_one_packet_is_generated(self):
+ result = get_result_for([self.option, self.test_id])
+ self.assertThat(
+ len(result._events),
+ Equals(3) # startTestRun and stopTestRun are also called, making 3 total.
+ )
+
+ def test_correct_status_is_generated(self):
+ result = get_result_for([self.option, self.test_id])
+
+ self.assertThat(
+ result._events[1],
+ MatchesStatusCall(test_status=self.status)
+ )
+
+ def test_all_commands_generate_tags(self):
+ result = get_result_for([self.option, self.test_id, '--tag', 'hello', '--tag', 'world'])
+ self.assertThat(
+ result._events[1],
+ MatchesStatusCall(test_tags=set(['hello', 'world']))
+ )
+
+ def test_all_commands_generate_timestamp(self):
+ result = get_result_for([self.option, self.test_id])
+
+ self.assertThat(
+ result._events[1],
+ MatchesStatusCall(timestamp=self._dummy_timestamp)
+ )
+
+ def test_all_commands_generate_correct_test_id(self):
+ result = get_result_for([self.option, self.test_id])
+
+ self.assertThat(
+ result._events[1],
+ MatchesStatusCall(test_id=self.test_id)
+ )
+
+ def test_file_is_sent_in_single_packet(self):
+ with temp_file_contents(b"Hello") as f:
+ result = get_result_for([self.option, self.test_id, '--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_bytes=b'Hello', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_can_read_binary_files(self):
+ with temp_file_contents(b"\xDE\xAD\xBE\xEF") as f:
+ result = get_result_for([self.option, self.test_id, '--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_bytes=b"\xDE\xAD\xBE\xEF", eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_can_read_empty_files(self):
+ with temp_file_contents(b"") as f:
+ result = get_result_for([self.option, self.test_id, '--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_bytes=b"", file_name=f.name, eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_can_read_stdin(self):
+ self.patch(_o.sys, 'stdin', TextIOWrapper(BytesIO(b"\xFE\xED\xFA\xCE")))
+ result = get_result_for([self.option, self.test_id, '--attach-file', '-'])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_bytes=b"\xFE\xED\xFA\xCE", file_name='stdin', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_file_is_sent_with_test_id(self):
+ with temp_file_contents(b"Hello") as f:
+ result = get_result_for([self.option, self.test_id, '--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(test_id=self.test_id, file_bytes=b'Hello', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_file_is_sent_with_test_status(self):
+ with temp_file_contents(b"Hello") as f:
+ result = get_result_for([self.option, self.test_id, '--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(test_status=self.status, file_bytes=b'Hello', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_file_chunk_size_is_honored(self):
+ with temp_file_contents(b"Hello") as f:
+ self.patch(_o, '_CHUNK_SIZE', 1)
+ result = get_result_for([self.option, self.test_id, '--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(test_id=self.test_id, file_bytes=b'H', eof=False),
+ MatchesStatusCall(test_id=self.test_id, file_bytes=b'e', eof=False),
+ MatchesStatusCall(test_id=self.test_id, file_bytes=b'l', eof=False),
+ MatchesStatusCall(test_id=self.test_id, file_bytes=b'l', eof=False),
+ MatchesStatusCall(test_id=self.test_id, file_bytes=b'o', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_file_mimetype_specified_once_only(self):
+ with temp_file_contents(b"Hi") as f:
+ self.patch(_o, '_CHUNK_SIZE', 1)
+ result = get_result_for([
+ self.option,
+ self.test_id,
+ '--attach-file',
+ f.name,
+ '--mimetype',
+ 'text/plain',
+ ])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(test_id=self.test_id, mime_type='text/plain', file_bytes=b'H', eof=False),
+ MatchesStatusCall(test_id=self.test_id, mime_type=None, file_bytes=b'i', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_tags_specified_once_only(self):
+ with temp_file_contents(b"Hi") as f:
+ self.patch(_o, '_CHUNK_SIZE', 1)
+ result = get_result_for([
+ self.option,
+ self.test_id,
+ '--attach-file',
+ f.name,
+ '--tag',
+ 'foo',
+ '--tag',
+ 'bar',
+ ])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(test_id=self.test_id, test_tags=set(['foo', 'bar'])),
+ MatchesStatusCall(test_id=self.test_id, test_tags=None),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_timestamp_specified_once_only(self):
+ with temp_file_contents(b"Hi") as f:
+ self.patch(_o, '_CHUNK_SIZE', 1)
+ result = get_result_for([
+ self.option,
+ self.test_id,
+ '--attach-file',
+ f.name,
+ ])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(test_id=self.test_id, timestamp=self._dummy_timestamp),
+ MatchesStatusCall(test_id=self.test_id, timestamp=None),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_test_status_specified_once_only(self):
+ with temp_file_contents(b"Hi") as f:
+ self.patch(_o, '_CHUNK_SIZE', 1)
+ result = get_result_for([
+ self.option,
+ self.test_id,
+ '--attach-file',
+ f.name,
+ ])
+
+ # 'inprogress' status should be on the first packet only, all other
+ # statuses should be on the last packet.
+ if self.status in _FINAL_ACTIONS:
+ first_call = MatchesStatusCall(test_id=self.test_id, test_status=None)
+ last_call = MatchesStatusCall(test_id=self.test_id, test_status=self.status)
+ else:
+ first_call = MatchesStatusCall(test_id=self.test_id, test_status=self.status)
+ last_call = MatchesStatusCall(test_id=self.test_id, test_status=None)
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ first_call,
+ last_call,
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_filename_can_be_overridden(self):
+ with temp_file_contents(b"Hello") as f:
+ specified_file_name = self.getUniqueString()
+ result = get_result_for([
+ self.option,
+ self.test_id,
+ '--attach-file',
+ f.name,
+ '--file-name',
+ specified_file_name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_name=specified_file_name, file_bytes=b'Hello'),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_file_name_is_used_by_default(self):
+ with temp_file_contents(b"Hello") as f:
+ result = get_result_for([self.option, self.test_id, '--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_name=f.name, file_bytes=b'Hello', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+
+class FileDataTests(TestCase):
+
+ def test_can_attach_file_without_test_id(self):
+ with temp_file_contents(b"Hello") as f:
+ result = get_result_for(['--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(test_id=None, file_bytes=b'Hello', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_file_name_is_used_by_default(self):
+ with temp_file_contents(b"Hello") as f:
+ result = get_result_for(['--attach-file', f.name])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_name=f.name, file_bytes=b'Hello', eof=True),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_filename_can_be_overridden(self):
+ with temp_file_contents(b"Hello") as f:
+ specified_file_name = self.getUniqueString()
+ result = get_result_for([
+ '--attach-file',
+ f.name,
+ '--file-name',
+ specified_file_name
+ ])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_name=specified_file_name, file_bytes=b'Hello'),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_files_have_timestamp(self):
+ _dummy_timestamp = datetime.datetime(2013, 1, 1, 0, 0, 0, 0, UTC)
+ self.patch(_o, 'create_timestamp', lambda: _dummy_timestamp)
+
+ with temp_file_contents(b"Hello") as f:
+ specified_file_name = self.getUniqueString()
+ result = get_result_for([
+ '--attach-file',
+ f.name,
+ ])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(file_bytes=b'Hello', timestamp=_dummy_timestamp),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+ def test_can_specify_tags_without_test_status(self):
+ result = get_result_for([
+ '--tag',
+ 'foo',
+ ])
+
+ self.assertThat(
+ result._events,
+ MatchesListwise([
+ MatchesStatusCall(call='startTestRun'),
+ MatchesStatusCall(test_tags=set(['foo'])),
+ MatchesStatusCall(call='stopTestRun'),
+ ])
+ )
+
+
+class MatchesStatusCall(Matcher):
+
+ _position_lookup = {
+ 'call': 0,
+ 'test_id': 1,
+ 'test_status': 2,
+ 'test_tags': 3,
+ 'runnable': 4,
+ 'file_name': 5,
+ 'file_bytes': 6,
+ 'eof': 7,
+ 'mime_type': 8,
+ 'route_code': 9,
+ 'timestamp': 10,
+ }
+
+ def __init__(self, **kwargs):
+ unknown_kwargs = list(filter(
+ lambda k: k not in self._position_lookup,
+ kwargs
+ ))
+ if unknown_kwargs:
+ raise ValueError("Unknown keywords: %s" % ','.join(unknown_kwargs))
+ self._filters = kwargs
+
+ def match(self, call_tuple):
+ for k, v in self._filters.items():
+ try:
+ pos = self._position_lookup[k]
+ if call_tuple[pos] != v:
+ return Mismatch(
+ "Value for key is %r, not %r" % (call_tuple[pos], v)
+ )
+ except IndexError:
+ return Mismatch("Key %s is not present." % k)
+
+ def __str__(self):
+ return "<MatchesStatusCall %r>" % self._filters
diff --git a/lib/subunit/python/subunit/tests/test_progress_model.py b/lib/subunit/python/subunit/tests/test_progress_model.py
index 76200c6..2ca0888 100644
--- a/lib/subunit/python/subunit/tests/test_progress_model.py
+++ b/lib/subunit/python/subunit/tests/test_progress_model.py
@@ -110,9 +110,3 @@ class TestProgressModel(unittest.TestCase):
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
index 10519ed..d92ed04 100644
--- a/lib/subunit/python/subunit/tests/test_run.py
+++ b/lib/subunit/python/subunit/tests/test_run.py
@@ -14,39 +14,75 @@
# limitations under that license.
#
-from testtools.compat import BytesIO
+import io
import unittest
-from testtools import PlaceHolder
+from testtools import PlaceHolder, TestCase
+from testtools.compat import _b
+from testtools.matchers import StartsWith
+from testtools.testresult.doubles import StreamResult
import subunit
+from subunit import run
from subunit.run import SubunitTestRunner
-def test_suite():
- loader = subunit.tests.TestUtil.TestLoader()
- result = loader.loadTestsFromName(__name__)
- return result
+class TestSubunitTestRunner(TestCase):
+ def test_includes_timing_output(self):
+ bytestream = io.BytesIO()
+ runner = SubunitTestRunner(stream=bytestream)
+ test = PlaceHolder('name')
+ runner.run(test)
+ bytestream.seek(0)
+ eventstream = StreamResult()
+ subunit.ByteStreamToStreamResult(bytestream).run(eventstream)
+ timestamps = [event[-1] for event in eventstream._events
+ if event is not None]
+ self.assertNotEqual([], timestamps)
-class TimeCollectingTestResult(unittest.TestResult):
-
- def __init__(self, *args, **kwargs):
- super(TimeCollectingTestResult, self).__init__(*args, **kwargs)
- self.time_called = []
+ def test_enumerates_tests_before_run(self):
+ bytestream = io.BytesIO()
+ runner = SubunitTestRunner(stream=bytestream)
+ test1 = PlaceHolder('name1')
+ test2 = PlaceHolder('name2')
+ case = unittest.TestSuite([test1, test2])
+ runner.run(case)
+ bytestream.seek(0)
+ eventstream = StreamResult()
+ subunit.ByteStreamToStreamResult(bytestream).run(eventstream)
+ self.assertEqual([
+ ('status', 'name1', 'exists'),
+ ('status', 'name2', 'exists'),
+ ], [event[:3] for event in eventstream._events[:2]])
- def time(self, a_time):
- self.time_called.append(a_time)
+ def test_list_errors_if_errors_from_list_test(self):
+ bytestream = io.BytesIO()
+ runner = SubunitTestRunner(stream=bytestream)
+ def list_test(test):
+ return [], ['failed import']
+ self.patch(run, 'list_test', list_test)
+ exc = self.assertRaises(SystemExit, runner.list, None)
+ self.assertEqual((2,), exc.args)
+ class FailingTest(TestCase):
+ def test_fail(self):
+ 1/0
-class TestSubunitTestRunner(unittest.TestCase):
+ def test_exits_zero_when_tests_fail(self):
+ bytestream = io.BytesIO()
+ stream = io.TextIOWrapper(bytestream, encoding="utf8")
+ try:
+ self.assertEqual(None, run.main(
+ argv=["progName", "subunit.tests.test_run.TestSubunitTestRunner.FailingTest"],
+ stdout=stream))
+ except SystemExit:
+ self.fail("SystemExit raised")
+ self.assertThat(bytestream.getvalue(), StartsWith(_b('\xb3')))
- 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)
+ def test_exits_nonzero_when_execution_errors(self):
+ bytestream = io.BytesIO()
+ stream = io.TextIOWrapper(bytestream, encoding="utf8")
+ exc = self.assertRaises(Exception, run.main,
+ argv=["progName", "subunit.tests.test_run.TestSubunitTestRunner.MissingTest"],
+ stdout=stream)
diff --git a/lib/subunit/python/subunit/tests/test_subunit_filter.py b/lib/subunit/python/subunit/tests/test_subunit_filter.py
index 33b9248..5f34b3b 100644
--- a/lib/subunit/python/subunit/tests/test_subunit_filter.py
+++ b/lib/subunit/python/subunit/tests/test_subunit_filter.py
@@ -25,10 +25,11 @@ import unittest
from testtools import TestCase
from testtools.compat import _b, BytesIO
-from testtools.testresult.doubles import ExtendedTestResult
+from testtools.testresult.doubles import ExtendedTestResult, StreamResult
import subunit
from subunit.test_results import make_tag_filter, TestResultFilter
+from subunit import ByteStreamToStreamResult, StreamResultToBytes
class TestTestResultFilter(TestCase):
@@ -286,23 +287,6 @@ xfail todo
class TestFilterCommand(TestCase):
- example_subunit_stream = _b("""\
-tags: global
-test passed
-success passed
-test failed
-tags: local
-failure failed
-test error
-error error [
-error details
-]
-test skipped
-skip skipped
-test todo
-xfail todo
-""")
-
def run_command(self, args, stream):
root = os.path.dirname(
os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
@@ -316,55 +300,47 @@ xfail todo
raise RuntimeError("%s failed: %s" % (command, err))
return out
- def to_events(self, stream):
- test = subunit.ProtocolTestCase(BytesIO(stream))
- result = ExtendedTestResult()
- test.run(result)
- return result._events
-
def test_default(self):
- output = self.run_command([], _b(
- "test: foo\n"
- "skip: foo\n"
- ))
- events = self.to_events(output)
- foo = subunit.RemotedTestCase('foo')
- self.assertEqual(
- [('startTest', foo),
- ('addSkip', foo, {}),
- ('stopTest', foo)],
- events)
+ byte_stream = BytesIO()
+ stream = StreamResultToBytes(byte_stream)
+ stream.status(test_id="foo", test_status="inprogress")
+ stream.status(test_id="foo", test_status="skip")
+ output = self.run_command([], byte_stream.getvalue())
+ events = StreamResult()
+ ByteStreamToStreamResult(BytesIO(output)).run(events)
+ ids = set(event[1] for event in events._events)
+ self.assertEqual([
+ ('status', 'foo', 'inprogress'),
+ ('status', 'foo', 'skip'),
+ ], [event[:3] for event in events._events])
def test_tags(self):
- output = self.run_command(['-s', '--with-tag', 'a'], _b(
- "tags: a\n"
- "test: foo\n"
- "success: foo\n"
- "tags: -a\n"
- "test: bar\n"
- "success: bar\n"
- "test: baz\n"
- "tags: a\n"
- "success: baz\n"
- ))
- events = self.to_events(output)
- foo = subunit.RemotedTestCase('foo')
- baz = subunit.RemotedTestCase('baz')
- self.assertEqual(
- [('tags', set(['a']), set()),
- ('startTest', foo),
- ('addSuccess', foo),
- ('stopTest', foo),
- ('tags', set(), set(['a'])),
- ('startTest', baz),
- ('tags', set(['a']), set()),
- ('addSuccess', baz),
- ('stopTest', baz),
- ],
- events)
-
-
-def test_suite():
- loader = subunit.tests.TestUtil.TestLoader()
- result = loader.loadTestsFromName(__name__)
- return result
+ byte_stream = BytesIO()
+ stream = StreamResultToBytes(byte_stream)
+ stream.status(
+ test_id="foo", test_status="inprogress", test_tags=set(["a"]))
+ stream.status(
+ test_id="foo", test_status="success", test_tags=set(["a"]))
+ stream.status(test_id="bar", test_status="inprogress")
+ stream.status(test_id="bar", test_status="inprogress")
+ stream.status(
+ test_id="baz", test_status="inprogress", test_tags=set(["a"]))
+ stream.status(
+ test_id="baz", test_status="success", test_tags=set(["a"]))
+ output = self.run_command(
+ ['-s', '--with-tag', 'a'], byte_stream.getvalue())
+ events = StreamResult()
+ ByteStreamToStreamResult(BytesIO(output)).run(events)
+ ids = set(event[1] for event in events._events)
+ self.assertEqual(set(['foo', 'baz']), ids)
+
+ def test_no_passthrough(self):
+ output = self.run_command(['--no-passthrough'], b'hi thar')
+ self.assertEqual(b'', output)
+
+ def test_passthrough(self):
+ output = self.run_command([], b'hi thar')
+ byte_stream = BytesIO()
+ stream = StreamResultToBytes(byte_stream)
+ stream.status(file_name="stdout", file_bytes=b'hi thar')
+ self.assertEqual(byte_stream.getvalue(), output)
diff --git a/lib/subunit/python/subunit/tests/test_subunit_stats.py b/lib/subunit/python/subunit/tests/test_subunit_stats.py
index 6fd3301..7c5e42d 100644
--- a/lib/subunit/python/subunit/tests/test_subunit_stats.py
+++ b/lib/subunit/python/subunit/tests/test_subunit_stats.py
@@ -76,9 +76,3 @@ Seen tags: global, local
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
index c98506a..a16edc1 100644
--- a/lib/subunit/python/subunit/tests/test_subunit_tags.py
+++ b/lib/subunit/python/subunit/tests/test_subunit_tags.py
@@ -16,54 +16,70 @@
"""Tests for subunit.tag_stream."""
-import unittest
+from io import BytesIO
-from testtools.compat import StringIO
+import testtools
+from testtools.matchers import Contains
import subunit
import subunit.test_results
-class TestSubUnitTags(unittest.TestCase):
+class TestSubUnitTags(testtools.TestCase):
def setUp(self):
- self.original = StringIO()
- self.filtered = StringIO()
+ super(TestSubUnitTags, self).setUp()
+ self.original = BytesIO()
+ self.filtered = BytesIO()
def test_add_tag(self):
- self.original.write("tags: foo\n")
- self.original.write("test: test\n")
- self.original.write("tags: bar -quux\n")
- self.original.write("success: test\n")
+ # Literal values to avoid set sort-order dependencies. Python code show
+ # derivation.
+ # reference = BytesIO()
+ # stream = subunit.StreamResultToBytes(reference)
+ # stream.status(
+ # test_id='test', test_status='inprogress', test_tags=set(['quux', 'foo']))
+ # stream.status(
+ # test_id='test', test_status='success', test_tags=set(['bar', 'quux', 'foo']))
+ reference = [
+ b'\xb3)\x82\x17\x04test\x02\x04quux\x03foo\x05\x97n\x86\xb3)'
+ b'\x83\x1b\x04test\x03\x03bar\x04quux\x03fooqn\xab)',
+ b'\xb3)\x82\x17\x04test\x02\x04quux\x03foo\x05\x97n\x86\xb3)'
+ b'\x83\x1b\x04test\x03\x04quux\x03foo\x03bar\xaf\xbd\x9d\xd6',
+ b'\xb3)\x82\x17\x04test\x02\x04quux\x03foo\x05\x97n\x86\xb3)'
+ b'\x83\x1b\x04test\x03\x04quux\x03bar\x03foo\x03\x04b\r',
+ b'\xb3)\x82\x17\x04test\x02\x04quux\x03foo\x05\x97n\x86\xb3)'
+ b'\x83\x1b\x04test\x03\x03bar\x03foo\x04quux\xd2\x18\x1bC',
+ b'\xb3)\x82\x17\x04test\x02\x03foo\x04quux\xa6\xe1\xde\xec\xb3)'
+ b'\x83\x1b\x04test\x03\x03foo\x04quux\x03bar\x08\xc2X\x83',
+ b'\xb3)\x82\x17\x04test\x02\x03foo\x04quux\xa6\xe1\xde\xec\xb3)'
+ b'\x83\x1b\x04test\x03\x03bar\x03foo\x04quux\xd2\x18\x1bC',
+ b'\xb3)\x82\x17\x04test\x02\x03foo\x04quux\xa6\xe1\xde\xec\xb3)'
+ b'\x83\x1b\x04test\x03\x03foo\x03bar\x04quux:\x05e\x80',
+ ]
+ stream = subunit.StreamResultToBytes(self.original)
+ stream.status(
+ test_id='test', test_status='inprogress', test_tags=set(['foo']))
+ stream.status(
+ test_id='test', test_status='success', test_tags=set(['foo', 'bar']))
self.original.seek(0)
- result = subunit.tag_stream(self.original, self.filtered, ["quux"])
- self.assertEqual([
- "tags: quux",
- "tags: foo",
- "test: test",
- "tags: bar",
- "success: test",
- ],
- self.filtered.getvalue().splitlines())
+ self.assertEqual(
+ 0, subunit.tag_stream(self.original, self.filtered, ["quux"]))
+ self.assertThat(reference, Contains(self.filtered.getvalue()))
def test_remove_tag(self):
- self.original.write("tags: foo\n")
- self.original.write("test: test\n")
- self.original.write("tags: bar -quux\n")
- self.original.write("success: test\n")
+ reference = BytesIO()
+ stream = subunit.StreamResultToBytes(reference)
+ stream.status(
+ test_id='test', test_status='inprogress', test_tags=set(['foo']))
+ stream.status(
+ test_id='test', test_status='success', test_tags=set(['foo']))
+ stream = subunit.StreamResultToBytes(self.original)
+ stream.status(
+ test_id='test', test_status='inprogress', test_tags=set(['foo']))
+ stream.status(
+ test_id='test', test_status='success', test_tags=set(['foo', 'bar']))
self.original.seek(0)
- result = subunit.tag_stream(self.original, self.filtered, ["-bar"])
- self.assertEqual([
- "tags: -bar",
- "tags: foo",
- "test: test",
- "tags: -quux",
- "success: test",
- ],
- self.filtered.getvalue().splitlines())
-
-
-def test_suite():
- loader = subunit.tests.TestUtil.TestLoader()
- result = loader.loadTestsFromName(__name__)
- return result
+ self.assertEqual(
+ 0, subunit.tag_stream(self.original, self.filtered, ["-bar"]))
+ self.assertEqual(reference.getvalue(), self.filtered.getvalue())
diff --git a/lib/subunit/python/subunit/tests/test_tap2subunit.py b/lib/subunit/python/subunit/tests/test_tap2subunit.py
index 11bc191..5b7c07a 100644
--- a/lib/subunit/python/subunit/tests/test_tap2subunit.py
+++ b/lib/subunit/python/subunit/tests/test_tap2subunit.py
@@ -16,14 +16,19 @@
"""Tests for TAP2SubUnit."""
+from io import BytesIO, StringIO
import unittest
-from testtools.compat import StringIO
+from testtools import TestCase
+from testtools.compat import _u
+from testtools.testresult.doubles import StreamResult
import subunit
+UTF8_TEXT = 'text/plain; charset=UTF8'
-class TestTAP2SubUnit(unittest.TestCase):
+
+class TestTAP2SubUnit(TestCase):
"""Tests for TAP2SubUnit.
These tests test TAP string data in, and subunit string data out.
@@ -34,24 +39,21 @@ class TestTAP2SubUnit(unittest.TestCase):
"""
def setUp(self):
+ super(TestTAP2SubUnit, self).setUp()
self.tap = StringIO()
- self.subunit = StringIO()
+ self.subunit = BytesIO()
def test_skip_entire_file(self):
# A file
# 1..- # Skipped: comment
# results in a single skipped test.
- self.tap.write("1..0 # Skipped: entire file skipped\n")
+ self.tap.write(_u("1..0 # Skipped: entire file skipped\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test file skip",
- "skip file skip [",
- "Skipped: entire file skipped",
- "]",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'file skip', 'skip', None, True,
+ 'tap comment', b'Skipped: entire file skipped', True, None, None,
+ None)])
def test_ok_test_pass(self):
# A file
@@ -59,164 +61,128 @@ class TestTAP2SubUnit(unittest.TestCase):
# results in a passed test with name 'test 1' (a synthetic name as tap
# does not require named fixtures - it is the first test in the tap
# stream).
- self.tap.write("ok\n")
+ self.tap.write(_u("ok\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1",
- "success test 1",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1', 'success', None, False, None,
+ None, True, None, None, None)])
def test_ok_test_number_pass(self):
# A file
# ok 1
# results in a passed test with name 'test 1'
- self.tap.write("ok 1\n")
+ self.tap.write(_u("ok 1\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1",
- "success test 1",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1', 'success', None, False, None,
+ None, True, None, None, None)])
def test_ok_test_number_description_pass(self):
# A file
# ok 1 - There is a description
# results in a passed test with name 'test 1 - There is a description'
- self.tap.write("ok 1 - There is a description\n")
+ self.tap.write(_u("ok 1 - There is a description\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1 - There is a description",
- "success test 1 - There is a description",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1 - There is a description',
+ 'success', None, False, None, None, True, None, None, None)])
def test_ok_test_description_pass(self):
# A file
# ok There is a description
# results in a passed test with name 'test 1 There is a description'
- self.tap.write("ok There is a description\n")
+ self.tap.write(_u("ok There is a description\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1 There is a description",
- "success test 1 There is a description",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1 There is a description',
+ 'success', None, False, None, None, True, None, None, None)])
def test_ok_SKIP_skip(self):
# A file
# ok # SKIP
# results in a skkip test with name 'test 1'
- self.tap.write("ok # SKIP\n")
+ self.tap.write(_u("ok # SKIP\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1",
- "skip test 1",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1', 'skip', None, False, None,
+ None, True, None, None, None)])
def test_ok_skip_number_comment_lowercase(self):
- self.tap.write("ok 1 # skip no samba environment available, skipping compilation\n")
+ self.tap.write(_u("ok 1 # skip no samba environment available, skipping compilation\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1",
- "skip test 1 [",
- "no samba environment available, skipping compilation",
- "]"
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1', 'skip', None, False, 'tap comment',
+ b'no samba environment available, skipping compilation', True,
+ 'text/plain; charset=UTF8', None, None)])
def test_ok_number_description_SKIP_skip_comment(self):
# A file
# ok 1 foo # SKIP Not done yet
# results in a skip test with name 'test 1 foo' and a log of
# Not done yet
- self.tap.write("ok 1 foo # SKIP Not done yet\n")
+ self.tap.write(_u("ok 1 foo # SKIP Not done yet\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1 foo",
- "skip test 1 foo [",
- "Not done yet",
- "]",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1 foo', 'skip', None, False,
+ 'tap comment', b'Not done yet', True, 'text/plain; charset=UTF8',
+ None, None)])
def test_ok_SKIP_skip_comment(self):
# A file
# ok # SKIP Not done yet
# results in a skip test with name 'test 1' and a log of Not done yet
- self.tap.write("ok # SKIP Not done yet\n")
+ self.tap.write(_u("ok # SKIP Not done yet\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1",
- "skip test 1 [",
- "Not done yet",
- "]",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1', 'skip', None, False,
+ 'tap comment', b'Not done yet', True, 'text/plain; charset=UTF8',
+ None, None)])
def test_ok_TODO_xfail(self):
# A file
# ok # TODO
# results in a xfail test with name 'test 1'
- self.tap.write("ok # TODO\n")
+ self.tap.write(_u("ok # TODO\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1",
- "xfail test 1",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1', 'xfail', None, False, None,
+ None, True, None, None, None)])
def test_ok_TODO_xfail_comment(self):
# A file
# ok # TODO Not done yet
# results in a xfail test with name 'test 1' and a log of Not done yet
- self.tap.write("ok # TODO Not done yet\n")
+ self.tap.write(_u("ok # TODO Not done yet\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1",
- "xfail test 1 [",
- "Not done yet",
- "]",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([('status', 'test 1', 'xfail', None, False,
+ 'tap comment', b'Not done yet', True, 'text/plain; charset=UTF8',
+ None, None)])
def test_bail_out_errors(self):
# A file with line in it
# Bail out! COMMENT
# is treated as an error
- self.tap.write("ok 1 foo\n")
- self.tap.write("Bail out! Lifejacket engaged\n")
+ self.tap.write(_u("ok 1 foo\n"))
+ self.tap.write(_u("Bail out! Lifejacket engaged\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- "test test 1 foo",
- "success test 1 foo",
- "test Bail out! Lifejacket engaged",
- "error Bail out! Lifejacket engaged",
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([
+ ('status', 'test 1 foo', 'success', None, False, None, None, True,
+ None, None, None),
+ ('status', 'Bail out! Lifejacket engaged', 'fail', None, False,
+ None, None, True, None, None, None)])
def test_missing_test_at_end_with_plan_adds_error(self):
# A file
@@ -224,23 +190,20 @@ class TestTAP2SubUnit(unittest.TestCase):
# ok first test
# not ok third test
# results in three tests, with the third being created
- self.tap.write('1..3\n')
- self.tap.write('ok first test\n')
- self.tap.write('not ok second test\n')
+ self.tap.write(_u('1..3\n'))
+ self.tap.write(_u('ok first test\n'))
+ self.tap.write(_u('not ok second test\n'))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1 first test',
- 'success test 1 first test',
- 'test test 2 second test',
- 'failure test 2 second test',
- 'test test 3',
- 'error test 3 [',
- 'test missing from TAP output',
- ']',
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([
+ ('status', 'test 1 first test', 'success', None, False, None,
+ None, True, None, None, None),
+ ('status', 'test 2 second test', 'fail', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 3', 'fail', None, False, 'tap meta',
+ b'test missing from TAP output', True, 'text/plain; charset=UTF8',
+ None, None)])
def test_missing_test_with_plan_adds_error(self):
# A file
@@ -248,45 +211,39 @@ class TestTAP2SubUnit(unittest.TestCase):
# ok first test
# not ok 3 third test
# results in three tests, with the second being created
- self.tap.write('1..3\n')
- self.tap.write('ok first test\n')
- self.tap.write('not ok 3 third test\n')
+ self.tap.write(_u('1..3\n'))
+ self.tap.write(_u('ok first test\n'))
+ self.tap.write(_u('not ok 3 third test\n'))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1 first test',
- 'success test 1 first test',
- 'test test 2',
- 'error test 2 [',
- 'test missing from TAP output',
- ']',
- 'test test 3 third test',
- 'failure test 3 third test',
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([
+ ('status', 'test 1 first test', 'success', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 2', 'fail', None, False, 'tap meta',
+ b'test missing from TAP output', True, 'text/plain; charset=UTF8',
+ None, None),
+ ('status', 'test 3 third test', 'fail', None, False, None, None,
+ True, None, None, None)])
def test_missing_test_no_plan_adds_error(self):
# A file
# ok first test
# not ok 3 third test
# results in three tests, with the second being created
- self.tap.write('ok first test\n')
- self.tap.write('not ok 3 third test\n')
+ self.tap.write(_u('ok first test\n'))
+ self.tap.write(_u('not ok 3 third test\n'))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1 first test',
- 'success test 1 first test',
- 'test test 2',
- 'error test 2 [',
- 'test missing from TAP output',
- ']',
- 'test test 3 third test',
- 'failure test 3 third test',
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([
+ ('status', 'test 1 first test', 'success', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 2', 'fail', None, False, 'tap meta',
+ b'test missing from TAP output', True, 'text/plain; charset=UTF8',
+ None, None),
+ ('status', 'test 3 third test', 'fail', None, False, None, None,
+ True, None, None, None)])
def test_four_tests_in_a_row_trailing_plan(self):
# A file
@@ -296,25 +253,23 @@ class TestTAP2SubUnit(unittest.TestCase):
# not ok 4 - fourth
# 1..4
# results in four tests numbered and named
- self.tap.write('ok 1 - first test in a script with trailing plan\n')
- self.tap.write('not ok 2 - second\n')
- self.tap.write('ok 3 - third\n')
- self.tap.write('not ok 4 - fourth\n')
- self.tap.write('1..4\n')
+ self.tap.write(_u('ok 1 - first test in a script with trailing plan\n'))
+ self.tap.write(_u('not ok 2 - second\n'))
+ self.tap.write(_u('ok 3 - third\n'))
+ self.tap.write(_u('not ok 4 - fourth\n'))
+ self.tap.write(_u('1..4\n'))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1 - first test in a script with trailing plan',
- 'success test 1 - first test in a script with trailing plan',
- 'test test 2 - second',
- 'failure test 2 - second',
- 'test test 3 - third',
- 'success test 3 - third',
- 'test test 4 - fourth',
- 'failure test 4 - fourth'
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([
+ ('status', 'test 1 - first test in a script with trailing plan',
+ 'success', None, False, None, None, True, None, None, None),
+ ('status', 'test 2 - second', 'fail', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 3 - third', 'success', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 4 - fourth', 'fail', None, False, None, None,
+ True, None, None, None)])
def test_four_tests_in_a_row_with_plan(self):
# A file
@@ -324,25 +279,23 @@ class TestTAP2SubUnit(unittest.TestCase):
# ok 3 - third
# not ok 4 - fourth
# results in four tests numbered and named
- self.tap.write('1..4\n')
- self.tap.write('ok 1 - first test in a script with a plan\n')
- self.tap.write('not ok 2 - second\n')
- self.tap.write('ok 3 - third\n')
- self.tap.write('not ok 4 - fourth\n')
+ self.tap.write(_u('1..4\n'))
+ self.tap.write(_u('ok 1 - first test in a script with a plan\n'))
+ self.tap.write(_u('not ok 2 - second\n'))
+ self.tap.write(_u('ok 3 - third\n'))
+ self.tap.write(_u('not ok 4 - fourth\n'))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1 - first test in a script with a plan',
- 'success test 1 - first test in a script with a plan',
- 'test test 2 - second',
- 'failure test 2 - second',
- 'test test 3 - third',
- 'success test 3 - third',
- 'test test 4 - fourth',
- 'failure test 4 - fourth'
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([
+ ('status', 'test 1 - first test in a script with a plan',
+ 'success', None, False, None, None, True, None, None, None),
+ ('status', 'test 2 - second', 'fail', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 3 - third', 'success', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 4 - fourth', 'fail', None, False, None, None,
+ True, None, None, None)])
def test_four_tests_in_a_row_no_plan(self):
# A file
@@ -351,46 +304,43 @@ class TestTAP2SubUnit(unittest.TestCase):
# ok 3 - third
# not ok 4 - fourth
# results in four tests numbered and named
- self.tap.write('ok 1 - first test in a script with no plan at all\n')
- self.tap.write('not ok 2 - second\n')
- self.tap.write('ok 3 - third\n')
- self.tap.write('not ok 4 - fourth\n')
+ self.tap.write(_u('ok 1 - first test in a script with no plan at all\n'))
+ self.tap.write(_u('not ok 2 - second\n'))
+ self.tap.write(_u('ok 3 - third\n'))
+ self.tap.write(_u('not ok 4 - fourth\n'))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1 - first test in a script with no plan at all',
- 'success test 1 - first test in a script with no plan at all',
- 'test test 2 - second',
- 'failure test 2 - second',
- 'test test 3 - third',
- 'success test 3 - third',
- 'test test 4 - fourth',
- 'failure test 4 - fourth'
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([
+ ('status', 'test 1 - first test in a script with no plan at all',
+ 'success', None, False, None, None, True, None, None, None),
+ ('status', 'test 2 - second', 'fail', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 3 - third', 'success', None, False, None, None,
+ True, None, None, None),
+ ('status', 'test 4 - fourth', 'fail', None, False, None, None,
+ True, None, None, None)])
def test_todo_and_skip(self):
# A file
# not ok 1 - a fail but # TODO but is TODO
# not ok 2 - another fail # SKIP instead
# results in two tests, numbered and commented.
- self.tap.write("not ok 1 - a fail but # TODO but is TODO\n")
- self.tap.write("not ok 2 - another fail # SKIP instead\n")
+ self.tap.write(_u("not ok 1 - a fail but # TODO but is TODO\n"))
+ self.tap.write(_u("not ok 2 - another fail # SKIP instead\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1 - a fail but',
- 'xfail test 1 - a fail but [',
- 'but is TODO',
- ']',
- 'test test 2 - another fail',
- 'skip test 2 - another fail [',
- 'instead',
- ']',
- ],
- self.subunit.getvalue().splitlines())
+ self.subunit.seek(0)
+ events = StreamResult()
+ subunit.ByteStreamToStreamResult(self.subunit).run(events)
+ self.check_events([
+ ('status', 'test 1 - a fail but', 'xfail', None, False,
+ 'tap comment', b'but is TODO', True, 'text/plain; charset=UTF8',
+ None, None),
+ ('status', 'test 2 - another fail', 'skip', None, False,
+ 'tap comment', b'instead', True, 'text/plain; charset=UTF8',
+ None, None)])
def test_leading_comments_add_to_next_test_log(self):
# A file
@@ -399,21 +349,17 @@ class TestTAP2SubUnit(unittest.TestCase):
# ok
# results in a single test with the comment included
# in the first test and not the second.
- self.tap.write("# comment\n")
- self.tap.write("ok\n")
- self.tap.write("ok\n")
+ self.tap.write(_u("# comment\n"))
+ self.tap.write(_u("ok\n"))
+ self.tap.write(_u("ok\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1',
- 'success test 1 [',
- '# comment',
- ']',
- 'test test 2',
- 'success test 2',
- ],
- self.subunit.getvalue().splitlines())
+ self.check_events([
+ ('status', 'test 1', 'success', None, False, 'tap comment',
+ b'# comment', True, 'text/plain; charset=UTF8', None, None),
+ ('status', 'test 2', 'success', None, False, None, None, True,
+ None, None, None)])
def test_trailing_comments_are_included_in_last_test_log(self):
# A file
@@ -422,24 +368,20 @@ class TestTAP2SubUnit(unittest.TestCase):
# # comment
# results in a two tests, with the second having the comment
# attached to its log.
- self.tap.write("ok\n")
- self.tap.write("ok\n")
- self.tap.write("# comment\n")
+ self.tap.write(_u("ok\n"))
+ self.tap.write(_u("ok\n"))
+ self.tap.write(_u("# comment\n"))
self.tap.seek(0)
result = subunit.TAP2SubUnit(self.tap, self.subunit)
self.assertEqual(0, result)
- self.assertEqual([
- 'test test 1',
- 'success test 1',
- 'test test 2',
- 'success test 2 [',
- '# comment',
- ']',
- ],
- self.subunit.getvalue().splitlines())
-
+ self.check_events([
+ ('status', 'test 1', 'success', None, False, None, None, True,
+ None, None, None),
+ ('status', 'test 2', 'success', None, False, 'tap comment',
+ b'# comment', True, 'text/plain; charset=UTF8', None, None)])
-def test_suite():
- loader = subunit.tests.TestUtil.TestLoader()
- result = loader.loadTestsFromName(__name__)
- return result
+ def check_events(self, events):
+ self.subunit.seek(0)
+ eventstream = StreamResult()
+ subunit.ByteStreamToStreamResult(self.subunit).run(eventstream)
+ self.assertEqual(events, eventstream._events)
diff --git a/lib/subunit/python/subunit/tests/test_test_protocol.py b/lib/subunit/python/subunit/tests/test_test_protocol.py
index 7831ba1..c6008f4 100644
--- a/lib/subunit/python/subunit/tests/test_test_protocol.py
+++ b/lib/subunit/python/subunit/tests/test_test_protocol.py
@@ -34,9 +34,14 @@ except ImportError:
Python27TestResult,
ExtendedTestResult,
)
+from testtools.matchers import Contains
import subunit
-from subunit import _remote_exception_str, _remote_exception_str_chunked
+from subunit.tests import (
+ _remote_exception_repr,
+ _remote_exception_str,
+ _remote_exception_str_chunked,
+ )
import subunit.iso8601 as iso8601
@@ -104,10 +109,10 @@ class TestTestProtocolServerPipe(unittest.TestCase):
bing = subunit.RemotedTestCase("bing crosby")
an_error = subunit.RemotedTestCase("an error")
self.assertEqual(client.errors,
- [(an_error, _remote_exception_str + '\n')])
+ [(an_error, _remote_exception_repr + '\n')])
self.assertEqual(
client.failures,
- [(bing, _remote_exception_str + ": "
+ [(bing, _remote_exception_repr + ": "
+ details_to_str({'traceback': text_content(traceback)}) + "\n")])
self.assertEqual(client.testsRun, 3)
@@ -962,7 +967,7 @@ class TestRemotedTestCase(unittest.TestCase):
"'A test description'>", "%r" % test)
result = unittest.TestResult()
test.run(result)
- self.assertEqual([(test, _remote_exception_str + ": "
+ self.assertEqual([(test, _remote_exception_repr + ": "
"Cannot run RemotedTestCases.\n\n")],
result.errors)
self.assertEqual(1, result.testsRun)
@@ -1128,9 +1133,10 @@ class TestIsolatedTestSuite(TestCase):
self.assertEqual(self.SampleTestToIsolate.TEST, False)
-class TestTestProtocolClient(unittest.TestCase):
+class TestTestProtocolClient(TestCase):
def setUp(self):
+ super(TestTestProtocolClient, self).setUp()
self.io = BytesIO()
self.protocol = subunit.TestProtocolClient(self.io)
self.unicode_test = PlaceHolder(_u('\u2603'))
@@ -1191,15 +1197,23 @@ class TestTestProtocolClient(unittest.TestCase):
"""Test addFailure on a TestProtocolClient with details."""
self.protocol.addFailure(
self.test, details=self.sample_tb_details)
- self.assertEqual(
- self.io.getvalue(),
+ self.assertThat([
_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()))
+ "traceback\n" + _remote_exception_str_chunked +
+ "]\n") % self.test.id()),
+ _b(("failure: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "something\n"
+ "F\r\nserialised\nform0\r\n"
+ "Content-Type: text/x-traceback;language=python,charset=utf8\n"
+ "traceback\n" + _remote_exception_str_chunked +
+ "]\n") % self.test.id()),
+ ],
+ Contains(self.io.getvalue())),
def test_add_error(self):
"""Test stopTest on a TestProtocolClient."""
@@ -1215,15 +1229,23 @@ class TestTestProtocolClient(unittest.TestCase):
"""Test stopTest on a TestProtocolClient with details."""
self.protocol.addError(
self.test, details=self.sample_tb_details)
- self.assertEqual(
- self.io.getvalue(),
+ self.assertThat([
_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()))
+ "traceback\n" + _remote_exception_str_chunked +
+ "]\n") % self.test.id()),
+ _b(("error: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "something\n"
+ "F\r\nserialised\nform0\r\n"
+ "Content-Type: text/x-traceback;language=python,charset=utf8\n"
+ "traceback\n" + _remote_exception_str_chunked +
+ "]\n") % self.test.id()),
+ ],
+ Contains(self.io.getvalue())),
def test_add_expected_failure(self):
"""Test addExpectedFailure on a TestProtocolClient."""
@@ -1239,16 +1261,23 @@ class TestTestProtocolClient(unittest.TestCase):
"""Test addExpectedFailure on a TestProtocolClient with details."""
self.protocol.addExpectedFailure(
self.test, details=self.sample_tb_details)
- self.assertEqual(
- self.io.getvalue(),
+ self.assertThat([
_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()))
-
+ "traceback\n" + _remote_exception_str_chunked +
+ "]\n") % self.test.id()),
+ _b(("xfail: %s [ multipart\n"
+ "Content-Type: text/plain\n"
+ "something\n"
+ "F\r\nserialised\nform0\r\n"
+ "Content-Type: text/x-traceback;language=python,charset=utf8\n"
+ "traceback\n" + _remote_exception_str_chunked +
+ "]\n") % self.test.id()),
+ ],
+ Contains(self.io.getvalue())),
def test_add_skip(self):
"""Test addSkip on a TestProtocolClient."""
@@ -1324,14 +1353,10 @@ class TestTestProtocolClient(unittest.TestCase):
def test_tags_both(self):
self.protocol.tags(set(['quux']), set(['bar']))
- self.assertEqual(_b("tags: quux -bar\n"), self.io.getvalue())
+ self.assertThat(
+ [b"tags: quux -bar\n", b"tags: -bar quux\n"],
+ Contains(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_protocol2.py b/lib/subunit/python/subunit/tests/test_test_protocol2.py
new file mode 100644
index 0000000..c21392c
--- /dev/null
+++ b/lib/subunit/python/subunit/tests/test_test_protocol2.py
@@ -0,0 +1,436 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Robert Collins <robertc at 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 io import BytesIO
+import datetime
+
+from testtools import TestCase
+from testtools.matchers import Contains, HasLength
+from testtools.tests.test_testresult import TestStreamResultContract
+from testtools.testresult.doubles import StreamResult
+
+import subunit
+import subunit.iso8601 as iso8601
+
+CONSTANT_ENUM = b'\xb3)\x01\x0c\x03foo\x08U_\x1b'
+CONSTANT_INPROGRESS = b'\xb3)\x02\x0c\x03foo\x8e\xc1-\xb5'
+CONSTANT_SUCCESS = b'\xb3)\x03\x0c\x03fooE\x9d\xfe\x10'
+CONSTANT_UXSUCCESS = b'\xb3)\x04\x0c\x03fooX\x98\xce\xa8'
+CONSTANT_SKIP = b'\xb3)\x05\x0c\x03foo\x93\xc4\x1d\r'
+CONSTANT_FAIL = b'\xb3)\x06\x0c\x03foo\x15Po\xa3'
+CONSTANT_XFAIL = b'\xb3)\x07\x0c\x03foo\xde\x0c\xbc\x06'
+CONSTANT_EOF = b'\xb3!\x10\x08S\x15\x88\xdc'
+CONSTANT_FILE_CONTENT = b'\xb3!@\x13\x06barney\x03wooA5\xe3\x8c'
+CONSTANT_MIME = b'\xb3! #\x1aapplication/foo; charset=1x3Q\x15'
+CONSTANT_TIMESTAMP = b'\xb3+\x03\x13<\x17T\xcf\x80\xaf\xc8\x03barI\x96>-'
+CONSTANT_ROUTE_CODE = b'\xb3-\x03\x13\x03bar\x06source\x9cY9\x19'
+CONSTANT_RUNNABLE = b'\xb3(\x03\x0c\x03foo\xe3\xea\xf5\xa4'
+CONSTANT_TAGS = [
+ b'\xb3)\x80\x15\x03bar\x02\x03foo\x03barTHn\xb4',
+ b'\xb3)\x80\x15\x03bar\x02\x03bar\x03foo\xf8\xf1\x91o',
+ ]
+
+
+class TestStreamResultToBytesContract(TestCase, TestStreamResultContract):
+ """Check that StreamResult behaves as testtools expects."""
+
+ def _make_result(self):
+ return subunit.StreamResultToBytes(BytesIO())
+
+
+class TestStreamResultToBytes(TestCase):
+
+ def _make_result(self):
+ output = BytesIO()
+ return subunit.StreamResultToBytes(output), output
+
+ def test_numbers(self):
+ result = subunit.StreamResultToBytes(BytesIO())
+ packet = []
+ self.assertRaises(Exception, result._write_number, -1, packet)
+ self.assertEqual([], packet)
+ result._write_number(0, packet)
+ self.assertEqual([b'\x00'], packet)
+ del packet[:]
+ result._write_number(63, packet)
+ self.assertEqual([b'\x3f'], packet)
+ del packet[:]
+ result._write_number(64, packet)
+ self.assertEqual([b'\x40\x40'], packet)
+ del packet[:]
+ result._write_number(16383, packet)
+ self.assertEqual([b'\x7f\xff'], packet)
+ del packet[:]
+ result._write_number(16384, packet)
+ self.assertEqual([b'\x80\x40', b'\x00'], packet)
+ del packet[:]
+ result._write_number(4194303, packet)
+ self.assertEqual([b'\xbf\xff', b'\xff'], packet)
+ del packet[:]
+ result._write_number(4194304, packet)
+ self.assertEqual([b'\xc0\x40\x00\x00'], packet)
+ del packet[:]
+ result._write_number(1073741823, packet)
+ self.assertEqual([b'\xff\xff\xff\xff'], packet)
+ del packet[:]
+ self.assertRaises(Exception, result._write_number, 1073741824, packet)
+ self.assertEqual([], packet)
+
+ def test_volatile_length(self):
+ # if the length of the packet data before the length itself is
+ # considered is right on the boundary for length's variable length
+ # encoding, it is easy to get the length wrong by not accounting for
+ # length itself.
+ # that is, the encoder has to ensure that length == sum (length_of_rest
+ # + length_of_length)
+ result, output = self._make_result()
+ # 1 byte short:
+ result.status(file_name="", file_bytes=b'\xff'*0)
+ self.assertThat(output.getvalue(), HasLength(10))
+ self.assertEqual(b'\x0a', output.getvalue()[3:4])
+ output.seek(0)
+ output.truncate()
+ # 1 byte long:
+ result.status(file_name="", file_bytes=b'\xff'*53)
+ self.assertThat(output.getvalue(), HasLength(63))
+ self.assertEqual(b'\x3f', output.getvalue()[3:4])
+ output.seek(0)
+ output.truncate()
+ # 2 bytes short
+ result.status(file_name="", file_bytes=b'\xff'*54)
+ self.assertThat(output.getvalue(), HasLength(65))
+ self.assertEqual(b'\x40\x41', output.getvalue()[3:5])
+ output.seek(0)
+ output.truncate()
+ # 2 bytes long
+ result.status(file_name="", file_bytes=b'\xff'*16371)
+ self.assertThat(output.getvalue(), HasLength(16383))
+ self.assertEqual(b'\x7f\xff', output.getvalue()[3:5])
+ output.seek(0)
+ output.truncate()
+ # 3 bytes short
+ result.status(file_name="", file_bytes=b'\xff'*16372)
+ self.assertThat(output.getvalue(), HasLength(16385))
+ self.assertEqual(b'\x80\x40\x01', output.getvalue()[3:6])
+ output.seek(0)
+ output.truncate()
+ # 3 bytes long
+ result.status(file_name="", file_bytes=b'\xff'*4194289)
+ self.assertThat(output.getvalue(), HasLength(4194303))
+ self.assertEqual(b'\xbf\xff\xff', output.getvalue()[3:6])
+ output.seek(0)
+ output.truncate()
+ self.assertRaises(Exception, result.status, file_name="",
+ file_bytes=b'\xff'*4194290)
+
+ def test_trivial_enumeration(self):
+ result, output = self._make_result()
+ result.status("foo", 'exists')
+ self.assertEqual(CONSTANT_ENUM, output.getvalue())
+
+ def test_inprogress(self):
+ result, output = self._make_result()
+ result.status("foo", 'inprogress')
+ self.assertEqual(CONSTANT_INPROGRESS, output.getvalue())
+
+ def test_success(self):
+ result, output = self._make_result()
+ result.status("foo", 'success')
+ self.assertEqual(CONSTANT_SUCCESS, output.getvalue())
+
+ def test_uxsuccess(self):
+ result, output = self._make_result()
+ result.status("foo", 'uxsuccess')
+ self.assertEqual(CONSTANT_UXSUCCESS, output.getvalue())
+
+ def test_skip(self):
+ result, output = self._make_result()
+ result.status("foo", 'skip')
+ self.assertEqual(CONSTANT_SKIP, output.getvalue())
+
+ def test_fail(self):
+ result, output = self._make_result()
+ result.status("foo", 'fail')
+ self.assertEqual(CONSTANT_FAIL, output.getvalue())
+
+ def test_xfail(self):
+ result, output = self._make_result()
+ result.status("foo", 'xfail')
+ self.assertEqual(CONSTANT_XFAIL, output.getvalue())
+
+ def test_unknown_status(self):
+ result, output = self._make_result()
+ self.assertRaises(Exception, result.status, "foo", 'boo')
+ self.assertEqual(b'', output.getvalue())
+
+ def test_eof(self):
+ result, output = self._make_result()
+ result.status(eof=True)
+ self.assertEqual(CONSTANT_EOF, output.getvalue())
+
+ def test_file_content(self):
+ result, output = self._make_result()
+ result.status(file_name="barney", file_bytes=b"woo")
+ self.assertEqual(CONSTANT_FILE_CONTENT, output.getvalue())
+
+ def test_mime(self):
+ result, output = self._make_result()
+ result.status(mime_type="application/foo; charset=1")
+ self.assertEqual(CONSTANT_MIME, output.getvalue())
+
+ def test_route_code(self):
+ result, output = self._make_result()
+ result.status(test_id="bar", test_status='success',
+ route_code="source")
+ self.assertEqual(CONSTANT_ROUTE_CODE, output.getvalue())
+
+ def test_runnable(self):
+ result, output = self._make_result()
+ result.status("foo", 'success', runnable=False)
+ self.assertEqual(CONSTANT_RUNNABLE, output.getvalue())
+
+ def test_tags(self):
+ result, output = self._make_result()
+ result.status(test_id="bar", test_tags=set(['foo', 'bar']))
+ self.assertThat(CONSTANT_TAGS, Contains(output.getvalue()))
+
+ def test_timestamp(self):
+ timestamp = datetime.datetime(2001, 12, 12, 12, 59, 59, 45,
+ iso8601.Utc())
+ result, output = self._make_result()
+ result.status(test_id="bar", test_status='success', timestamp=timestamp)
+ self.assertEqual(CONSTANT_TIMESTAMP, output.getvalue())
+
+
+class TestByteStreamToStreamResult(TestCase):
+
+ def test_non_subunit_encapsulated(self):
+ source = BytesIO(b"foo\nbar\n")
+ result = StreamResult()
+ subunit.ByteStreamToStreamResult(
+ source, non_subunit_name="stdout").run(result)
+ self.assertEqual([
+ ('status', None, None, None, True, 'stdout', b'f', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'o', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'o', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'\n', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'b', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'a', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'r', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'\n', False, None, None, None),
+ ], result._events)
+ self.assertEqual(b'', source.read())
+
+ def test_signature_middle_utf8_char(self):
+ utf8_bytes = b'\xe3\xb3\x8a'
+ source = BytesIO(utf8_bytes)
+ # Should be treated as one character (it is u'\u3cca') and wrapped
+ result = StreamResult()
+ subunit.ByteStreamToStreamResult(
+ source, non_subunit_name="stdout").run(
+ result)
+ self.assertEqual([
+ ('status', None, None, None, True, 'stdout', b'\xe3', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'\xb3', False, None, None, None),
+ ('status', None, None, None, True, 'stdout', b'\x8a', False, None, None, None),
+ ], result._events)
+
+ def test_non_subunit_disabled_raises(self):
+ source = BytesIO(b"foo\nbar\n")
+ result = StreamResult()
+ case = subunit.ByteStreamToStreamResult(source)
+ e = self.assertRaises(Exception, case.run, result)
+ self.assertEqual(b'f', e.args[1])
+ self.assertEqual(b'oo\nbar\n', source.read())
+ self.assertEqual([], result._events)
+
+ def test_trivial_enumeration(self):
+ source = BytesIO(CONSTANT_ENUM)
+ result = StreamResult()
+ subunit.ByteStreamToStreamResult(
+ source, non_subunit_name="stdout").run(result)
+ self.assertEqual(b'', source.read())
+ self.assertEqual([
+ ('status', 'foo', 'exists', None, True, None, None, False, None, None, None),
+ ], result._events)
+
+ def test_multiple_events(self):
+ source = BytesIO(CONSTANT_ENUM + CONSTANT_ENUM)
+ result = StreamResult()
+ subunit.ByteStreamToStreamResult(
+ source, non_subunit_name="stdout").run(result)
+ self.assertEqual(b'', source.read())
+ self.assertEqual([
+ ('status', 'foo', 'exists', None, True, None, None, False, None, None, None),
+ ('status', 'foo', 'exists', None, True, None, None, False, None, None, None),
+ ], result._events)
+
+ def test_inprogress(self):
+ self.check_event(CONSTANT_INPROGRESS, 'inprogress')
+
+ def test_success(self):
+ self.check_event(CONSTANT_SUCCESS, 'success')
+
+ def test_uxsuccess(self):
+ self.check_event(CONSTANT_UXSUCCESS, 'uxsuccess')
+
+ def test_skip(self):
+ self.check_event(CONSTANT_SKIP, 'skip')
+
+ def test_fail(self):
+ self.check_event(CONSTANT_FAIL, 'fail')
+
+ def test_xfail(self):
+ self.check_event(CONSTANT_XFAIL, 'xfail')
+
+ def check_events(self, source_bytes, events):
+ source = BytesIO(source_bytes)
+ result = StreamResult()
+ subunit.ByteStreamToStreamResult(
+ source, non_subunit_name="stdout").run(result)
+ self.assertEqual(b'', source.read())
+ self.assertEqual(events, result._events)
+ #- any file attachments should be byte contents [as users assume that].
+ for event in result._events:
+ if event[5] is not None:
+ self.assertIsInstance(event[6], bytes)
+
+ def check_event(self, source_bytes, test_status=None, test_id="foo",
+ route_code=None, timestamp=None, tags=None, mime_type=None,
+ file_name=None, file_bytes=None, eof=False, runnable=True):
+ event = self._event(test_id=test_id, test_status=test_status,
+ tags=tags, runnable=runnable, file_name=file_name,
+ file_bytes=file_bytes, eof=eof, mime_type=mime_type,
+ route_code=route_code, timestamp=timestamp)
+ self.check_events(source_bytes, [event])
+
+ def _event(self, test_status=None, test_id=None, route_code=None,
+ timestamp=None, tags=None, mime_type=None, file_name=None,
+ file_bytes=None, eof=False, runnable=True):
+ return ('status', test_id, test_status, tags, runnable, file_name,
+ file_bytes, eof, mime_type, route_code, timestamp)
+
+ def test_eof(self):
+ self.check_event(CONSTANT_EOF, test_id=None, eof=True)
+
+ def test_file_content(self):
+ self.check_event(CONSTANT_FILE_CONTENT,
+ test_id=None, file_name="barney", file_bytes=b"woo")
+
+ def test_file_content_length_into_checksum(self):
+ # A bad file content length which creeps into the checksum.
+ bad_file_length_content = b'\xb3!@\x13\x06barney\x04woo\xdc\xe2\xdb\x35'
+ self.check_events(bad_file_length_content, [
+ self._event(test_id="subunit.parser", eof=True,
+ file_name="Packet data", file_bytes=bad_file_length_content,
+ mime_type="application/octet-stream"),
+ self._event(test_id="subunit.parser", test_status="fail", eof=True,
+ file_name="Parser Error",
+ file_bytes=b"File content extends past end of packet: claimed 4 bytes, 3 available",
+ mime_type="text/plain;charset=utf8"),
+ ])
+
+ def test_packet_length_4_word_varint(self):
+ packet_data = b'\xb3!@\xc0\x00\x11'
+ self.check_events(packet_data, [
+ self._event(test_id="subunit.parser", eof=True,
+ file_name="Packet data", file_bytes=packet_data,
+ mime_type="application/octet-stream"),
+ self._event(test_id="subunit.parser", test_status="fail", eof=True,
+ file_name="Parser Error",
+ file_bytes=b"3 byte maximum given but 4 byte value found.",
+ mime_type="text/plain;charset=utf8"),
+ ])
+
+ def test_mime(self):
+ self.check_event(CONSTANT_MIME,
+ test_id=None, mime_type='application/foo; charset=1')
+
+ def test_route_code(self):
+ self.check_event(CONSTANT_ROUTE_CODE,
+ 'success', route_code="source", test_id="bar")
+
+ def test_runnable(self):
+ self.check_event(CONSTANT_RUNNABLE,
+ test_status='success', runnable=False)
+
+ def test_tags(self):
+ self.check_event(CONSTANT_TAGS[0],
+ None, tags=set(['foo', 'bar']), test_id="bar")
+
+ def test_timestamp(self):
+ timestamp = datetime.datetime(2001, 12, 12, 12, 59, 59, 45,
+ iso8601.Utc())
+ self.check_event(CONSTANT_TIMESTAMP,
+ 'success', test_id='bar', timestamp=timestamp)
+
+ def test_bad_crc_errors_via_status(self):
+ file_bytes = CONSTANT_MIME[:-1] + b'\x00'
+ self.check_events( file_bytes, [
+ self._event(test_id="subunit.parser", eof=True,
+ file_name="Packet data", file_bytes=file_bytes,
+ mime_type="application/octet-stream"),
+ self._event(test_id="subunit.parser", test_status="fail", eof=True,
+ file_name="Parser Error",
+ file_bytes=b'Bad checksum - calculated (0x78335115), '
+ b'stored (0x78335100)',
+ mime_type="text/plain;charset=utf8"),
+ ])
+
+ def test_not_utf8_in_string(self):
+ file_bytes = CONSTANT_ROUTE_CODE[:5] + b'\xb4' + CONSTANT_ROUTE_CODE[6:-4] + b'\xce\x56\xc6\x17'
+ self.check_events(file_bytes, [
+ self._event(test_id="subunit.parser", eof=True,
+ file_name="Packet data", file_bytes=file_bytes,
+ mime_type="application/octet-stream"),
+ self._event(test_id="subunit.parser", test_status="fail", eof=True,
+ file_name="Parser Error",
+ file_bytes=b'UTF8 string at offset 2 is not UTF8',
+ mime_type="text/plain;charset=utf8"),
+ ])
+
+ def test_NULL_in_string(self):
+ file_bytes = CONSTANT_ROUTE_CODE[:6] + b'\x00' + CONSTANT_ROUTE_CODE[7:-4] + b'\xd7\x41\xac\xfe'
+ self.check_events(file_bytes, [
+ self._event(test_id="subunit.parser", eof=True,
+ file_name="Packet data", file_bytes=file_bytes,
+ mime_type="application/octet-stream"),
+ self._event(test_id="subunit.parser", test_status="fail", eof=True,
+ file_name="Parser Error",
+ file_bytes=b'UTF8 string at offset 2 contains NUL byte',
+ mime_type="text/plain;charset=utf8"),
+ ])
+
+ def test_bad_utf8_stringlength(self):
+ file_bytes = CONSTANT_ROUTE_CODE[:4] + b'\x3f' + CONSTANT_ROUTE_CODE[5:-4] + b'\xbe\x29\xe0\xc2'
+ self.check_events(file_bytes, [
+ self._event(test_id="subunit.parser", eof=True,
+ file_name="Packet data", file_bytes=file_bytes,
+ mime_type="application/octet-stream"),
+ self._event(test_id="subunit.parser", test_status="fail", eof=True,
+ file_name="Parser Error",
+ file_bytes=b'UTF8 string at offset 2 extends past end of '
+ b'packet: claimed 63 bytes, 10 available',
+ mime_type="text/plain;charset=utf8"),
+ ])
+
+ def test_route_code_and_file_content(self):
+ content = BytesIO()
+ subunit.StreamResultToBytes(content).status(
+ route_code='0', mime_type='text/plain', file_name='bar',
+ file_bytes=b'foo')
+ self.check_event(content.getvalue(), test_id=None, file_name='bar',
+ route_code='0', mime_type='text/plain', file_bytes=b'foo')
diff --git a/lib/subunit/python/subunit/tests/test_test_results.py b/lib/subunit/python/subunit/tests/test_test_results.py
index ff74b9a..44f95b3 100644
--- a/lib/subunit/python/subunit/tests/test_test_results.py
+++ b/lib/subunit/python/subunit/tests/test_test_results.py
@@ -564,9 +564,3 @@ class TestCsvResult(testtools.TestCase):
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
diff --git a/lib/subunit/python/subunit/v2.py b/lib/subunit/python/subunit/v2.py
new file mode 100644
index 0000000..057f65c
--- /dev/null
+++ b/lib/subunit/python/subunit/v2.py
@@ -0,0 +1,495 @@
+#
+# subunit: extensions to Python unittest to get test results from subprocesses.
+# Copyright (C) 2013 Robert Collins <robertc at 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 codecs
+utf_8_decode = codecs.utf_8_decode
+import datetime
+from io import UnsupportedOperation
+import os
+import select
+import struct
+import zlib
+
+from extras import safe_hasattr, try_imports
+builtins = try_imports(['__builtin__', 'builtins'])
+
+import subunit
+import subunit.iso8601 as iso8601
+
+__all__ = [
+ 'ByteStreamToStreamResult',
+ 'StreamResultToBytes',
+ ]
+
+SIGNATURE = b'\xb3'
+FMT_8 = '>B'
+FMT_16 = '>H'
+FMT_24 = '>HB'
+FMT_32 = '>I'
+FMT_TIMESTAMP = '>II'
+FLAG_TEST_ID = 0x0800
+FLAG_ROUTE_CODE = 0x0400
+FLAG_TIMESTAMP = 0x0200
+FLAG_RUNNABLE = 0x0100
+FLAG_TAGS = 0x0080
+FLAG_MIME_TYPE = 0x0020
+FLAG_EOF = 0x0010
+FLAG_FILE_CONTENT = 0x0040
+EPOCH = datetime.datetime.utcfromtimestamp(0).replace(tzinfo=iso8601.Utc())
+NUL_ELEMENT = b'\0'[0]
+# Contains True for types for which 'nul in thing' falsely returns false.
+_nul_test_broken = {}
+
+
+def has_nul(buffer_or_bytes):
+ """Return True if a null byte is present in buffer_or_bytes."""
+ # Simple "if NUL_ELEMENT in utf8_bytes:" fails on Python 3.1 and 3.2 with
+ # memoryviews. See https://bugs.launchpad.net/subunit/+bug/1216246
+ buffer_type = type(buffer_or_bytes)
+ broken = _nul_test_broken.get(buffer_type)
+ if broken is None:
+ reference = buffer_type(b'\0')
+ broken = not NUL_ELEMENT in reference
+ _nul_test_broken[buffer_type] = broken
+ if broken:
+ return b'\0' in buffer_or_bytes
+ else:
+ return NUL_ELEMENT in buffer_or_bytes
+
+
+class ParseError(Exception):
+ """Used to pass error messages within the parser."""
+
+
+class StreamResultToBytes(object):
+ """Convert StreamResult API calls to bytes.
+
+ The StreamResult API is defined by testtools.StreamResult.
+ """
+
+ status_mask = {
+ None: 0,
+ 'exists': 0x1,
+ 'inprogress': 0x2,
+ 'success': 0x3,
+ 'uxsuccess': 0x4,
+ 'skip': 0x5,
+ 'fail': 0x6,
+ 'xfail': 0x7,
+ }
+
+ zero_b = b'\0'[0]
+
+ def __init__(self, output_stream):
+ """Create a StreamResultToBytes with output written to output_stream.
+
+ :param output_stream: A file-like object. Must support write(bytes)
+ and flush() methods. Flush will be called after each write.
+ The stream will be passed through subunit.make_stream_binary,
+ to handle regular cases such as stdout.
+ """
+ self.output_stream = subunit.make_stream_binary(output_stream)
+
+ def startTestRun(self):
+ pass
+
+ def stopTestRun(self):
+ pass
+
+ def status(self, test_id=None, test_status=None, test_tags=None,
+ runnable=True, file_name=None, file_bytes=None, eof=False,
+ mime_type=None, route_code=None, timestamp=None):
+ self._write_packet(test_id=test_id, test_status=test_status,
+ test_tags=test_tags, runnable=runnable, file_name=file_name,
+ file_bytes=file_bytes, eof=eof, mime_type=mime_type,
+ route_code=route_code, timestamp=timestamp)
+
+ def _write_utf8(self, a_string, packet):
+ utf8 = a_string.encode('utf-8')
+ self._write_number(len(utf8), packet)
+ packet.append(utf8)
+
+ def _write_len16(self, length, packet):
+ assert length < 65536
+ packet.append(struct.pack(FMT_16, length))
+
+ def _write_number(self, value, packet):
+ packet.extend(self._encode_number(value))
+
+ def _encode_number(self, value):
+ assert value >= 0
+ if value < 64:
+ return [struct.pack(FMT_8, value)]
+ elif value < 16384:
+ value = value | 0x4000
+ return [struct.pack(FMT_16, value)]
+ elif value < 4194304:
+ value = value | 0x800000
+ return [struct.pack(FMT_16, value >> 8),
+ struct.pack(FMT_8, value & 0xff)]
+ elif value < 1073741824:
+ value = value | 0xc0000000
+ return [struct.pack(FMT_32, value)]
+ else:
+ raise ValueError('value too large to encode: %r' % (value,))
+
+ def _write_packet(self, test_id=None, test_status=None, test_tags=None,
+ runnable=True, file_name=None, file_bytes=None, eof=False,
+ mime_type=None, route_code=None, timestamp=None):
+ packet = [SIGNATURE]
+ packet.append(b'FF') # placeholder for flags
+ # placeholder for length, but see below as length is variable.
+ packet.append(b'')
+ flags = 0x2000 # Version 0x2
+ if timestamp is not None:
+ flags = flags | FLAG_TIMESTAMP
+ since_epoch = timestamp - EPOCH
+ nanoseconds = since_epoch.microseconds * 1000
+ seconds = (since_epoch.seconds + since_epoch.days * 24 * 3600)
+ packet.append(struct.pack(FMT_32, seconds))
+ self._write_number(nanoseconds, packet)
+ if test_id is not None:
+ flags = flags | FLAG_TEST_ID
+ self._write_utf8(test_id, packet)
+ if test_tags:
+ flags = flags | FLAG_TAGS
+ self._write_number(len(test_tags), packet)
+ for tag in test_tags:
+ self._write_utf8(tag, packet)
+ if runnable:
+ flags = flags | FLAG_RUNNABLE
+ if mime_type:
+ flags = flags | FLAG_MIME_TYPE
+ self._write_utf8(mime_type, packet)
+ if file_name is not None:
+ flags = flags | FLAG_FILE_CONTENT
+ self._write_utf8(file_name, packet)
+ self._write_number(len(file_bytes), packet)
+ packet.append(file_bytes)
+ if eof:
+ flags = flags | FLAG_EOF
+ if route_code is not None:
+ flags = flags | FLAG_ROUTE_CODE
+ self._write_utf8(route_code, packet)
+ # 0x0008 - not used in v2.
+ flags = flags | self.status_mask[test_status]
+ packet[1] = struct.pack(FMT_16, flags)
+ base_length = sum(map(len, packet)) + 4
+ if base_length <= 62:
+ # one byte to encode length, 62+1 = 63
+ length_length = 1
+ elif base_length <= 16381:
+ # two bytes to encode length, 16381+2 = 16383
+ length_length = 2
+ elif base_length <= 4194300:
+ # three bytes to encode length, 419430+3=4194303
+ length_length = 3
+ else:
+ # Longer than policy:
+ # TODO: chunk the packet automatically?
+ # - strip all but file data
+ # - do 4M chunks of that till done
+ # - include original data in final chunk.
+ raise ValueError("Length too long: %r" % base_length)
+ packet[2:3] = self._encode_number(base_length + length_length)
+ # We could either do a partial application of crc32 over each chunk
+ # or a single join to a temp variable then a final join
+ # or two writes (that python might then split).
+ # For now, simplest code: join, crc32, join, output
+ content = b''.join(packet)
+ self.output_stream.write(content + struct.pack(
+ FMT_32, zlib.crc32(content) & 0xffffffff))
+ self.output_stream.flush()
+
+
+class ByteStreamToStreamResult(object):
+ """Parse a subunit byte stream.
+
+ Mixed streams that contain non-subunit content is supported when a
+ non_subunit_name is passed to the contructor. The default is to raise an
+ error containing the non-subunit byte after it has been read from the
+ stream.
+
+ Typical use:
+
+ >>> case = ByteStreamToStreamResult(sys.stdin.buffer)
+ >>> result = StreamResult()
+ >>> result.startTestRun()
+ >>> case.run(result)
+ >>> result.stopTestRun()
+ """
+
+ status_lookup = {
+ 0x0: None,
+ 0x1: 'exists',
+ 0x2: 'inprogress',
+ 0x3: 'success',
+ 0x4: 'uxsuccess',
+ 0x5: 'skip',
+ 0x6: 'fail',
+ 0x7: 'xfail',
+ }
+
+ def __init__(self, source, non_subunit_name=None):
+ """Create a ByteStreamToStreamResult.
+
+ :param source: A file like object to read bytes from. Must support
+ read(<count>) and return bytes. The file is not closed by
+ ByteStreamToStreamResult. subunit.make_stream_binary() is
+ called on the stream to get it into bytes mode.
+ :param non_subunit_name: If set to non-None, non subunit content
+ encountered in the stream will be converted into file packets
+ labelled with this name.
+ """
+ self.non_subunit_name = non_subunit_name
+ self.source = subunit.make_stream_binary(source)
+ self.codec = codecs.lookup('utf8').incrementaldecoder()
+
+ def run(self, result):
+ """Parse source and emit events to result.
+
+ This is a blocking call: it will run until EOF is detected on source.
+ """
+ self.codec.reset()
+ mid_character = False
+ while True:
+ # We're in blocking mode; read one char
+ content = self.source.read(1)
+ if not content:
+ # EOF
+ return
+ if not mid_character and content[0] == SIGNATURE[0]:
+ self._parse_packet(result)
+ continue
+ if self.non_subunit_name is None:
+ raise Exception("Non subunit content", content)
+ try:
+ if self.codec.decode(content):
+ # End of a character
+ mid_character = False
+ else:
+ mid_character = True
+ except UnicodeDecodeError:
+ # Bad unicode, not our concern.
+ mid_character = False
+ # Aggregate all content that is not subunit until either
+ # 1MiB is accumulated or 50ms has passed with no input.
+ # Both are arbitrary amounts intended to give a simple
+ # balance between efficiency (avoiding death by a thousand
+ # one-byte packets), buffering (avoiding overlarge state
+ # being hidden on intermediary nodes) and interactivity
+ # (when driving a debugger, slow response to typing is
+ # annoying).
+ buffered = [content]
+ while len(buffered[-1]):
+ try:
+ self.source.fileno()
+ except:
+ # Won't be able to select, fallback to
+ # one-byte-at-a-time.
+ break
+ # Note: this has a very low timeout because with stdin, the
+ # BufferedIO layer typically has all the content available
+ # from the stream when e.g. pdb is dropped into, leading to
+ # select always timing out when in fact we could have read
+ # (from the buffer layer) - we typically fail to aggregate
+ # any content on 3.x Pythons.
+ readable = select.select([self.source], [], [], 0.000001)[0]
+ if readable:
+ content = self.source.read(1)
+ if not len(content):
+ # EOF - break and emit buffered.
+ break
+ if not mid_character and content[0] == SIGNATURE[0]:
+ # New packet, break, emit buffered, then parse.
+ break
+ buffered.append(content)
+ # Feed into the codec.
+ try:
+ if self.codec.decode(content):
+ # End of a character
+ mid_character = False
+ else:
+ mid_character = True
+ except UnicodeDecodeError:
+ # Bad unicode, not our concern.
+ mid_character = False
+ if not readable or len(buffered) >= 1048576:
+ # timeout or too much data, emit what we have.
+ break
+ result.status(
+ file_name=self.non_subunit_name,
+ file_bytes=b''.join(buffered))
+ if mid_character or not len(content) or content[0] != SIGNATURE[0]:
+ continue
+ # Otherwise, parse a data packet.
+ self._parse_packet(result)
+
+ def _parse_packet(self, result):
+ try:
+ packet = [SIGNATURE]
+ self._parse(packet, result)
+ except ParseError as error:
+ result.status(test_id="subunit.parser", eof=True,
+ file_name="Packet data", file_bytes=b''.join(packet),
+ mime_type="application/octet-stream")
+ result.status(test_id="subunit.parser", test_status='fail',
+ eof=True, file_name="Parser Error",
+ file_bytes=(error.args[0]).encode('utf8'),
+ mime_type="text/plain;charset=utf8")
+
+ def _to_bytes(self, data, pos, length):
+ """Return a slice of data from pos for length as bytes."""
+ # memoryview in 2.7.3 and 3.2 isn't directly usable with struct :(.
+ # see https://bugs.launchpad.net/subunit/+bug/1216163
+ result = data[pos:pos+length]
+ if type(result) is not bytes:
+ return result.tobytes()
+ return result
+
+ def _parse_varint(self, data, pos, max_3_bytes=False):
+ # because the only incremental IO we do is at the start, and the 32 bit
+ # CRC means we can always safely read enough to cover any varint, we
+ # can be sure that there should be enough data - and if not it is an
+ # error not a normal situation.
+ data_0 = struct.unpack(FMT_8, self._to_bytes(data, pos, 1))[0]
+ typeenum = data_0 & 0xc0
+ value_0 = data_0 & 0x3f
+ if typeenum == 0x00:
+ return value_0, 1
+ elif typeenum == 0x40:
+ data_1 = struct.unpack(FMT_8, self._to_bytes(data, pos+1, 1))[0]
+ return (value_0 << 8) | data_1, 2
+ elif typeenum == 0x80:
+ data_1 = struct.unpack(FMT_16, self._to_bytes(data, pos+1, 2))[0]
+ return (value_0 << 16) | data_1, 3
+ else:
+ if max_3_bytes:
+ raise ParseError('3 byte maximum given but 4 byte value found.')
+ data_1, data_2 = struct.unpack(FMT_24, self._to_bytes(data, pos+1, 3))
+ result = (value_0 << 24) | data_1 << 8 | data_2
+ return result, 4
+
+ def _parse(self, packet, result):
+ # 2 bytes flags, at most 3 bytes length.
+ packet.append(self.source.read(5))
+ flags = struct.unpack(FMT_16, packet[-1][:2])[0]
+ length, consumed = self._parse_varint(
+ packet[-1], 2, max_3_bytes=True)
+ remainder = self.source.read(length - 6)
+ if len(remainder) != length - 6:
+ raise ParseError(
+ 'Short read - got %d bytes, wanted %d bytes' % (
+ len(remainder), length - 6))
+ if consumed != 3:
+ # Avoid having to parse torn values
+ packet[-1] += remainder
+ pos = 2 + consumed
+ else:
+ # Avoid copying potentially lots of data.
+ packet.append(remainder)
+ pos = 0
+ crc = zlib.crc32(packet[0])
+ for fragment in packet[1:-1]:
+ crc = zlib.crc32(fragment, crc)
+ crc = zlib.crc32(packet[-1][:-4], crc) & 0xffffffff
+ packet_crc = struct.unpack(FMT_32, packet[-1][-4:])[0]
+ if crc != packet_crc:
+ # Bad CRC, report it and stop parsing the packet.
+ raise ParseError(
+ 'Bad checksum - calculated (0x%x), stored (0x%x)'
+ % (crc, packet_crc))
+ if safe_hasattr(builtins, 'memoryview'):
+ body = memoryview(packet[-1])
+ else:
+ body = packet[-1]
+ # Discard CRC-32
+ body = body[:-4]
+ # One packet could have both file and status data; the Python API
+ # presents these separately (perhaps it shouldn't?)
+ if flags & FLAG_TIMESTAMP:
+ seconds = struct.unpack(FMT_32, self._to_bytes(body, pos, 4))[0]
+ nanoseconds, consumed = self._parse_varint(body, pos+4)
+ pos = pos + 4 + consumed
+ timestamp = EPOCH + datetime.timedelta(
+ seconds=seconds, microseconds=nanoseconds/1000)
+ else:
+ timestamp = None
+ if flags & FLAG_TEST_ID:
+ test_id, pos = self._read_utf8(body, pos)
+ else:
+ test_id = None
+ if flags & FLAG_TAGS:
+ tag_count, consumed = self._parse_varint(body, pos)
+ pos += consumed
+ test_tags = set()
+ for _ in range(tag_count):
+ tag, pos = self._read_utf8(body, pos)
+ test_tags.add(tag)
+ else:
+ test_tags = None
+ if flags & FLAG_MIME_TYPE:
+ mime_type, pos = self._read_utf8(body, pos)
+ else:
+ mime_type = None
+ if flags & FLAG_FILE_CONTENT:
+ file_name, pos = self._read_utf8(body, pos)
+ content_length, consumed = self._parse_varint(body, pos)
+ pos += consumed
+ file_bytes = self._to_bytes(body, pos, content_length)
+ if len(file_bytes) != content_length:
+ raise ParseError('File content extends past end of packet: '
+ 'claimed %d bytes, %d available' % (
+ content_length, len(file_bytes)))
+ pos += content_length
+ else:
+ file_name = None
+ file_bytes = None
+ if flags & FLAG_ROUTE_CODE:
+ route_code, pos = self._read_utf8(body, pos)
+ else:
+ route_code = None
+ runnable = bool(flags & FLAG_RUNNABLE)
+ eof = bool(flags & FLAG_EOF)
+ test_status = self.status_lookup[flags & 0x0007]
+ result.status(test_id=test_id, test_status=test_status,
+ test_tags=test_tags, runnable=runnable, mime_type=mime_type,
+ eof=eof, file_name=file_name, file_bytes=file_bytes,
+ route_code=route_code, timestamp=timestamp)
+ __call__ = run
+
+ def _read_utf8(self, buf, pos):
+ length, consumed = self._parse_varint(buf, pos)
+ pos += consumed
+ utf8_bytes = buf[pos:pos+length]
+ if length != len(utf8_bytes):
+ raise ParseError(
+ 'UTF8 string at offset %d extends past end of packet: '
+ 'claimed %d bytes, %d available' % (pos - 2, length,
+ len(utf8_bytes)))
+ if has_nul(utf8_bytes):
+ raise ParseError('UTF8 string at offset %d contains NUL byte' % (
+ pos-2,))
+ try:
+ utf8, decoded_bytes = utf_8_decode(utf8_bytes)
+ if decoded_bytes != length:
+ raise ParseError("Invalid (partially decodable) string at "
+ "offset %d, %d undecoded bytes" % (
+ pos-2, length - decoded_bytes))
+ return utf8, length+pos
+ except UnicodeDecodeError:
+ raise ParseError('UTF8 string at offset %d is not UTF8' % (pos-2,))
+
diff --git a/lib/subunit/runtests.py b/lib/subunit/runtests.py
deleted file mode 100755
index 8ecc6cd..0000000
--- a/lib/subunit/runtests.py
+++ /dev/null
@@ -1,138 +0,0 @@
-#!/usr/bin/env python
-# -*- Mode: python -*-
-#
-# Copyright (C) 2004 Canonical.com
-# Author: Robert Collins <robert.collins at 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 unittest
-from subunit.tests.TestUtil import TestVisitor, TestSuite
-import subunit
-import sys
-import os
-import shutil
-import logging
-
-class ParameterisableTextTestRunner(unittest.TextTestRunner):
- """I am a TextTestRunner whose result class is
- parameterisable without further subclassing"""
- def __init__(self, **args):
- unittest.TextTestRunner.__init__(self, **args)
- self._resultFactory=None
- def resultFactory(self, *args):
- """set or retrieve the result factory"""
- if args:
- self._resultFactory=args[0]
- return self
- if self._resultFactory is None:
- self._resultFactory=unittest._TextTestResult
- return self._resultFactory
-
- def _makeResult(self):
- return self.resultFactory()(self.stream, self.descriptions, self.verbosity)
-
-
-class EarlyStoppingTextTestResult(unittest._TextTestResult):
- """I am a TextTestResult that can optionally stop at the first failure
- or error"""
-
- def addError(self, test, err):
- unittest._TextTestResult.addError(self, test, err)
- if self.stopOnError():
- self.stop()
-
- def addFailure(self, test, err):
- unittest._TextTestResult.addError(self, test, err)
- if self.stopOnFailure():
- self.stop()
-
- def stopOnError(self, *args):
- """should this result indicate an abort when an error occurs?
- TODO parameterise this"""
- return True
-
- def stopOnFailure(self, *args):
- """should this result indicate an abort when a failure error occurs?
- TODO parameterise this"""
- return True
-
-
-def earlyStopFactory(*args, **kwargs):
- """return a an early stopping text test result"""
- result=EarlyStoppingTextTestResult(*args, **kwargs)
- return result
-
-
-class ShellTests(subunit.ExecTestCase):
-
- def test_sourcing(self):
- """./shell/tests/test_source_library.sh"""
-
- def test_functions(self):
- """./shell/tests/test_function_output.sh"""
-
-
-def test_suite():
- result = TestSuite()
- result.addTest(subunit.test_suite())
- result.addTest(ShellTests('test_sourcing'))
- result.addTest(ShellTests('test_functions'))
- return result
-
-
-class filteringVisitor(TestVisitor):
- """I accrue all the testCases I visit that pass a regexp filter on id
- into my suite
- """
-
- def __init__(self, filter):
- import re
- TestVisitor.__init__(self)
- self._suite=None
- self.filter=re.compile(filter)
-
- def suite(self):
- """answer the suite we are building"""
- if self._suite is None:
- self._suite=TestSuite()
- return self._suite
-
- def visitCase(self, aCase):
- if self.filter.match(aCase.id()):
- self.suite().addTest(aCase)
-
-
-def main(argv):
- """To parameterise what tests are run, run this script like so:
- python test_all.py REGEX
- i.e.
- python test_all.py .*Protocol.*
- to run all tests with Protocol in their id."""
- if len(argv) > 1:
- pattern = argv[1]
- else:
- pattern = ".*"
- visitor = filteringVisitor(pattern)
- test_suite().visit(visitor)
- runner = ParameterisableTextTestRunner(verbosity=2)
- runner.resultFactory(unittest._TextTestResult)
- if not runner.run(visitor.suite()).wasSuccessful():
- return 1
- return 0
-
-if __name__ == '__main__':
- sys.exit(main(sys.argv))
diff --git a/lib/subunit/setup.py b/lib/subunit/setup.py
index 1a0b192..d42d3d7 100755
--- a/lib/subunit/setup.py
+++ b/lib/subunit/setup.py
@@ -1,4 +1,5 @@
#!/usr/bin/env python
+import os.path
try:
# If the user has setuptools / distribute installed, use it
from setuptools import setup
@@ -9,8 +10,12 @@ except ImportError:
else:
extra = {
'install_requires': [
- 'testtools>=0.9.23',
- ]
+ 'extras',
+ 'testtools>=0.9.34',
+ ],
+ 'tests_require': [
+ 'testscenarios',
+ ],
}
@@ -31,6 +36,9 @@ VERSION = (
or "0.0")
+relpath = os.path.dirname(__file__)
+if relpath:
+ os.chdir(relpath)
setup(
name='python-subunit',
version=VERSION,
@@ -49,14 +57,18 @@ setup(
packages=['subunit', 'subunit.tests'],
package_dir={'subunit': 'python/subunit'},
scripts = [
- 'filters/subunit2gtk',
- 'filters/subunit2junitxml',
- 'filters/subunit2pyunit',
+ 'filters/subunit-1to2',
+ 'filters/subunit-2to1',
'filters/subunit-filter',
'filters/subunit-ls',
'filters/subunit-notify',
+ 'filters/subunit-output',
'filters/subunit-stats',
'filters/subunit-tags',
+ 'filters/subunit2csv',
+ 'filters/subunit2gtk',
+ 'filters/subunit2junitxml',
+ 'filters/subunit2pyunit',
'filters/tap2subunit',
],
**extra
--
2.1.1
More information about the samba-technical
mailing list