Add sphinx.util.logging.SafeEncodingWriter

sphinx/util/logging.py

@@ -373,6 +373,25 @@ class ColorizeFormatter(logging.Formatter):
             return message
 
 
+class SafeEncodingWriter(object):
+    """Stream writer which ignores UnicodeEncodeError silently"""
+    def __init__(self, stream):
+        self.stream = stream
+        self.encoding = getattr(stream, 'encoding', 'ascii') or 'ascii'
+
+    def write(self, data):
+        try:
+            self.stream.write(data)
+        except UnicodeEncodeError:
+            # stream accept only str, not bytes.  So, we encode and replace
+            # non-encodable characters, then decode them.
+            self.stream.write(data.encode(self.encoding, 'replace').decode(self.encoding))
+
+    def flush(self):
+        if hasattr(self.stream, 'flush'):
+            self.stream.flush()
+
+
 def setup(app, status, warning):
     # type: (Sphinx, IO, IO) -> None
     """Setup root logger for Sphinx"""
@@ -383,12 +402,12 @@ def setup(app, status, warning):
     for handler in logger.handlers[:]:
         logger.removeHandler(handler)
 
-    info_handler = NewLineStreamHandler(status)
+    info_handler = NewLineStreamHandler(SafeEncodingWriter(status))
     info_handler.addFilter(InfoFilter())
     info_handler.setLevel(VERBOSITY_MAP.get(app.verbosity))
     info_handler.setFormatter(ColorizeFormatter())
 
-    warning_handler = WarningStreamHandler(warning)
+    warning_handler = WarningStreamHandler(SafeEncodingWriter(warning))
     warning_handler.addFilter(WarningSuppressor(app))
     warning_handler.addFilter(WarningIsErrorFilter(app))
     warning_handler.addFilter(WarningLogRecordTranslator(app))

tests/test_application.py

@@ -8,7 +8,6 @@
     :copyright: Copyright 2007-2016 by the Sphinx team, see AUTHORS.
     :license: BSD, see LICENSE for details.
 """
-import codecs
 
 from docutils import nodes
 
@@ -68,22 +67,6 @@ def test_output(app, status, warning):
     assert app._warncount == old_count + 1
 
 
-@with_app()
-def test_output_with_unencodable_char(app, status, warning):
-
-    class StreamWriter(codecs.StreamWriter):
-        def write(self, object):
-            self.stream.write(object.encode('cp1252').decode('cp1252'))
-
-    app._status = StreamWriter(status)
-
-    # info with UnicodeEncodeError
-    status.truncate(0)
-    status.seek(0)
-    app.info(u"unicode \u206d...")
-    assert status.getvalue() == "unicode ?...\n"
-
-
 @with_app()
 def test_extensions(app, status, warning):
     app.setup_extension('shutil')

tests/test_util_logging.py

@@ -10,6 +10,7 @@
 """
 from __future__ import print_function
 
+import codecs
 from docutils import nodes
 
 from sphinx.errors import SphinxWarning
@@ -286,3 +287,18 @@ def test_logging_in_ParallelTasks(app, status, warning):
     tasks.join()
     assert 'message1' in status.getvalue()
     assert 'index.txt: WARNING: message2' in warning.getvalue()
+
+
+@with_app()
+def test_output_with_unencodable_char(app, status, warning):
+    class StreamWriter(codecs.StreamWriter):
+        def write(self, object):
+            self.stream.write(object.encode('cp1252').decode('cp1252'))
+
+    logging.setup(app, StreamWriter(status), warning)
+
+    # info with UnicodeEncodeError
+    status.truncate(0)
+    status.seek(0)
+    app.info(u"unicode \u206d...")
+    assert status.getvalue() == "unicode ?...\n"