From a6755701e3b7190c7850e0861f3cadc22865df23 Mon Sep 17 00:00:00 2001 From: Jim Huang Date: Sun, 1 Mar 2026 15:55:11 +0800 Subject: [PATCH] Fix Windows terminal regressions across Py version This restructures _flush() so flush always runs regardless of whether os.get_blocking succeeds. The old code coupled the get_blocking probe with the flush inside single try/except OSError when the probe failed (AttributeError on Python <3.12, OSError on 3.11 console handles), the flush was skipped and nothing reached the screen. Each operation (fileno, get_blocking, set_blocking, flush, restore) is now isolated in its own try/except, with a finally block to guarantee blocking-state restoration. It also sets DISABLE_NEWLINE_AUTO_RETURN (0x0008) alongside VT100 in _init_windows() to prevent auto-CR on LF causing cursor drift, with graceful fallback for pre-1607 builds. Close #48 --- .ci/validate-rawterm.py | 92 +++++++++++++++++++++++++++++++++++++++++ menuconfig.py | 4 +- rawterm.py | 65 +++++++++++++++++++++++------ 3 files changed, 148 insertions(+), 13 deletions(-) diff --git a/.ci/validate-rawterm.py b/.ci/validate-rawterm.py index 2d07590..27fe6b1 100755 --- a/.ci/validate-rawterm.py +++ b/.ci/validate-rawterm.py @@ -18,6 +18,30 @@ _IS_WINDOWS = os.name == "nt" +def check_version(): + """Minimal Python version checks and platform capability report.""" + ver = sys.version_info + assert ver >= (3, 6), "Python >= 3.6 required, got {}.{}".format(ver[0], ver[1]) + print("Python {}.{}.{} on {}".format(ver[0], ver[1], ver[2], sys.platform)) + + if _IS_WINDOWS: + has_get_blocking = hasattr(os, "get_blocking") + print( + " os.get_blocking: {}".format( + "available" if has_get_blocking else "unavailable (Python <3.12)" + ) + ) + if has_get_blocking: + # Probe whether it actually works on console handles + try: + os.get_blocking(sys.stdout.fileno()) + print(" os.get_blocking(stdout): works") + except OSError: + print(" os.get_blocking(stdout): OSError (console handle)") + + print("version checks passed") + + def check_rawterm_units(): """rawterm Color, Style, Key, Box -- no terminal required.""" from rawterm import Style, Color, Key, Box, NAMED_COLORS @@ -165,6 +189,21 @@ def check_windows_console(): stdout_h, out_mode.value ), "SetConsoleMode(stdout, restore) failed" + # Test VT100 + DISABLE_NEWLINE_AUTO_RETURN combination + DISABLE_NEWLINE_AUTO_RETURN = 0x0008 + new_out_both = ( + out_mode.value + | ENABLE_VIRTUAL_TERMINAL_PROCESSING + | DISABLE_NEWLINE_AUTO_RETURN + ) + dnar_ok = kernel32.SetConsoleMode(stdout_h, new_out_both) + kernel32.SetConsoleMode(stdout_h, out_mode.value) # always restore + print( + " DISABLE_NEWLINE_AUTO_RETURN: {}".format( + "supported" if dnar_ok else "not supported (pre-1607)" + ) + ) + # Test VT100 input mode (may fail on older Windows -- not fatal) ENABLE_VIRTUAL_TERMINAL_INPUT = 0x0200 new_in = (in_mode.value | ENABLE_VIRTUAL_TERMINAL_INPUT) & ~0x0007 @@ -265,8 +304,61 @@ def check_menuconfig_headless(): print("menuconfig headless + style validation passed") +def check_flush_robustness(): + """Verify _flush() survives missing or broken os.get_blocking. + + Exercises the (AttributeError, OSError) handling added to fix + issue #48: on Python <3.11, os.get_blocking does not exist; on + Windows 3.11 console handles, it raises OSError. In both cases + the flush must still run. + """ + import io + from unittest import mock + from rawterm import Terminal + + # Construct a Terminal without entering raw mode -- we only need + # _flush() and _write_raw(), which operate on sys.stdout.buffer. + term = object.__new__(Terminal) + + # Redirect stdout.buffer to a BytesIO so we can verify output + buf = io.BytesIO() + fake_stdout = mock.MagicMock() + fake_stdout.buffer = buf + fake_stdout.fileno.return_value = 1 + + with mock.patch("sys.stdout", fake_stdout): + # Case 1: os.get_blocking raises AttributeError (Python <3.11) + # create=True: os.get_blocking may not exist on Windows Python <3.12 + with mock.patch("os.get_blocking", side_effect=AttributeError, create=True): + term._write_raw("hello") + term._flush() + assert buf.getvalue() == b"hello", "flush after AttributeError" + + buf.seek(0) + buf.truncate() + + # Case 2: os.get_blocking raises OSError (Windows console handle) + with mock.patch("os.get_blocking", side_effect=OSError, create=True): + term._write_raw(" world") + term._flush() + assert buf.getvalue() == b" world", "flush after OSError" + + buf.seek(0) + buf.truncate() + + # Case 3: os.get_blocking works normally (returns True) + with mock.patch("os.get_blocking", return_value=True, create=True): + term._write_raw("ok") + term._flush() + assert buf.getvalue() == b"ok", "flush with normal get_blocking" + + print("_flush() robustness checks passed") + + if __name__ == "__main__": + check_version() check_rawterm_units() + check_flush_robustness() check_windows_console() check_terminal_init() check_menuconfig_headless() diff --git a/menuconfig.py b/menuconfig.py index a8005f0..2da19dc 100755 --- a/menuconfig.py +++ b/menuconfig.py @@ -515,7 +515,9 @@ def menuconfig(kconf, headless=False): # Enter terminal mode via rawterm. _menuconfig() returns a string to print # on exit. - print(rawterm.run(_menuconfig)) + result = rawterm.run(_menuconfig) + if result is not None: + print(result) def _load_config(): diff --git a/rawterm.py b/rawterm.py index e1772be..49d89ee 100644 --- a/rawterm.py +++ b/rawterm.py @@ -636,10 +636,24 @@ def _init_windows(self): self._old_in_mode = wintypes.DWORD() kernel32.GetConsoleMode(self._stdin_handle, ctypes.byref(self._old_in_mode)) - # Enable VT100 output + # Enable VT100 output. DISABLE_NEWLINE_AUTO_RETURN (0x0008) + # prevents the console from inserting a CR before every LF, + # which causes cursor-positioning drift in TUI apps. Microsoft + # recommends setting both flags together for VT100 applications. + # + # Graceful fallback: if the combination is rejected (pre-1607 + # builds), retry with VT100 alone. ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004 - new_out = self._old_out_mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING - kernel32.SetConsoleMode(self._stdout_handle, new_out) + DISABLE_NEWLINE_AUTO_RETURN = 0x0008 + new_out = ( + self._old_out_mode.value + | ENABLE_VIRTUAL_TERMINAL_PROCESSING + | DISABLE_NEWLINE_AUTO_RETURN + ) + if not kernel32.SetConsoleMode(self._stdout_handle, new_out): + # Pre-1607: DISABLE_NEWLINE_AUTO_RETURN unsupported, VT100 only + new_out = self._old_out_mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING + kernel32.SetConsoleMode(self._stdout_handle, new_out) # Try VT100 input ENABLE_VIRTUAL_TERMINAL_INPUT = 0x0200 @@ -777,20 +791,47 @@ def _write_raw(self, s): pass def _flush(self): - """Flush stdout.""" + """Flush stdout. + + Terminal fds are always blocking, so the non-blocking dance is + purely defensive. Each step (get_blocking, set_blocking, flush, + restore) is isolated so that a failure in the probe never + prevents the flush from running. + + os.get_blocking raises AttributeError when unavailable (Windows + Python <3.12) and OSError on Windows console handles even when + present, so we catch both and default to was_blocking=True + (skip the set_blocking detour). + """ + was_blocking = True + fd = None try: - # Ensure blocking I/O for flush fd = sys.stdout.fileno() - was_blocking = os.get_blocking(fd) - if not was_blocking: - os.set_blocking(fd, True) + except OSError: + pass + + if fd is not None: try: - sys.stdout.buffer.flush() - finally: - if not was_blocking: - os.set_blocking(fd, False) + was_blocking = os.get_blocking(fd) + except (AttributeError, OSError): + was_blocking = True + + if not was_blocking: + try: + os.set_blocking(fd, True) + except OSError: + pass + + try: + sys.stdout.buffer.flush() except OSError: pass + finally: + if fd is not None and not was_blocking: + try: + os.set_blocking(fd, False) + except OSError: + pass def update(self): """Composite all regions and flush to terminal.