diff --git a/chromaterm/__main__.py b/chromaterm/__main__.py index bf3b0fd..ed0f40c 100644 --- a/chromaterm/__main__.py +++ b/chromaterm/__main__.py @@ -345,7 +345,7 @@ def process_input(config, data_fd, forward_fd=None, max_wait=None): buffer = data + separator # 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: + elif 0 < len(buffer) < 2 + data_read.count(b'\b') * 2: sys.stdout.buffer.write(data + separator) buffer = b'' # There's more data; fetch it before highlighting, if buffer isn't full diff --git a/tests/test__main__.py b/tests/test__main__.py index 7c63206..0ef6942 100644 --- a/tests/test__main__.py +++ b/tests/test__main__.py @@ -586,12 +586,13 @@ def test_process_input_read_size(capsys, pcre): assert capsys.readouterr().out == '\x1b[1m' + 'x' * write_size + '\x1b[22m' -def test_process_input_single_character(capsys, pcre): +def test_process_input_single_character(capsys, monkeypatch, pcre): '''Input processing for a single character. Even with a rule that matches single character, the output should not be highlighted as it is typically just keyboard input.''' pipe_r, pipe_w = os.pipe() config = chromaterm.__main__.Config() + run_once = threading.Event() rule = chromaterm.Rule('x', chromaterm.Color('bold'), pcre=pcre) config.rules.append(rule) @@ -601,11 +602,22 @@ def test_process_input_single_character(capsys, pcre): assert capsys.readouterr().out == 'x' - # A single character after a separator is not keyboard input - os.write(pipe_w, b'hi\nx') + # A single character in a trailing chunk is not considered keyboard input + def patched_read_ready(*_, timeout=None): + # If timeout is used, then it's the call to check for more data + if timeout and not run_once.is_set(): + # Write a single character to attempt to trick it into thinking it's keyboard input + os.write(pipe_w, b'x') + os.close(pipe_w) + run_once.set() + + return [pipe_r] + + monkeypatch.setattr(chromaterm.__main__, 'read_ready', patched_read_ready) + os.write(pipe_w, b'hi') chromaterm.__main__.process_input(config, pipe_r, max_wait=0) - assert capsys.readouterr().out == 'hi\n\x1b[1mx\x1b[22m' + assert capsys.readouterr().out == 'hi\x1b[1mx\x1b[22m' def test_process_input_socket(capsys):