Improve keyboard typing detection

chromaterm/__main__.py

@@ -322,9 +322,9 @@ def process_input(config, data_fd, forward_fd=None, max_wait=None):
             # Separator is an incomplete control strings; wait for the rest
             if data_read and separator.startswith(SPLIT_CONTROL_STRINGS):
                 buffer = data + separator
-            # Zero or one characters indicates keyboard typing; don't highlight
-            # Account for the backspaces added by some shells, like zsh
-            elif len(data) < 2 + data.count(b'\b') * 2:
+            # A single character indicates keyboard typing; don't highlight
+            # Account for backspaces added by some shells, like zsh
+            elif 0 < len(data_read) < 2 + data_read.count(b'\b') * 2:
                 sys.stdout.buffer.write(data + separator)
                 buffer = b''
             # There's more data; fetch it before highlighting

tests/test__main__.py

@@ -506,7 +506,7 @@ def patched_read_ready(*_1, timeout=None):
         for code in ['\x50', '\x58', '\x5d', '\x5e', '\x5f']:
             # Data (printed), followed by the first part (not printed)
             event.clear()
-            os.write(pipe_w, b'hello\n\x1b' + code.encode('utf-8') + b'p1')
+            os.write(pipe_w, b'hello\n\x1b' + code.encode() + b'p1')
             event.wait()
             assert capsys.readouterr().out == 'hello\n'
 
@@ -548,14 +548,20 @@ def test_process_input_single_character(capsys):
     pipe_r, pipe_w = os.pipe()
     config = chromaterm.__main__.Config()
 
-    rule = chromaterm.Rule('.', color=chromaterm.Color('bold'))
+    rule = chromaterm.Rule('x', color=chromaterm.Color('bold'))
     config.rules.append(rule)
 
     os.write(pipe_w, b'x')
     chromaterm.__main__.process_input(config, pipe_r, max_wait=0)
 
     assert capsys.readouterr().out == 'x'
 
+    # A single character after a separator is not keyboard input
+    os.write(pipe_w, b'hi\nx')
+    chromaterm.__main__.process_input(config, pipe_r, max_wait=0)
+
+    assert capsys.readouterr().out == 'hi\n\x1b[1mx\x1b[22m'
+
 
 def test_process_input_trailing_chunk(capsys):
     '''Ensure that a trailing chunk is joined with the next chunk if the latter
@@ -574,6 +580,8 @@ def test_process_input_trailing_chunk(capsys):
     os.write(pipe_w, b'hello ')
     while select.select([pipe_r], [], [], 0)[0]:
         pass
+    assert capsys.readouterr().out == ''
+
     os.write(pipe_w, b'world')
 
     os.close(pipe_w)