Fix test_multithreading on Python 3
* On Python 3, the printer doesn't encode Unicode to utf8 anymore, since print() expects a Unicode string. * Update unit tests for Python 3 since repr() doesn't escape non-ASCII characters in Unicode strings anymore: http://legacy.python.org/dev/peps/pep-3138/ Change-Id: I89471019d691a46651312d6a49964b719192148a
This commit is contained in:
parent
155053ea61
commit
ea498fb052
@ -16,6 +16,7 @@
|
||||
from __future__ import print_function
|
||||
|
||||
from itertools import chain
|
||||
import six
|
||||
import sys
|
||||
from time import sleep
|
||||
from six.moves.queue import Queue
|
||||
@ -184,7 +185,7 @@ class MultiThreadingManager(object):
|
||||
(defaults to ``sys.stdout``) and the :meth:`error` method will print to the
|
||||
supplied ``error_stream`` (defaults to ``sys.stderr``). Both of these
|
||||
printing methods will format the given string with any supplied ``*args``
|
||||
(a la printf) and encode the result to utf8 if necessary.
|
||||
(a la printf). On Python 2, Unicode messages are encoded to utf8.
|
||||
|
||||
The attribute :attr:`self.error_count` is incremented once per error
|
||||
message printed, so an application can tell if any worker threads
|
||||
@ -196,9 +197,11 @@ class MultiThreadingManager(object):
|
||||
def __init__(self, print_stream=sys.stdout, error_stream=sys.stderr):
|
||||
"""
|
||||
:param print_stream: The stream to which :meth:`print_msg` sends
|
||||
formatted messages, encoded to utf8 if necessary.
|
||||
formatted messages
|
||||
:param error_stream: The stream to which :meth:`error` sends formatted
|
||||
messages, encoded to utf8 if necessary.
|
||||
messages
|
||||
|
||||
On Python 2, Unicode messages are encoded to utf8.
|
||||
"""
|
||||
self.print_stream = print_stream
|
||||
self.printer = QueueFunctionManager(self._print, 1, self)
|
||||
@ -259,7 +262,7 @@ class MultiThreadingManager(object):
|
||||
def _print(self, item, stream=None):
|
||||
if stream is None:
|
||||
stream = self.print_stream
|
||||
if isinstance(item, unicode):
|
||||
if six.PY2 and isinstance(item, unicode):
|
||||
item = item.encode('utf8')
|
||||
print(item, file=stream)
|
||||
|
||||
|
@ -320,16 +320,22 @@ class TestMultiThreadingManager(ThreadTestCase):
|
||||
self.assertEqual(self.starting_thread_count, threading.active_count())
|
||||
|
||||
out_stream.seek(0)
|
||||
if six.PY3:
|
||||
over_the = "over the '\u062a\u062a'\n"
|
||||
else:
|
||||
over_the = "over the u'\\u062a\\u062a'\n"
|
||||
self.assertEqual([
|
||||
'one-argument\n',
|
||||
'one fish, 88 fish\n',
|
||||
'some\n', 'where\n', "over the u'\\u062a\\u062a'\n",
|
||||
'some\n', 'where\n', over_the,
|
||||
], list(out_stream.readlines()))
|
||||
|
||||
err_stream.seek(0)
|
||||
first_item = u'I have 99 problems, but a \u062A\u062A is not one\n'
|
||||
if six.PY2:
|
||||
first_item = first_item.encode('utf8')
|
||||
self.assertEqual([
|
||||
u'I have 99 problems, but a \u062A\u062A is not one\n'.encode(
|
||||
'utf8'),
|
||||
first_item,
|
||||
'one-error-argument\n',
|
||||
'Sometimes\n', '3.1% just\n', 'does not\n', 'work!\n',
|
||||
], list(err_stream.readlines()))
|
||||
|
Loading…
x
Reference in New Issue
Block a user