Fix input of long unicode characters

- with two codepoints or more

Author: sergey-miryanov

Lib/_pyrepl/base_eventqueue.py

@@ -73,14 +73,19 @@ def push(self, char: int | bytes | str) -> None:
         """
         Processes a character by updating the buffer and handling special key mappings.
         """
-        ord_char = char if isinstance(char, int) else ord(char)
-        if ord_char > 255:
-            assert isinstance(char, str)
-            char = bytes(char.encode(self.encoding, "replace"))
+
+        if isinstance(char, bytes):
             self.buf.extend(char)
         else:
-            char = bytes(bytearray((ord_char,)))
-            self.buf.append(ord_char)
+
+            ord_char = char if isinstance(char, int) else ord(char)
+            if ord_char > 255:
+                assert isinstance(char, str)
+                char = bytes(char.encode(self.encoding, "replace"))
+                self.buf.extend(char)
+            else:
+                char = bytes(bytearray((ord_char,)))
+                self.buf.append(ord_char)
 
         if char in self.keymap:
             if self.keymap is self.compiled_keymap:
@@ -108,7 +113,9 @@ def push(self, char: int | bytes | str) -> None:
             try:
                 decoded = bytes(self.buf).decode(self.encoding)
             except UnicodeError:
-                return
+                self.flush_buf()
+                raise
             else:
                 self.insert(Event('key', decoded, self.flush_buf()))
-            self.keymap = self.compiled_keymap
+            finally:
+                self.keymap = self.compiled_keymap

Lib/_pyrepl/simple_interact.py

@@ -158,3 +158,8 @@ def maybe_run_command(statement: str) -> bool:
         except MemoryError:
             console.write("\nMemoryError\n")
             console.resetbuffer()
+        # except SystemExit:
+        #     break
+        # except:
+        #     console.showtraceback()
+        #     console.resetbuffer()

Lib/_pyrepl/windows_console.py

@@ -468,7 +468,8 @@ def get_event(self, block: bool = True) -> Event | None:
                 return None
             elif self.__vt_support:
                 # If virtual terminal is enabled, scanning VT sequences
-                self.event_queue.push(rec.Event.KeyEvent.uChar.UnicodeChar)
+                key_bytes = raw_key.encode(self.event_queue.encoding)
+                self.event_queue.push(key_bytes)
                 continue
 
             if key_event.dwControlKeyState & ALT_ACTIVE:

Lib/test/test_pyrepl/test_eventqueue.py

@@ -129,6 +129,77 @@ def test_push_unicode_character(self):
         self.assertEqual(eq.events[0].evt, "key")
         self.assertEqual(eq.events[0].data, "ч")
 
+    def test_push_unicode_character_as_bytes(self):
+        eq = self.make_eventqueue()
+        eq.keymap = {}
+
+        eq.push("ч".encode(eq.encoding, "replace"))
+        e = eq.get()
+        self.assertEqual(e.evt, "key")
+        self.assertEqual(e.data, "ч")
+
+    def test_push_long_unicode_character_as_bytes(self):
+        eq = self.make_eventqueue()
+        eq.keymap = {}
+
+        def _event(evt, data, raw=None):
+            r = raw if raw is not None else data.encode(eq.encoding)
+            e = Event(evt, data, r)
+            return e
+
+        def _push(keys):
+            for k in keys:
+                eq.push(k)
+
+        _push("\x1b[200")
+        eq.push("ñ".encode(eq.encoding, "replace"))
+        _push("\x1b[201")
+
+        self.assertEqual(eq.get(), _event("key", "\x1b"))
+        self.assertEqual(eq.get(), _event("key", "["))
+        self.assertEqual(eq.get(), _event("key", "2"))
+        self.assertEqual(eq.get(), _event("key", "0"))
+        self.assertEqual(eq.get(), _event("key", "0"))
+
+        self.assertEqual(eq.get(), _event("key", "ñ", bytearray(b'\xc3\xb1')))
+
+        self.assertEqual(eq.get(), _event("key", "\x1b"))
+        self.assertEqual(eq.get(), _event("key", "["))
+        self.assertEqual(eq.get(), _event("key", "2"))
+        self.assertEqual(eq.get(), _event("key", "0"))
+        self.assertEqual(eq.get(), _event("key", "1"))
+
+    def test_push_long_unicode_character(self):
+        eq = self.make_eventqueue()
+        eq.keymap = {}
+
+        def _event(evt, data, raw=None):
+            r = raw if raw is not None else data.encode(eq.encoding)
+            e = Event(evt, data, r)
+            return e
+
+        def _push(keys):
+            for k in keys:
+                eq.push(k)
+
+        _push("\x1b[200")
+        msg = "'utf-8' codec can't decode byte 0xf1 in position 0: unexpected end of data"
+        with self.assertRaisesRegex(UnicodeDecodeError, msg):
+            eq.push("ñ")
+        _push("\x1b[201")
+
+        self.assertEqual(eq.get(), _event("key", "\x1b"))
+        self.assertEqual(eq.get(), _event("key", "["))
+        self.assertEqual(eq.get(), _event("key", "2"))
+        self.assertEqual(eq.get(), _event("key", "0"))
+        self.assertEqual(eq.get(), _event("key", "0"))
+
+        self.assertEqual(eq.get(), _event("key", "\x1b"))
+        self.assertEqual(eq.get(), _event("key", "["))
+        self.assertEqual(eq.get(), _event("key", "2"))
+        self.assertEqual(eq.get(), _event("key", "0"))
+        self.assertEqual(eq.get(), _event("key", "1"))
+
 
 @unittest.skipIf(support.MS_WINDOWS, "No Unix event queue on Windows")
 class TestUnixEventQueue(EventQueueTestBase, unittest.TestCase):