Merge pull request #1097 from anthrotype/py23-redirect-stream
[py23] Add backports for redirect_stdout/stderr context managers
This commit is contained in:
commit
bb60cc4e2b
@ -496,6 +496,53 @@ except ImportError:
|
|||||||
return self.__dict__ == other.__dict__
|
return self.__dict__ == other.__dict__
|
||||||
|
|
||||||
|
|
||||||
|
if sys.version_info[:2] > (3, 4):
|
||||||
|
from contextlib import redirect_stdout, redirect_stderr
|
||||||
|
else:
|
||||||
|
# `redirect_stdout` was added with python3.4, while `redirect_stderr`
|
||||||
|
# with python3.5. For simplicity, I redefine both for any versions
|
||||||
|
# less than or equal to 3.4.
|
||||||
|
# The code below is copied from:
|
||||||
|
# https://github.com/python/cpython/blob/57161aa/Lib/contextlib.py
|
||||||
|
|
||||||
|
class _RedirectStream(object):
|
||||||
|
|
||||||
|
_stream = None
|
||||||
|
|
||||||
|
def __init__(self, new_target):
|
||||||
|
self._new_target = new_target
|
||||||
|
# We use a list of old targets to make this CM re-entrant
|
||||||
|
self._old_targets = []
|
||||||
|
|
||||||
|
def __enter__(self):
|
||||||
|
self._old_targets.append(getattr(sys, self._stream))
|
||||||
|
setattr(sys, self._stream, self._new_target)
|
||||||
|
return self._new_target
|
||||||
|
|
||||||
|
def __exit__(self, exctype, excinst, exctb):
|
||||||
|
setattr(sys, self._stream, self._old_targets.pop())
|
||||||
|
|
||||||
|
|
||||||
|
class redirect_stdout(_RedirectStream):
|
||||||
|
"""Context manager for temporarily redirecting stdout to another file.
|
||||||
|
# How to send help() to stderr
|
||||||
|
with redirect_stdout(sys.stderr):
|
||||||
|
help(dir)
|
||||||
|
# How to write help() to a file
|
||||||
|
with open('help.txt', 'w') as f:
|
||||||
|
with redirect_stdout(f):
|
||||||
|
help(pow)
|
||||||
|
"""
|
||||||
|
|
||||||
|
_stream = "stdout"
|
||||||
|
|
||||||
|
|
||||||
|
class redirect_stderr(_RedirectStream):
|
||||||
|
"""Context manager for temporarily redirecting stderr to another file."""
|
||||||
|
|
||||||
|
_stream = "stderr"
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
import doctest, sys
|
import doctest, sys
|
||||||
sys.exit(doctest.testmod().failed)
|
sys.exit(doctest.testmod().failed)
|
||||||
|
@ -8,7 +8,8 @@ import sys
|
|||||||
import os
|
import os
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
from fontTools.misc.py23 import round2, round3, isclose
|
from fontTools.misc.py23 import (
|
||||||
|
round2, round3, isclose, redirect_stdout, redirect_stderr)
|
||||||
|
|
||||||
|
|
||||||
PIPE_SCRIPT = """\
|
PIPE_SCRIPT = """\
|
||||||
@ -418,5 +419,67 @@ class NarrowUnicodeBuildTest(unittest.TestCase):
|
|||||||
self.assertEqual(byteord(u'\U0010FFFF'), 1114111)
|
self.assertEqual(byteord(u'\U0010FFFF'), 1114111)
|
||||||
|
|
||||||
|
|
||||||
|
class TestRedirectStream:
|
||||||
|
|
||||||
|
redirect_stream = None
|
||||||
|
orig_stream = None
|
||||||
|
|
||||||
|
def test_no_redirect_in_init(self):
|
||||||
|
orig_stdout = getattr(sys, self.orig_stream)
|
||||||
|
self.redirect_stream(None)
|
||||||
|
self.assertIs(getattr(sys, self.orig_stream), orig_stdout)
|
||||||
|
|
||||||
|
def test_redirect_to_string_io(self):
|
||||||
|
f = StringIO()
|
||||||
|
msg = "Consider an API like help(), which prints directly to stdout"
|
||||||
|
orig_stdout = getattr(sys, self.orig_stream)
|
||||||
|
with self.redirect_stream(f):
|
||||||
|
print(msg, file=getattr(sys, self.orig_stream))
|
||||||
|
self.assertIs(getattr(sys, self.orig_stream), orig_stdout)
|
||||||
|
s = f.getvalue().strip()
|
||||||
|
self.assertEqual(s, msg)
|
||||||
|
|
||||||
|
def test_enter_result_is_target(self):
|
||||||
|
f = StringIO()
|
||||||
|
with self.redirect_stream(f) as enter_result:
|
||||||
|
self.assertIs(enter_result, f)
|
||||||
|
|
||||||
|
def test_cm_is_reusable(self):
|
||||||
|
f = StringIO()
|
||||||
|
write_to_f = self.redirect_stream(f)
|
||||||
|
orig_stdout = getattr(sys, self.orig_stream)
|
||||||
|
with write_to_f:
|
||||||
|
print("Hello", end=" ", file=getattr(sys, self.orig_stream))
|
||||||
|
with write_to_f:
|
||||||
|
print("World!", file=getattr(sys, self.orig_stream))
|
||||||
|
self.assertIs(getattr(sys, self.orig_stream), orig_stdout)
|
||||||
|
s = f.getvalue()
|
||||||
|
self.assertEqual(s, "Hello World!\n")
|
||||||
|
|
||||||
|
def test_cm_is_reentrant(self):
|
||||||
|
f = StringIO()
|
||||||
|
write_to_f = self.redirect_stream(f)
|
||||||
|
orig_stdout = getattr(sys, self.orig_stream)
|
||||||
|
with write_to_f:
|
||||||
|
print("Hello", end=" ", file=getattr(sys, self.orig_stream))
|
||||||
|
with write_to_f:
|
||||||
|
print("World!", file=getattr(sys, self.orig_stream))
|
||||||
|
self.assertIs(getattr(sys, self.orig_stream), orig_stdout)
|
||||||
|
s = f.getvalue()
|
||||||
|
self.assertEqual(s, "Hello World!\n")
|
||||||
|
|
||||||
|
|
||||||
|
class TestRedirectStdout(TestRedirectStream, unittest.TestCase):
|
||||||
|
|
||||||
|
redirect_stream = redirect_stdout
|
||||||
|
orig_stream = "stdout"
|
||||||
|
|
||||||
|
|
||||||
|
class TestRedirectStderr(TestRedirectStream, unittest.TestCase):
|
||||||
|
|
||||||
|
redirect_stream = redirect_stderr
|
||||||
|
orig_stream = "stderr"
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
sys.exit(unittest.main())
|
sys.exit(unittest.main())
|
||||||
|
Loading…
x
Reference in New Issue
Block a user