From df2c69b59978bd5ba90db298bbae08bbd4e53595 Mon Sep 17 00:00:00 2001 From: Matthew Treinish Date: Tue, 8 Aug 2017 17:15:18 -0400 Subject: Check for ValueError in write only check in _unwrap_text() In the _unwrap_text() function it checks by trying to read or write to the stream to see if it's wrapped or not and return the buffer. The check there first tries to read and if an exception of the proper type is raised it will assume it was opened in write mode and proceed to check for a buffer by writing to the stram. However this check is missing an exception to check for. If a FileIO object is passed in on Python 2.7 (which can be the case sometimes for sys.stdout, although I haven't figured out under which circumstances) then a ValueError will be raised. [1] To enable using _unwrap_text() when a FileIO object is used this commit adds a ValueError to the list of exception types to handle this case. [1] https://github.com/python/cpython/blob/2.7/Modules/_io/fileio.c#L431 --- python/subunit/__init__.py | 4 ++- python/subunit/tests/test_test_protocol.py | 41 ++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) (limited to 'python') diff --git a/python/subunit/__init__.py b/python/subunit/__init__.py index 69acef5..dd3e951 100644 --- a/python/subunit/__init__.py +++ b/python/subunit/__init__.py @@ -1287,15 +1287,17 @@ def _make_binary_on_windows(fileno): def _unwrap_text(stream): """Unwrap stream if it is a text stream to get the original buffer.""" + exceptions = (_UnsupportedOperation, IOError) if sys.version_info > (3, 0): unicode_type = str else: unicode_type = unicode + exceptions += (ValueError,) try: # Read streams if type(stream.read(0)) is unicode_type: return stream.buffer - except (_UnsupportedOperation, IOError): + except exceptions: # Cannot read from the stream: try via writes try: stream.write(_b('')) diff --git a/python/subunit/tests/test_test_protocol.py b/python/subunit/tests/test_test_protocol.py index f3cd0e3..7427b12 100644 --- a/python/subunit/tests/test_test_protocol.py +++ b/python/subunit/tests/test_test_protocol.py @@ -15,8 +15,11 @@ # import datetime +import io import unittest2 as unittest import os +import sys +import tempfile from testtools import PlaceHolder, skipIf, TestCase, TestResult from testtools.compat import _b, _u, BytesIO @@ -52,6 +55,44 @@ def details_to_str(details): return TestResult()._err_details_to_string(None, details=details) +class TestHelpers(TestCase): + def test__unwrap_text_file_read_mode(self): + fd, file_path = tempfile.mkstemp() + self.addCleanup(os.remove, file_path) + fake_file = os.fdopen(fd, 'r') + if sys.version_info > (3, 0): + self.assertEqual(fake_file.buffer, + subunit._unwrap_text(fake_file)) + else: + self.assertEqual(fake_file, subunit._unwrap_text(fake_file)) + + def test__unwrap_text_file_write_mode(self): + fd, file_path = tempfile.mkstemp() + self.addCleanup(os.remove, file_path) + fake_file = os.fdopen(fd, 'w') + if sys.version_info > (3, 0): + self.assertEqual(fake_file.buffer, + subunit._unwrap_text(fake_file)) + else: + self.assertEqual(fake_file, subunit._unwrap_text(fake_file)) + + def test__unwrap_text_fileIO_read_mode(self): + fd, file_path = tempfile.mkstemp() + self.addCleanup(os.remove, file_path) + fake_file = io.FileIO(file_path, 'r') + self.assertEqual(fake_file, subunit._unwrap_text(fake_file)) + + def test__unwrap_text_fileIO_write_mode(self): + fd, file_path = tempfile.mkstemp() + self.addCleanup(os.remove, file_path) + fake_file = io.FileIO(file_path, 'w') + self.assertEqual(fake_file, subunit._unwrap_text(fake_file)) + + def test__unwrap_text_BytesIO(self): + fake_stream = io.BytesIO() + self.assertEqual(fake_stream, subunit._unwrap_text(fake_stream)) + + class TestTestImports(unittest.TestCase): def test_imports(self): -- cgit v1.2.1