Issue #13342: input() used to ignore sys.stdin's and sys.stdout's unicode

error handler in interactive mode (when calling into PyOS_Readline()).

3 files changed, 114 insertions, 42 deletions

diff --git a/Lib/test/test_builtin.py b/Lib/test/test_builtin.py
index 1469e36..40c1ab4 100644
--- a/Lib/test/test_builtin.py
+++ b/Lib/test/test_builtin.py
@@ -6,12 +6,17 @@ import sys
 import warnings
 import collections
 import io
+import os
 import ast
 import types
 import builtins
 import random
 from test.support import fcmp, TESTFN, unlink,  run_unittest, check_warnings
 from operator import neg
+try:
+    import pty
+except ImportError:
+    pty = None
 
 
 class Squares:
@@ -988,6 +993,71 @@ class BuiltinTest(unittest.TestCase):
             fp.close()
             unlink(TESTFN)
 
+    @unittest.skipUnless(pty, "the pty module isn't available")
+    def check_input_tty(self, prompt, terminal_input, stdio_encoding=None):
+        r, w = os.pipe()
+        try:
+            pid, fd = pty.fork()
+        except (OSError, AttributeError) as e:
+            os.close(r)
+            os.close(w)
+            self.skipTest("pty.fork() raised {}".format(e))
+        if pid == 0:
+            # Child
+            os.close(r)
+            # Check the error handlers are accounted for
+            if stdio_encoding:
+                sys.stdin = io.TextIOWrapper(sys.stdin.detach(),
+                                             encoding=stdio_encoding,
+                                             errors='surrogateescape')
+                sys.stdout = io.TextIOWrapper(sys.stdout.detach(),
+                                              encoding=stdio_encoding,
+                                              errors='replace')
+            with open(w, "w") as wpipe:
+                try:
+                    print("tty =", sys.stdin.isatty() and sys.stdout.isatty(), file=wpipe)
+                    print(ascii(input(prompt)), file=wpipe)
+                finally:
+                    print(";EOF", file=wpipe)
+            # We don't want to return to unittest...
+            os._exit(0)
+        # Parent
+        os.close(w)
+        os.write(fd, terminal_input + b"\r\n")
+        # Get results from the pipe
+        with open(r, "r") as rpipe:
+            lines = []
+            while True:
+                line = rpipe.readline().strip()
+                if line == ";EOF":
+                    break
+                lines.append(line)
+        # Check we did exercise the GNU readline path
+        self.assertIn(lines[0], {'tty = True', 'tty = False'})
+        if lines[0] != 'tty = True':
+            self.skipTest("standard IO in should have been a tty")
+        # Check the result was got and corresponds to the user's terminal input
+        self.assertEqual(len(lines), 2)
+        input_result = eval(lines[1])   # ascii() -> eval() roundtrip
+        if stdio_encoding:
+            expected = terminal_input.decode(stdio_encoding, 'surrogateescape')
+        else:
+            expected = terminal_input.decode(sys.stdin.encoding)  # what else?
+        self.assertEqual(input_result, expected)
+
+    def test_input_tty(self):
+        # Test input() functionality when wired to a tty (the code path
+        # is different and invokes GNU readline if available).
+        self.check_input_tty("prompt", b"quux")
+
+    def test_input_tty_non_ascii(self):
+        # Check stdin/stdout encoding is used when invoking GNU readline
+        self.check_input_tty("prompté", b"quux\xe9", "utf-8")
+
+    def test_input_tty_non_ascii_unicode_errors(self):
+        # Check stdin/stdout error handler is used when invoking GNU readline
+        self.check_input_tty("prompté", b"quux\xe9", "ascii")
+
     def test_repr(self):
         self.assertEqual(repr(''), '\'\'')
         self.assertEqual(repr(0), '0')
diff --git a/Misc/NEWS b/Misc/NEWS
index 720fef3..e550b52 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -10,6 +10,9 @@ What's New in Python 3.2.3?
 Core and Builtins
 -----------------
 
+- Issue #13342: input() used to ignore sys.stdin's and sys.stdout's unicode
+  error handler in interactive mode (when calling into PyOS_Readline()).
+
 - Issue #13340: Accept None as start and stop parameters for
   list.index() and tuple.index().
 
