import subprocess
import sys
import unittest
+if sys.version_info > (3, 0):
+ from io import UnsupportedOperation as _UnsupportedOperation
+else:
+ _UnsupportedOperation = AttributeError
+
from testtools import content, content_type, ExtendedToOriginalDecorator
from testtools.content import TracebackContent
class DiscardStream(object):
"""A filelike object which discards what is written to it."""
+ def fileno(self):
+ raise _UnsupportedOperation()
+
def write(self, bytes):
pass
+ def read(self, len=0):
+ return _b('')
+
class _ParserState(object):
"""State for the subunit parser."""
def __init__(self, stream):
testresult.TestResult.__init__(self)
+ stream = _make_stream_binary(stream)
self._stream = stream
- _make_stream_binary(stream)
self._progress_fmt = _b("progress: ")
self._bytes_eol = _b("\n")
self._progress_plus = _b("+")
:param forward: A stream to pass subunit input on to. If not supplied
subunit input is not forwarded.
"""
+ stream = _make_stream_binary(stream)
self._stream = stream
- _make_stream_binary(stream)
self._passthrough = passthrough
if forward is not None:
- _make_stream_binary(forward)
+ forward = _make_stream_binary(forward)
self._forward = forward
def __call__(self, result=None):
return stream
-if sys.version_info > (3, 0):
- from io import UnsupportedOperation as _NoFilenoError
-else:
- _NoFilenoError = AttributeError
-
def read_test_list(path):
"""Read a list of test ids from a file on disk.
def _make_stream_binary(stream):
- """Ensure that a stream will be binary safe. See _make_binary_on_windows."""
+ """Ensure that a stream will be binary safe. See _make_binary_on_windows.
+
+ :return: A binary version of the same stream (some streams cannot be
+ 'fixed' but can be unwrapped).
+ """
try:
fileno = stream.fileno()
- except _NoFilenoError:
- return
- _make_binary_on_windows(fileno)
+ except _UnsupportedOperation:
+ pass
+ else:
+ _make_binary_on_windows(fileno)
+ return _unwrap_text(stream)
def _make_binary_on_windows(fileno):
"""Win32 mangles \r\n to \n and that breaks streams. See bug lp:505078."""
if sys.platform == "win32":
import msvcrt
msvcrt.setmode(fileno, os.O_BINARY)
+
+
+def _unwrap_text(stream):
+ """Unwrap stream if it is a text stream to get the original buffer."""
+ if sys.version_info > (3, 0):
+ try:
+ # Read streams
+ if type(stream.read(0)) is str:
+ return stream.buffer
+ except _UnsupportedOperation:
+ # Cannot read from the stream: try via writes
+ try:
+ stream.write(_b(''))
+ except TypeError:
+ return stream.buffer
+ return stream
self.run_tests(result_filter)
tests_included = [
event[1] for event in result._events if event[0] == 'startTest']
- tests_expected = map(
+ tests_expected = list(map(
subunit.RemotedTestCase,
- ['passed', 'error', 'skipped', 'todo'])
+ ['passed', 'error', 'skipped', 'todo']))
self.assertEquals(tests_expected, tests_included)
def test_tags_tracked_correctly(self):
result = ExtendedTestResult()
result_filter = TestResultFilter(
result, filter_success=False, filter_predicate=tag_filter)
- input_stream = (
+ input_stream = _b(
"test: foo\n"
"tags: a\n"
"successful: foo\n"
return result._events
def test_default(self):
- output = self.run_command([], (
+ output = self.run_command([], _b(
"test: foo\n"
"skip: foo\n"
))
events)
def test_tags(self):
- output = self.run_command(['-s', '--with-tag', 'a'], (
+ output = self.run_command(['-s', '--with-tag', 'a'], _b(
"tags: a\n"
"test: foo\n"
"success: foo\n"