Add tests for XTermParser chunking

src/textual/_xterm_parser.py

@@ -127,6 +127,16 @@ class XTermParser(Parser[events.Event]):
                 # Could be the escape key was pressed OR the start of an escape sequence
                 sequence: str = character
                 if not bracketed_paste:
+                    # TODO: There's nothing left in the buffer at the moment,
+                    #  but since we're on an escape, how can we be sure that the
+                    #  data that next gets fed to the parser isn't an escape sequence?
+
+                    #  This problem arises when an ESC falls at the end of a chunk.
+                    #  We'll be at an escape, but peek_buffer will return an empty
+                    #  string because there's nothing in the buffer yet.
+
+                    #  This code makes an assumption that an escape sequence will never be
+                    #  "chopped up", so buffers would never contain partial escape sequences.
                     peek_buffer = yield self.peek_buffer()
                     if not peek_buffer:
                         # An escape arrived without any following characters

tests/test_xterm_parser.py

@@ -1,3 +1,4 @@
+import itertools
 from unittest import mock
 
 import pytest
@@ -16,12 +17,19 @@ from textual.messages import TerminalSupportsSynchronizedOutput
 
 
 def chunks(data, size):
+    if size == 0:
+        yield data
+        return
+
     chunk_start = 0
     chunk_end = size
-    while chunk_end <= len(data):
+    while True:
         yield data[chunk_start:chunk_end]
         chunk_start = chunk_end
         chunk_end += size
+        if chunk_end >= len(data):
+            yield data[chunk_start:chunk_end]
+            break
 
 
 @pytest.fixture
@@ -29,6 +37,22 @@ def parser():
     return XTermParser(sender=mock.sentinel, more_data=lambda: False)
 
 
+@pytest.mark.parametrize("chunk_size", [2,3,4,5,6])
+def test_varying_parser_chunk_sizes_no_missing_data(parser, chunk_size):
+    end = "\x1b[8~"
+    text = "ABCDEFGH"
+
+    data = end + text
+    events = []
+    for chunk in chunks(data, chunk_size):
+        events.append(parser.feed(chunk))
+
+    events = list(itertools.chain.from_iterable(list(event) for event in events))
+
+    assert events[0].key == "end"
+    assert [event.key for event in events[1:]] == list(text)
+
+
 def test_bracketed_paste(parser):
     """ When bracketed paste mode is enabled in the terminal emulator and
     the user pastes in some text, it will surround the pasted input
@@ -85,7 +109,14 @@ def test_cant_match_escape_sequence_too_long(parser):
         assert events[index].key == character
 
 
-def test_unknown_sequence_followed_by_known_sequence(parser):
+@pytest.mark.parametrize("chunk_size", [
+    pytest.param(2, marks=pytest.mark.xfail(reason="Fails when ESC at end of chunk")),
+    3,
+    pytest.param(4, marks=pytest.mark.xfail(reason="Fails when ESC at end of chunk")),
+    5,
+    6,
+])
+def test_unknown_sequence_followed_by_known_sequence(parser, chunk_size):
     """ When we feed the parser an unknown sequence followed by a known
     sequence. The characters in the unknown sequence are delivered as keys,
     and the known escape sequence that follows is delivered as expected.
@@ -94,15 +125,20 @@ def test_unknown_sequence_followed_by_known_sequence(parser):
     known_sequence = "\x1b[8~"  # key = 'end'
 
     sequence = unknown_sequence + known_sequence
-    events = parser.feed(sequence)
 
-    assert next(events).key == "^"
+    events = []
-    assert next(events).key == "["
+    parser.more_data = lambda: True
-    assert next(events).key == "?"
+    for chunk in chunks(sequence, chunk_size):
-    assert next(events).key == "end"
+        events.append(parser.feed(chunk))
 
-    with pytest.raises(StopIteration):
+    events = list(itertools.chain.from_iterable(list(event) for event in events))
-        next(events)
+
+    assert [event.key for event in events] == [
+        "^",
+        "[",
+        "?",
+        "end",
+    ]
 
 
 def test_simple_key_presses_all_delivered_correct_order(parser):