""" Offline emulator tests: queries answered, modes tracked, parsing survives chunk boundaries. Run with: python3 tests/test_emulator.py """ import _bootstrap # sets sys.path for the imports below from _bootstrap import FakeProcess from vterm.emulator import VtScreen, VtStream def make(columns=80, lines=24): proc = FakeProcess() screen = VtScreen(columns, lines, process=proc) stream = VtStream(screen) return proc, screen, stream def feed(stream, data, char_by_char=False): if char_by_char: for c in data: stream.feed(c) else: stream.feed(data) def run_case(char_by_char): tag = "char-by-char" if char_by_char else "chunked" # DA1 answered; apps can wait on it with no timeout proc, screen, stream = make() feed(stream, "\x1b[c", char_by_char) assert proc.written == "\x1b[?62;22c", (tag, proc.written) # CPR, plain and private form proc, screen, stream = make() feed(stream, "\x1b[5;10H\x1b[6n\x1b[?6n", char_by_char) assert proc.written == "\x1b[5;10R\x1b[?5;10R", (tag, proc.written) # XTVERSION then DECRQM 2026, the sync-output probe sequence proc, screen, stream = make() feed(stream, "\x1b[>0q\x1b[c\x1b[?2026$p\x1b[c", char_by_char) assert proc.written == ( "\x1bP>|vterm 0.1\x1b\\" "\x1b[?62;22c" "\x1b[?2026;2$y" "\x1b[?62;22c"), (tag, proc.written) # DECRQM for DECAWM (7) and DECTCEM (25): both default-on, and the # wire-number list must answer them as set, not as unrecognized proc, screen, stream = make() feed(stream, "\x1b[?7$p\x1b[?25$p\x1b[?2026$p", char_by_char) assert proc.written == "\x1b[?7;1$y\x1b[?25;1$y\x1b[?2026;2$y", (tag, proc.written) # kitty probe: push ignored, query answered 0, DA1 fence answered proc, screen, stream = make() feed(stream, "\x1b[>7u\x1b[?u\x1b[c\x1b[4;2mhello", char_by_char) assert screen.modify_other_keys == 2, (tag, screen.modify_other_keys) assert screen.buffer[0][0].data == "h" and not screen.buffer[0][0].underscore, tag feed(stream, "\x1b[>4;0m", char_by_char) assert screen.modify_other_keys == 0, (tag, screen.modify_other_keys) # another resource, and the bare form, leave it alone feed(stream, "\x1b[>4;2m\x1b[>1;2m\x1b[>m", char_by_char) assert screen.modify_other_keys == 2, (tag, screen.modify_other_keys) # OSC 11 color query answered; OSC 0 sets title; OSC 133 dropped proc, screen, stream = make() screen.default_bg_color = "1a2b3c" feed(stream, "\x1b]0;mytitle\x07\x1b]11;?\x07\x1b]133;A\x07x", char_by_char) assert screen.title == "mytitle", (tag, screen.title) assert proc.written == "\x1b]11;rgb:1a1a/2b2b/3c3c\x07", (tag, proc.written) assert screen.buffer[0][0].data == "x", tag # OSC with ST terminator; a query asked with ST is answered with ST (meli) proc, screen, stream = make() screen.default_bg_color = "1a2b3c" feed(stream, "\x1b]2;st-title\x1b\\\x1b]11;?\x1b\\y", char_by_char) assert screen.title == "st-title", (tag, screen.title) assert proc.written == "\x1b]11;rgb:1a1a/2b2b/3c3c\x1b\\", (tag, proc.written) assert screen.buffer[0][0].data == "y", tag # OSC 52 (clipboard write) is dropped, not drawn: untrusted output must # not plant a paste proc, screen, stream = make() feed(stream, "\x1b]52;c;aGk=\x07z", char_by_char) assert screen.buffer[0][0].data == "z" and proc.written == "", tag # bracketed paste and sync modes tracked proc, screen, stream = make() feed(stream, "\x1b[?2004h\x1b[?2026h", char_by_char) assert screen.bracketed_paste_enabled(), tag assert screen.in_sync_frame(), tag feed(stream, "\x1b[?2026l\x1b[?2004l", char_by_char) assert not screen.in_sync_frame(), tag assert not screen.bracketed_paste_enabled(), tag # mouse modes proc, screen, stream = make() feed(stream, "\x1b[?1000h\x1b[?1002h\x1b[?1004h\x1b[?1006h", char_by_char) assert screen.mouse_reporting_enabled(), tag assert screen.focus_tracking_enabled(), tag # alt screen switch and restore proc, screen, stream = make() feed(stream, "primary\x1b[?1049h\x1b[2Jalt\x1b[?1049l", char_by_char) row0 = "".join(screen.buffer[0][i].data for i in range(7)) assert row0 == "primary", (tag, row0) assert not screen.alternate_buffer_mode, tag # truecolor SGR is quantized into the 256 palette; 256-color as hex; bold tracked proc, screen, stream = make() feed(stream, "\x1b[1;38;2;255;0;10mX\x1b[0m\x1b[38;5;196mY", char_by_char) assert screen.buffer[0][0].fg == "ff0000" and screen.buffer[0][0].bold, tag assert screen.buffer[0][1].fg == "ff0000", (tag, screen.buffer[0][1].fg) # every grey lands inside the palette; 248 used to map to slot 256 and # abort the whole SGR from vterm.emulator import quantize_256 assert all(0 <= quantize_256(v, v, v) <= 255 for v in range(256)) proc, screen, stream = make() feed(stream, "\x1b[38;2;248;248;248mX", char_by_char) assert screen.buffer[0][0].fg == "ffffff", (tag, screen.buffer[0][0].fg) # a zero-width non-combining char (VS16, ZWJ) or a stray control byte is # dropped; the text after it must survive proc, screen, stream = make() feed(stream, "⚠️ warn \U0001F468‍\U0001F4BB ab\x01cd", char_by_char) row = "".join(screen.buffer[0][x].data for x in range(20)).rstrip() assert row == "⚠ warn \U0001F468\U0001F4BB abcd", (tag, repr(row)) # SGR 5/25 (blink): ignored, not a crash — Char has no blink field proc, screen, stream = make() feed(stream, "\x1b[5;31mX\x1b[25mY", char_by_char) assert screen.buffer[0][0].fg == "red", (tag, screen.buffer[0][0]) assert screen.buffer[0][1].fg == "red" and screen.buffer[0][1].data == "Y", tag # DECSTBM + SU/SD don't crash and scroll within margins proc, screen, stream = make(10, 6) feed(stream, "\x1b[2;5r\x1b[2S\x1b[1T\x1b[r", char_by_char) assert screen.margins is None, tag # DECSCUSR ignored, CSI t ignored, ?996n answered proc, screen, stream = make() feed(stream, "\x1b[2 q\x1b[16t\x1b[?996n", char_by_char) assert proc.written == "\x1b[?997;1n", (tag, proc.written) # scrollback: lines pushed to history when scrolling at bottom proc, screen, stream = make(10, 3) feed(stream, "a\r\nb\r\nc\r\nd", char_by_char) assert len(screen.history) == 1, (tag, len(screen.history)) assert screen.history[0][0].data == "a", tag # an explicit full-screen scroll region still feeds scrollback proc, screen, stream = make(10, 3) feed(stream, "\x1b[1;3ra\r\nb\r\nc\r\nd", char_by_char) assert len(screen.history) == 1, (tag, len(screen.history)) # ED 3 clears history and fires callback cleared = [] proc = FakeProcess() screen = VtScreen(10, 3, process=proc, clear_callback=lambda: cleared.append(1)) stream = VtStream(screen) feed(stream, "a\r\nb\r\nc\r\nd\x1b[3J", char_by_char) assert len(screen.history) == 0 and cleared, tag # DCS/APC payloads are consumed, not drawn proc, screen, stream = make() feed(stream, "\x1bP1$rpayload\x1b\\Z\x1b_Gapc\x1b\\W", char_by_char) row0 = "".join(screen.buffer[0][i].data for i in range(2)) assert row0 == "ZW", (tag, row0) # DEC special graphics: xterm-256color's smacs is ESC(0, so ncurses apps # (mutt, dialog, less) draw borders with it when the locale is not UTF-8 proc, screen, stream = make(20, 3) feed(stream, "\x1b(0lqqqk\x1b(Bplain", char_by_char) row = "".join(screen.buffer[0][x].data for x in range(10)) assert row == "\u250c\u2500\u2500\u2500\u2510plain", (tag, repr(row)) # G1 designation plus SO/SI stays ignored in UTF-8 mode, so a stray 0x0e in # decoded text cannot silently switch the charset proc, screen, stream = make(20, 3) feed(stream, "\x1b)0\x0eq\x0f", char_by_char) assert screen.buffer[0][0].data == "q", (tag, screen.buffer[0][0].data) # multi-byte text is unaffected by a graphics designation proc, screen, stream = make(20, 3) feed(stream, "\x1b(0\u00e9\u2500", char_by_char) row = "".join(screen.buffer[0][x].data for x in range(2)) assert row == "\u00e9\u2500", (tag, repr(row)) # shift+enter esc-cr passthrough sanity: ESC CR fed back in draws nothing odd proc, screen, stream = make() feed(stream, "\x1b[?25l\x1b[?25h", char_by_char) assert not screen.cursor.hidden, tag # DECSCUSR tracked; missing parameter means 0; reset restores the default proc, screen, stream = make() feed(stream, "\x1b[2 q", char_by_char) assert screen.cursor_style == 2, (tag, screen.cursor_style) feed(stream, "\x1b[ q", char_by_char) assert screen.cursor_style == 0, (tag, screen.cursor_style) feed(stream, "\x1b[6 q", char_by_char) screen.reset() assert screen.cursor_style == 0, (tag, screen.cursor_style) # nothing leaked to the screen or the pty assert screen.buffer[0][0].data == " " and proc.written == "", tag print("all emulator tests passed ({})".format(tag)) def run_resize_cases(): """Shrinking must drop cells past the new width, in both buffers: the segmenter walks a line to its highest key, so leftovers keep rendering.""" proc, screen, stream = make(20, 3) feed(stream, "ABCDEFGHIJKLMNOPQRST", False) screen.resize(3, 10) assert max(screen.buffer[0]) < 10, sorted(screen.buffer[0]) # the buffer parked by the alt screen is trimmed too, or switching back # brings the stale cells with it proc, screen, stream = make(20, 3) feed(stream, "PRIMARYPRIMARYPRIMAR", False) feed(stream, "\x1b[?1049h", False) screen.resize(3, 8) feed(stream, "\x1b[?1049l", False) assert max(screen.buffer[0]) < 8, sorted(screen.buffer[0]) # a shrink clamps the cursor; without the clamp draw() writes to rows # above or below the screen proc, screen, stream = make(10, 8) feed(stream, "\r\n".join("line{}".format(i) for i in range(8)), False) feed(stream, "\x1b[1;1H", False) # full screen, cursor at home screen.resize(4, 10) assert screen.cursor.y == 0, screen.cursor.y proc, screen, stream = make(10, 8) feed(stream, "top\x1b[7;1H", False) # cursor on a blank row below content screen.resize(4, 10) assert screen.cursor.y == 3, screen.cursor.y proc, screen, stream = make(20, 4) feed(stream, "x" * 15, False) screen.resize(4, 10) assert screen.cursor.x == 10, screen.cursor.x # mode 2031 is how a live theme change reaches the app proc, screen, stream = make() assert not screen.theme_notify_enabled() feed(stream, "\x1b[?2031h", False) assert screen.theme_notify_enabled() print("all resize tests passed") def run_private_csi_cases(): """No private CSI may raise: an exception here kills the renderer thread and freezes the tab. aerc probes XTSMGRAPHICS (ESC[?2;1;0S).""" for final in "ABCDEFGHJKLMPSTXZbcdfghlmnqrstu@": proc, screen, stream = make() stream.feed("\x1b[?2;1;0" + final) # multi-parameter private CSI stream.feed("\x1b[?1" + final) # single-parameter stream.feed("\x1b[?" + final) # bare # the ones we do support still work proc, screen, stream = make() feed(stream, "\x1b[?1000h\x1b[?1002h\x1b[?1006h", False) assert screen.mouse_reporting_enabled(), "private h must still set modes" feed(stream, "\x1b[?1000l\x1b[?1002l", False) assert not screen.mouse_reporting_enabled(), "private l must still reset modes" proc, screen, stream = make() feed(stream, "\x1b[5;10H\x1b[?6n", False) assert proc.written == "\x1b[?5;10R", proc.written # ...and an unsupported one is ignored rather than answered or dispatched proc, screen, stream = make() feed(stream, "\x1b[?2;1;0S", False) assert proc.written == "", "XTSMGRAPHICS must be ignored, not answered" print("all private CSI tests passed") def run_decscnm_cases(): """A DECSCNM flash (vim's visual bell) must not outlive itself. The lines point at the blank cell, so one keeping a reverse default paints the columns the app never wrote — vim skips indentation with cursor-forward — as white blocks. """ from vterm.render import segment_buffer_line # a gutter, then the indent skipped with CSI 8C, then the text: vim's shape paint = "\x1b[1;1H\x1b[93m 47 \x1b[m\x1b[8C\x1b[96mreturn\x1b[m default" for order in (paint + "\x1b[?5h\x1b[?5l", "\x1b[?5h\x1b[?5l" + paint): proc, screen, stream = make(columns=60, lines=5) feed(stream, order, False) assert not screen.default_char.reverse, "the flash is over" assert not screen.buffer[0].default.reverse, "line kept the flash's blank cell" assert not screen.buffer[0][6].reverse, "untouched column left reverse" painted = [s for s in segment_buffer_line(screen.buffer[0]) if not s[0].strip() and s[3] == "reverse_default"] assert not painted, ("whitespace painted reverse", painted) # while the flash is on, the untouched columns invert too proc, screen, stream = make(columns=60, lines=5) feed(stream, paint + "\x1b[?5h", False) assert screen.buffer[0].default.reverse and screen.buffer[0][6].reverse, \ "DECSCNM must invert the whole screen, not only the written columns" print("all DECSCNM tests passed") # everything the two paths must agree on: text, SGR in each form, empty and # clamped params, cursor moves, private modes, prefixed CSI, DECRQM, OSC, # scroll regions, controls inside a CSI, a colon subparameter, wide chars GAUNTLET = ( "plain text\r\n" "\x1b[1;31mbold red\x1b[m normal \x1b[38;5;190mpalette\x1b[0m\r\n" "\x1b[38;2;10;200;30mtruecolor\x1b[49m\x1b[m\r\n" "\x1b[H\x1b[2Kover\twritten\x1b[3;5Hmoved\x1b[999999C clamp\x1b[m\r\n" "\x1b[?25l\x1b[?1049h\x1b[?25h\x1b[?1049l" "\x1b[>q\x1b[?2026$p\x1b[5n\x1b[6n" "\x1b]0;a title\x07\x1b]11;?\x07" "\x1b[2;5r\x1b[Sscrolled\x1b[T\x1b[r" "\x1b[38:5:100mcolon\x1b[m\r\n" "\x1b[2\x073m bel mid-CSI \x1b[3\x18X can cancels\r\n" "wide: 漢字 \x1b[7m漢\x1b[m ⚠️ after\r\n" "\x1b[m\x1b[;5m\x1b[5;m\x1b[25m params\r\n" ) def run_fast_path_cases(): """The feed() fast path must be indistinguishable from the parser.""" def state(screen): rows = [ tuple(screen.buffer[y][x] for x in range(screen.columns)) for y in range(screen.lines) ] c = screen.cursor return (rows, c.x, c.y, c.attrs, set(screen.mode), screen.title) def parser_only(blob): """The blob through the coroutine parser alone, bypassing the fast path.""" proc, screen, stream = make(columns=30, lines=8) import pyte pyte.Stream.feed(stream, blob) return proc, screen ref_proc, ref_screen = parser_only(GAUNTLET) reference = state(ref_screen) # chunk sizes that split every sequence somewhere; 1 exercises pure carry for size in (1, 2, 3, 7, 64, len(GAUNTLET)): proc, screen, stream = make(columns=30, lines=8) for i in range(0, len(GAUNTLET), size): stream.feed(GAUNTLET[i : i + size]) assert state(screen) == reference, "fast path diverged at chunk size {}".format(size) assert proc.written == ref_proc.written, (size, proc.written, ref_proc.written) # a param run past the carry bound still parses, via the parser proc, screen, stream = make() stream.feed("\x1b[" + "1;" * 40) stream.feed("31mX") assert screen.buffer[0][0].fg == "red", screen.buffer[0][0] # a carried partial does not linger: nothing dispatches until it completes proc, screen, stream = make() stream.feed("A\x1b[3") assert screen.buffer[0][1].data == " ", "partial CSI must not be drawn" stream.feed("1mB") assert screen.buffer[0][1].fg == "red" and screen.buffer[0][1].data == "B" print("all fast-path tests passed") def run_malformed_csi_cases(): """A malformed CSI must cost one sequence, not the chunk. Excess params used to TypeError out of the dispatch; feed_data caught it and dropped up to 256K of output. Triggered by cat'ing a binary file or a corrupted stream. The SGR range guards the blink class the same way. """ for char_by_char in (False, True): # fast path / parser for final in (chr(c) for c in range(0x40, 0x7F)): for params in ("2;3", "0;0;0"): proc, screen, stream = make(columns=20, lines=6) feed(stream, "ab\x1b[{}{}cd".format(params, final), char_by_char) row = "".join(screen.buffer[y][x].data for y in range(6) for x in range(20)) assert "c" in row and "d" in row, \ ("output after a malformed CSI was lost", params, final, char_by_char) for n in range(121): proc, screen, stream = make(columns=20, lines=6) feed(stream, "a\x1b[{}mb".format(n), char_by_char) assert screen.buffer[0][1].data == "b", ("SGR {} ate the chunk".format(n), char_by_char) print("all malformed-CSI tests passed") def run_clear_cases(): """Screen wipe and post-clear history suppression (vterm_clear_screen).""" proc, screen, stream = make(10, 4) feed(stream, "aaa\r\nbbb\r\nccc\r\nddd\r\neee", False) assert len(screen.history) > 0 assert screen.buffer[0][0].data != " " screen.wipe() rows = ["".join(screen.buffer[y][x].data for x in range(screen.columns)).strip() for y in range(screen.lines)] assert rows == ["", "", "", ""], rows assert (screen.cursor.x, screen.cursor.y) == (0, 0) assert screen.dirty == set(range(4)), screen.dirty # the app's repaint right after a clear must not refill scrollback screen.history.clear() feed(stream, "one\r\ntwo\r\nthree\r\nfour\r\nfive", False) assert len(screen.history) == 0, len(screen.history) # ... but scrollback works again once the window expires screen._suppress_history_until = 0 feed(stream, "\r\nsix\r\nseven", False) assert len(screen.history) > 0, "history suppressed for too long" # ED 3 is app-driven: it clears scrollback and fires the clear callback, # but (as in xterm) leaves the visible screen alone cleared = [] proc = FakeProcess() screen = VtScreen(10, 4, process=proc, clear_callback=lambda: cleared.append(1)) stream = VtStream(screen) feed(stream, "gone\r\nalso-gone\r\nmore\r\nlines\r\nx", False) assert len(screen.history) > 0 feed(stream, "\x1b[3J", False) assert len(screen.history) == 0 and cleared, (len(screen.history), cleared) assert screen.buffer[0][0].data == "a", screen.buffer[0][0] # with erase_scrollback off, ED 3 is a no-op: an app repainting with it # would take the history and the view with it cleared = [] proc = FakeProcess() screen = VtScreen(10, 4, process=proc, erase_scrollback=False, clear_callback=lambda: cleared.append(1)) stream = VtStream(screen) feed(stream, "kept\r\nalso-kept\r\nmore\r\nlines\r\nx", False) kept = len(screen.history) assert kept > 0 del cleared[:] # the screen's own reset() fired one at construction feed(stream, "\x1b[3J", False) assert len(screen.history) == kept and not cleared, (len(screen.history), cleared) assert screen.buffer[0][0].data == "a", screen.buffer[0][0] print("all clear-screen tests passed") def run_parser_edge_cases(): """Sequences the stock collector misread: intermediates, `:` subparams, REP, marks.""" for char_by_char in (False, True): tag = "char-by-char" if char_by_char else "chunked" # DECSTR (`reset` sends it): `!` is an intermediate, the `p` must not be drawn proc, screen, stream = make(20, 4) feed(stream, "\x1b[!pX", char_by_char) assert screen.display[0].rstrip() == "X", (tag, repr(screen.display[0])) # `4:3` is curly underline, not SGR 43 (yellow background) proc, screen, stream = make(20, 4) feed(stream, "\x1b[4:3mX\x1b[38:2:1:2:3mY", char_by_char) cell = screen.buffer[0][0] assert (cell.bg, cell.underscore) == ("default", True), (tag, cell) assert screen.buffer[0][1].data == "Y", (tag, screen.buffer[0][1]) # REP repeats the last printable; ncurses emits it for runs proc, screen, stream = make(20, 4) feed(stream, "a\x1b[3b\x1b[31mb\x1b[b", char_by_char) assert screen.display[0].rstrip() == "aaaabb", (tag, repr(screen.display[0])) assert screen.buffer[0][5].fg == "red", (tag, screen.buffer[0][5]) # a Thai vowel sign has combining class 0 but is still a mark: merged, not dropped proc, screen, stream = make(20, 4) feed(stream, "\u0e01\u0e31x", char_by_char) assert screen.buffer[0][0].data == "\u0e01\u0e31", (tag, screen.buffer[0][0]) assert screen.buffer[0][1].data == "x", (tag, screen.buffer[0][1]) # DECRQM: the modes we honour answer set/reset, never "unrecognized" proc, screen, stream = make() feed(stream, "\x1b[?2031$p\x1b[?1h\x1b[?1$p\x1b[4$p\x1b[20$p", char_by_char) assert proc.written == "\x1b[?2031;2$y\x1b[?1;1$y\x1b[4;2$y\x1b[20;2$y", (tag, proc.written) # RIS on the alternate screen lands back on a cleared primary screen proc, screen, stream = make(20, 4) feed(stream, "primary\x1b[?1049halt\x1bc", char_by_char) assert not screen.alternate_buffer_mode and not screen.savepoints, tag assert screen.display[0].strip() == "", (tag, repr(screen.display[0])) feed(stream, "\r\n" * 4, char_by_char) assert len(screen.history) == 1, (tag, len(screen.history)) # SCOSC/SCORC, and CPR relative to the region under DECOM proc, screen, stream = make(20, 4) feed(stream, "ab\x1b[s\x1b[3;3H\x1b[uX\x1b[2;3r\x1b[?6h\x1b[6n", char_by_char) assert screen.display[0].rstrip() == "abX", (tag, repr(screen.display[0])) assert proc.written == "\x1b[1;1R", (tag, proc.written) # a string with no terminator ends at the next ESC, as in xterm; a stray # ESC P from a binary must not swallow the rest of the session proc, screen, stream = make(20, 4) feed(stream, "\x1bPgarbage\x1bcX\x1b]0;title\x1b[1mY", char_by_char) assert screen.display[0].rstrip() == "XY", (tag, repr(screen.display[0])) assert screen.title == "title" and screen.buffer[0][1].bold, (tag, screen.title) print("all parser edge-case tests passed") run_case(char_by_char=False) run_case(char_by_char=True) run_parser_edge_cases() run_clear_cases() run_resize_cases() run_private_csi_cases() run_decscnm_cases() run_fast_path_cases() run_malformed_csi_cases()