diff --git a/Python/bltinmodule.c b/Python/bltinmodule.c
index d05232c..9e37d21 100644
--- a/Python/bltinmodule.c
+++ b/Python/bltinmodule.c
@@ -1640,76 +1640,65 @@ builtin_input(PyObject *self, PyObject *args)
 
     /* If we're interactive, use (GNU) readline */
     if (tty) {
-        PyObject *po;
+        PyObject *po = NULL;
         char *prompt;
-        char *s;
-        PyObject *stdin_encoding;
-        char *stdin_encoding_str;
+        char *s = NULL;
+        PyObject *stdin_encoding = NULL, *stdin_errors = NULL;
+        PyObject *stdout_encoding = NULL, *stdout_errors = NULL;
+        char *stdin_encoding_str, *stdin_errors_str;
         PyObject *result;
         size_t len;
 
         stdin_encoding = PyObject_GetAttrString(fin, "encoding");
-        if (!stdin_encoding)
+        stdin_errors = PyObject_GetAttrString(fin, "errors");
+        if (!stdin_encoding || !stdin_errors)
             /* stdin is a text stream, so it must have an
                encoding. */
-            return NULL;
+            goto _readline_errors;
         stdin_encoding_str = _PyUnicode_AsString(stdin_encoding);
-        if (stdin_encoding_str  == NULL) {
-            Py_DECREF(stdin_encoding);
-            return NULL;
-        }
+        stdin_errors_str = _PyUnicode_AsString(stdin_errors);
+        if (!stdin_encoding_str || !stdin_errors_str)
+            goto _readline_errors;
         tmp = PyObject_CallMethod(fout, "flush", "");
         if (tmp == NULL)
             PyErr_Clear();
         else
             Py_DECREF(tmp);
         if (promptarg != NULL) {
+            /* We have a prompt, encode it as stdout would */
+            char *stdout_encoding_str, *stdout_errors_str;
             PyObject *stringpo;
-            PyObject *stdout_encoding;
-            char *stdout_encoding_str;
             stdout_encoding = PyObject_GetAttrString(fout, "encoding");
-            if (stdout_encoding == NULL) {
-                Py_DECREF(stdin_encoding);
-                return NULL;
-            }
+            stdout_errors = PyObject_GetAttrString(fout, "errors");
+            if (!stdout_encoding || !stdout_errors)
+                goto _readline_errors;
             stdout_encoding_str = _PyUnicode_AsString(stdout_encoding);
-            if (stdout_encoding_str == NULL) {
-                Py_DECREF(stdin_encoding);
-                Py_DECREF(stdout_encoding);
-                return NULL;
-            }
+            stdout_errors_str = _PyUnicode_AsString(stdout_errors);
+            if (!stdout_encoding_str || !stdout_errors_str)
+                goto _readline_errors;
             stringpo = PyObject_Str(promptarg);
-            if (stringpo == NULL) {
-                Py_DECREF(stdin_encoding);
-                Py_DECREF(stdout_encoding);
-                return NULL;
-            }
+            if (stringpo == NULL)
+                goto _readline_errors;
             po = PyUnicode_AsEncodedString(stringpo,
-                stdout_encoding_str, NULL);
-            Py_DECREF(stdout_encoding);
-            Py_DECREF(stringpo);
-            if (po == NULL) {
-                Py_DECREF(stdin_encoding);
-                return NULL;
-            }
+                stdout_encoding_str, stdout_errors_str);
+            Py_CLEAR(stdout_encoding);
+            Py_CLEAR(stdout_errors);
+            Py_CLEAR(stringpo);
+            if (po == NULL)
+                goto _readline_errors;
             prompt = PyBytes_AsString(po);
-            if (prompt == NULL) {
-                Py_DECREF(stdin_encoding);
-                Py_DECREF(po);
-                return NULL;
-            }
+            if (prompt == NULL)
+                goto _readline_errors;
         }
         else {
             po = NULL;
             prompt = "";
         }
         s = PyOS_Readline(stdin, stdout, prompt);
-        Py_XDECREF(po);
         if (s == NULL) {
             if (!PyErr_Occurred())
                 PyErr_SetNone(PyExc_KeyboardInterrupt);
-            Py_DECREF(stdin_encoding);
-            return NULL;
+            goto _readline_errors;
         }
 
         len = strlen(s);
@@ -1727,12 +1716,22 @@ builtin_input(PyObject *self, PyObject *args)
                 len--;   /* strip trailing '\n' */
                 if (len != 0 && s[len-1] == '\r')
                     len--;   /* strip trailing '\r' */
-                result = PyUnicode_Decode(s, len, stdin_encoding_str, NULL);
+                result = PyUnicode_Decode(s, len, stdin_encoding_str,
+                                                  stdin_errors_str);
             }
         }
         Py_DECREF(stdin_encoding);
+        Py_DECREF(stdin_errors);
+        Py_XDECREF(po);
         PyMem_FREE(s);
         return result;
+    _readline_errors:
+        Py_XDECREF(stdin_encoding);
+        Py_XDECREF(stdout_encoding);
+        Py_XDECREF(stdin_errors);
+        Py_XDECREF(stdout_errors);
+        Py_XDECREF(po);
+        return NULL;
     }
 
     /* Fallback if we're not interactive */