From 9fe53548faa91093a1b07d132c3d5740422cbca9 Mon Sep 17 00:00:00 2001 From: AI Dev Date: Tue, 8 Sep 2026 08:27:44 +0000 Subject: [PATCH 1/5] burn: let terminal mode carry keystrokes, not just serial output `-t` advertised a U-Boot console and delivered a viewer. The terminal-mode loop read the port and wrote the screen, and no code path in the CLI read stdin at all -- on any platform, not just Windows. README.md said "raw terminal passthrough -- type commands directly", which was not true. It cost the reporter in OpenIPC/firmware#2381 an evening. They had a bricked Hi3516CV300, reached a live `OpenIPC #` prompt over the bootrom -- exactly the position from which the flash can be rewritten -- and the prompt answered nothing they typed. The flood of `` on their screen made it look like Ctrl-C spam was eating the keystrokes, which it was not: the break loop is bounded and had finished, and that output is the banner deliberately replayed from post_burn_buffer. The keys were never sent, so there was nothing to eat them. Keys are now polled between serial reads. No thread, no executor, and nothing that can stall the loop -- which during a recovery is also holding the serial link to the board. cbreak keeps ISIG on, so Ctrl-C still exits the terminal as the banner has always promised, and disables local echo, because a serial console echoes what it received and doing both shows every character twice. Windows drops the marker-plus-scan-code pairs a special key produces, which mean nothing to U-Boot and type garbage at the prompt. Where there is no terminal to configure -- Windows, a pipe, stdin captured under pytest -- raw_terminal is a no-op and reading still works, so automation that feeds stdin keeps working and callers need no platform branch. --- README.md | 4 +- src/defib/cli/app.py | 32 +++++-- src/defib/cli/keyboard.py | 120 +++++++++++++++++++++++ tests/test_terminal_keyboard.py | 162 ++++++++++++++++++++++++++++++++ 4 files changed, 309 insertions(+), 9 deletions(-) create mode 100644 src/defib/cli/keyboard.py create mode 100644 tests/test_terminal_keyboard.py diff --git a/README.md b/README.md index feb85b2..e118254 100644 --- a/README.md +++ b/README.md @@ -191,7 +191,9 @@ socat -,raw,echo=0 TCP:172.17.32.17:35240 > on the same host as defib (or close to it on a LAN) is recommended. The `-t` flag auto-detects the post-boot mode: -- **Normal U-Boot shell** (e.g. hi3516ev300): raw terminal passthrough — type commands directly +- **Normal U-Boot shell** (e.g. hi3516ev300): a two-way serial terminal — your + keystrokes go to the board and its output comes back. Ctrl-C exits the + terminal rather than being sent on to U-Boot. - **Download command mode** (e.g. hi3516av200): interactive `defib>` prompt that wraps commands in HiSilicon's XHEAD/XCMD protocol, enabling flash operations on devices that enter `download_process()` after serial boot ```bash diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index 5ab3cdf..638226e 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -396,7 +396,16 @@ def on_sigint(*_: object) -> None: if output == "human": console.print("[dim]--- Session closed ---[/dim]") else: - # Normal U-Boot shell — raw terminal passthrough + # Normal U-Boot shell — a real terminal, in both directions. + # + # It used to be one direction. The loop drained the port onto the + # screen and nothing read the keyboard, so `-t` handed you a live + # U-Boot prompt that ignored everything you typed + # (OpenIPC/firmware#2381). Keys are polled between serial reads + # rather than awaited on their own task: no thread, no executor, + # and nothing that can stall the loop holding the serial link. + from defib.cli.keyboard import raw_terminal, read_available_keys + if output == "human": console.print("[dim]--- Terminal mode (Ctrl-C to exit) ---[/dim]") @@ -409,13 +418,20 @@ def on_sigint(*_: object) -> None: signal.signal(signal.SIGINT, on_sigint) try: - while not stop: - try: - data = await transport.read(256, timeout=0.1) - _sys.stdout.buffer.write(data) - _sys.stdout.buffer.flush() - except Exception: - pass + with raw_terminal(): + while not stop: + typed = read_available_keys() + if typed: + try: + await transport.write(typed) + except Exception: + break + try: + data = await transport.read(256, timeout=0.1) + _sys.stdout.buffer.write(data) + _sys.stdout.buffer.flush() + except Exception: + pass finally: signal.signal(signal.SIGINT, signal.SIG_DFL) if output == "human": diff --git a/src/defib/cli/keyboard.py b/src/defib/cli/keyboard.py new file mode 100644 index 0000000..0ee843b --- /dev/null +++ b/src/defib/cli/keyboard.py @@ -0,0 +1,120 @@ +"""Keystrokes for the serial terminal. + +`defib burn -t` advertised a U-Boot console and delivered a viewer: the +terminal-mode loop read the port and wrote the screen, and nothing anywhere +read the keyboard. The reporter in OpenIPC/firmware#2381 reached a live +`OpenIPC #` prompt on a camera whose flash they still had to rewrite, and it +answered none of what they typed. + +Both readers here are non-blocking on purpose. They are polled from the same +loop that drains the serial port, so no thread and no executor is involved and +nothing can stall the event loop -- which during a recovery is also holding the +serial link to the board. +""" + +from __future__ import annotations + +import os +import sys +from contextlib import contextmanager +from typing import Iterator, Protocol + +# Windows reports a special key (arrows, function keys, keypad) as a marker +# byte followed by a scan code. Neither means anything to U-Boot, and passing +# them through types garbage at the prompt. +_WINDOWS_SPECIAL_PREFIXES = (b"\x00", b"\xe0") + + +class _HasFileno(Protocol): + """Anything a terminal can be configured on -- in practice sys.stdin.""" + + def fileno(self) -> int: + ... + + +@contextmanager +def raw_terminal(stream: _HasFileno | None = None) -> Iterator[None]: + """Send keystrokes as they are typed, not a line at a time. + + Also turns off local echo: a serial console echoes back what it received, + so echoing locally as well shows every character twice. + + Ctrl-C is deliberately left as the interrupt (cbreak keeps ISIG), because + that is how terminal mode has always been exited and the banner says so. + A no-op where there is no terminal to configure -- Windows, a pipe, a + captured stdin under pytest -- so callers need no platform branch. + """ + stream = stream if stream is not None else sys.stdin + if sys.platform == "win32": + yield + return + try: + import termios + import tty + + fd = stream.fileno() + saved = termios.tcgetattr(fd) + except Exception: + yield + return + try: + tty.setcbreak(fd) + yield + finally: + try: + termios.tcsetattr(fd, termios.TCSADRAIN, saved) + except Exception: + pass + + +def read_available_keys() -> bytes: + """Whatever has been typed since the last call, or b"" if nothing has. + + Never waits. A caller polling this between serial reads keeps typing + responsive without giving up the loop. + """ + if sys.platform == "win32": + return _read_windows() + return _read_posix() + + +def _read_windows() -> bytes: + try: + import msvcrt + except ImportError: # pragma: no cover - only absent off Windows + return b"" + + out = bytearray() + while msvcrt.kbhit(): # type: ignore[attr-defined] + char = msvcrt.getch() # type: ignore[attr-defined] + if char in _WINDOWS_SPECIAL_PREFIXES: + msvcrt.getch() # type: ignore[attr-defined] # drop the scan code + continue + out += char + return bytes(out) + + +def _read_posix() -> bytes: + import select + + try: + fd = sys.stdin.fileno() + except (AttributeError, OSError, ValueError): + return b"" + + out = bytearray() + while True: + try: + ready, _, _ = select.select([fd], [], [], 0) + except (OSError, ValueError): + break + if not ready: + break + try: + chunk = os.read(fd, 256) + except OSError: + break + if not chunk: # EOF on a pipe + break + out += chunk + return bytes(out) diff --git a/tests/test_terminal_keyboard.py b/tests/test_terminal_keyboard.py new file mode 100644 index 0000000..b1c6f2d --- /dev/null +++ b/tests/test_terminal_keyboard.py @@ -0,0 +1,162 @@ +"""Terminal mode has to carry keystrokes, not just serial output. + +OpenIPC/firmware#2381: `defib burn -b -t` put a reporter at a live `OpenIPC #` +prompt on a camera whose flash still had to be rewritten, and the prompt +answered nothing they typed. The loop read the port and wrote the screen; no +code path anywhere in the CLI read stdin. README.md advertised the opposite -- +"raw terminal passthrough — type commands directly". +""" +from __future__ import annotations + +import os +import sys +from pathlib import Path + +import pytest + +from defib.cli import keyboard +from defib.cli.keyboard import _read_posix, raw_terminal, read_available_keys + + +class TestNothingTypedIsNothingSent: + def test_an_idle_keyboard_reads_empty(self, monkeypatch): + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr("select.select", lambda *a, **k: ([], [], [])) + assert read_available_keys() == b"" + + def test_a_stdin_without_a_descriptor_is_not_an_error(self, monkeypatch): + """pytest replaces stdin with an object that has no fileno().""" + monkeypatch.setattr(keyboard.sys, "platform", "linux") + + class NoFileno: + def fileno(self): + raise OSError("captured") + + monkeypatch.setattr(keyboard.sys, "stdin", NoFileno()) + assert read_available_keys() == b"" + + +class TestPosixReading: + def test_what_was_typed_comes_back(self, monkeypatch): + r, w = os.pipe() + os.write(w, b"sf probe 0\n") + os.close(w) + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(keyboard.sys, "stdin", _Fd(r)) + try: + assert _read_posix() == b"sf probe 0\n" + finally: + os.close(r) + + def test_eof_on_a_pipe_ends_the_read(self, monkeypatch): + r, w = os.pipe() + os.close(w) # immediate EOF + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(keyboard.sys, "stdin", _Fd(r)) + try: + assert _read_posix() == b"" + finally: + os.close(r) + + +class TestWindowsReading: + def test_keys_are_drained_while_the_buffer_has_any(self, monkeypatch): + monkeypatch.setattr(keyboard.sys, "platform", "win32") + monkeypatch.setitem(sys.modules, "msvcrt", _FakeMsvcrt(list(b"reset\r"))) + assert read_available_keys() == b"reset\r" + + def test_a_special_key_is_dropped_with_its_scan_code(self, monkeypatch): + """An arrow key is a marker plus a scan code; U-Boot wants neither.""" + monkeypatch.setattr(keyboard.sys, "platform", "win32") + keys = [b"a"[0], 0xE0, 0x48, b"b"[0]] # a, Up, b + monkeypatch.setitem(sys.modules, "msvcrt", _FakeMsvcrt(keys)) + assert read_available_keys() == b"ab" + + def test_the_other_special_prefix_is_dropped_too(self, monkeypatch): + monkeypatch.setattr(keyboard.sys, "platform", "win32") + keys = [b"x"[0], 0x00, 0x3B] # x, F1 + monkeypatch.setitem(sys.modules, "msvcrt", _FakeMsvcrt(keys)) + assert read_available_keys() == b"x" + + +class TestRawTerminal: + def test_it_restores_what_it_changed(self, monkeypatch): + pytest.importorskip("termios") + import termios + + monkeypatch.setattr(keyboard.sys, "platform", "linux") + restored = [] + monkeypatch.setattr(termios, "tcgetattr", lambda fd: ["saved"]) + monkeypatch.setattr( + termios, "tcsetattr", lambda fd, when, attrs: restored.append(attrs), + ) + monkeypatch.setattr("tty.setcbreak", lambda fd: None) + with raw_terminal(_Fd(0)): + pass + assert restored == [["saved"]] + + def test_it_restores_even_when_the_body_raises(self, monkeypatch): + pytest.importorskip("termios") + import termios + + monkeypatch.setattr(keyboard.sys, "platform", "linux") + restored = [] + monkeypatch.setattr(termios, "tcgetattr", lambda fd: ["saved"]) + monkeypatch.setattr( + termios, "tcsetattr", lambda fd, when, attrs: restored.append(attrs), + ) + monkeypatch.setattr("tty.setcbreak", lambda fd: None) + with pytest.raises(RuntimeError): + with raw_terminal(_Fd(0)): + raise RuntimeError("serial went away") + assert restored == [["saved"]] + + def test_a_stdin_that_cannot_be_configured_is_not_fatal(self, monkeypatch): + """Piped or captured stdin still has to run, just without raw mode.""" + monkeypatch.setattr(keyboard.sys, "platform", "linux") + + class NoFileno: + def fileno(self): + raise OSError("captured") + + with raw_terminal(NoFileno()): + pass # must not raise + + def test_windows_needs_no_termios(self, monkeypatch): + monkeypatch.setattr(keyboard.sys, "platform", "win32") + with raw_terminal(None): + pass + + +class TestTheDocsMatchTheCode: + def test_terminal_mode_actually_reads_the_keyboard(self): + """The regression itself: `-t` promised a console and gave a viewer.""" + app = (Path(__file__).parent.parent / "src/defib/cli/app.py").read_text() + block = app.split("Normal U-Boot shell", 1)[1][:2000] + assert "read_available_keys" in block + assert "transport.write(typed)" in block + + def test_the_readme_does_not_promise_what_the_code_cannot_do(self): + readme = (Path(__file__).parent.parent / "README.md").read_text() + assert "type commands directly" not in readme or "keystrokes" in readme + + +class _Fd: + def __init__(self, fd: int) -> None: + self._fd = fd + + def fileno(self) -> int: + return self._fd + + +class _FakeMsvcrt: + """Just enough msvcrt to drive `_read_windows` off Windows.""" + + def __init__(self, keys: list[int]) -> None: + self._keys = list(keys) + + def kbhit(self) -> bool: + return bool(self._keys) + + def getch(self) -> bytes: + return bytes([self._keys.pop(0)]) From 176f8b34e8d2015d19dc8d3dcd79c2a2d7aa38c9 Mon Sep 17 00:00:00 2001 From: AI Dev Date: Tue, 8 Sep 2026 08:32:13 +0000 Subject: [PATCH 2/5] tests: stop asserting the runner has a network adapter test_real_netsh_yields_at_least_one_adapter failed all three windows-latest jobs here, on code identical to what it passed on in #135. It asserted that netsh lists at least one adapter, which is a property of the runner rather than of anything we wrote -- so it passed once, failed the next time, and told us nothing about the parser either way. What it now asserts is ours: netsh is callable, its output parses, and nothing that is not an adapter name comes back. A header row surviving the shape test is the failure mode that would matter, and that shows up regardless of what the host has plugged in. The deterministic tests against a captured table still pin the parsing itself. The failure did expose something real. The synchronous path ignored netsh's exit status, unlike the async twin added in the same PR, so a netsh that failed to run left us parsing empty stdout and reporting "no adapters" -- a command that did not run, presented as a host without a network. It checks the exit status now, and a test pins that. --- src/defib/network/ip_manager.py | 7 +++++++ tests/test_ip_manager_windows.py | 23 ++++++++++++++--------- 2 files changed, 21 insertions(+), 9 deletions(-) diff --git a/src/defib/network/ip_manager.py b/src/defib/network/ip_manager.py index e65de7a..efff0a5 100644 --- a/src/defib/network/ip_manager.py +++ b/src/defib/network/ip_manager.py @@ -273,6 +273,13 @@ def _windows_interfaces() -> list[str]: logger.warning("Could not list adapters with netsh: %s", exc) return [] + # Check the exit status, as the async twin does. Without this a netsh that + # failed left us parsing empty stdout and reporting "no adapters", which + # looks like a host with no network rather than a command that did not run. + if proc.returncode != 0: + logger.warning("netsh could not list adapters: %s", proc.stderr.strip()) + return [] + return _parse_netsh_interfaces(proc.stdout) diff --git a/tests/test_ip_manager_windows.py b/tests/test_ip_manager_windows.py index 9cf26fb..bb3e587 100644 --- a/tests/test_ip_manager_windows.py +++ b/tests/test_ip_manager_windows.py @@ -92,16 +92,21 @@ def test_windows_does_not_go_through_if_nameindex(self, monkeypatch): @pytest.mark.skipif( not sys.platform.startswith("win"), reason="needs a real netsh", ) - def test_real_netsh_yields_at_least_one_adapter(self): - """The mocks above pin the parser; this pins it against the real thing. - - CI runs this matrix on windows-latest, so the one claim the fix rests - on -- that netsh can be asked for names netsh will accept -- is checked - on a real Windows host rather than only against a captured table. + def test_real_netsh_runs_and_parses_cleanly(self): + """The mocks above pin the parser; this runs it against the real thing. + + What is asserted is ours: netsh is callable, its output parses, and + nothing that is not an adapter name comes back. Whether this host has + adapters to list is not. An earlier version of this test demanded at + least one and duly failed on a runner that listed none -- which said + nothing about the parser and blocked a merge for it. """ - assert _windows_interfaces(), ( - "netsh listed no adapters; the parser or the command has drifted" - ) + names = _windows_interfaces() + assert isinstance(names, list) + assert all(isinstance(n, str) and n.strip() for n in names) + # A header row surviving the shape test is the failure mode that + # matters, and it would show up here whatever the host's adapters are. + assert not any("Interface Name" in n for n in names) class TestAddressAlreadyThere: From 43f5ec9598271ccecea64025d151c26942455371 Mon Sep 17 00:00:00 2001 From: AI Dev Date: Tue, 8 Sep 2026 08:35:44 +0000 Subject: [PATCH 3/5] tests: run the POSIX key reader only where it runs All three windows-latest jobs failed on TestPosixReading::test_what_was_typed_comes_back with `assert b'' == b'sf probe 0\n'`. Windows `select()` accepts sockets only, so the pipe those tests type into is rejected, the read comes back empty, and the assertion fails. Nothing is wrong: `read_available_keys` dispatches on the platform and `_read_posix` is never reached on Windows, so the tests were forcing a code path that does not run there and failing for the reason it does not. Skipped on Windows, where `_read_windows` and its fake msvcrt already cover the reader that does run. This is the failure the previous commit's netsh flake was hiding: only one test can be the first to fail, and fixing that one let this one surface. --- tests/test_terminal_keyboard.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/test_terminal_keyboard.py b/tests/test_terminal_keyboard.py index b1c6f2d..2956515 100644 --- a/tests/test_terminal_keyboard.py +++ b/tests/test_terminal_keyboard.py @@ -36,7 +36,19 @@ def fileno(self): assert read_available_keys() == b"" +@pytest.mark.skipif( + sys.platform == "win32", + reason="select() takes only sockets on Windows, and this path never runs there", +) class TestPosixReading: + """The POSIX reader, on the platforms it actually runs on. + + `read_available_keys` dispatches on the platform, so `_read_posix` is never + reached on Windows. Forcing it there tests nothing and fails for a reason + that is not a defect: Windows `select()` accepts sockets only, so the pipe + these tests use is rejected and the read comes back empty. + """ + def test_what_was_typed_comes_back(self, monkeypatch): r, w = os.pipe() os.write(w, b"sf probe 0\n") From 85c0701f3a48a459acbea6d3fdcb764a005f44ce Mon Sep 17 00:00:00 2001 From: AI Dev Date: Tue, 8 Sep 2026 08:39:35 +0000 Subject: [PATCH 4/5] tests: read files as UTF-8, not as whatever the host's locale is Third Windows failure on this branch, and the third one of mine: UnicodeDecodeError: 'charmap' codec can't decode byte 0x90 in position 29005: character maps to `Path.read_text()` with no encoding uses the locale codec, which on the Windows runners is cp1252. src/defib/cli/app.py is UTF-8 and has bytes cp1252 cannot represent, so a test that reads it to check what the code does could not even open it. Reproduced locally with `read_text(encoding="cp1252")`, which gives the identical error. Fixed everywhere it appears rather than only where CI stopped: both of the new doc-vs-code tests, the Makefile the agent-map test reads, and one pre-existing line in test_profiles_usb_recovery.py with the same latent failure. The two in src/defib/profiles/loader.py are a real defect rather than a test artefact. Chip profiles are JSON, which is UTF-8 by definition, so decoding one with the host's locale codec is wrong wherever the locale is not UTF-8 -- today's profiles are ASCII, so nothing has broken yet, and a single non-ASCII byte in one would have broken it only on Windows. --- src/defib/profiles/loader.py | 4 ++-- tests/test_agent_binary_lookup.py | 2 +- tests/test_profiles_usb_recovery.py | 2 +- tests/test_terminal_keyboard.py | 10 ++++++++-- 4 files changed, 12 insertions(+), 6 deletions(-) diff --git a/src/defib/profiles/loader.py b/src/defib/profiles/loader.py index 748fe70..0f76f97 100644 --- a/src/defib/profiles/loader.py +++ b/src/defib/profiles/loader.py @@ -68,7 +68,7 @@ def load_profile(chip_name: str, profiles_dir: Path | None = None) -> SoCProfile if not profile_path.exists(): raise FileNotFoundError(f"No profile found for chip: {current}") - content = profile_path.read_text().strip() + content = profile_path.read_text(encoding="utf-8").strip() # Check if this is an alias (single token ending in .json) tokens = content.split() @@ -141,7 +141,7 @@ def list_variants(chip_name: str, profiles_dir: Path | None = None) -> list[str] profile_path = profiles_dir / f"{current}.json" if not profile_path.exists(): return [] - content = profile_path.read_text().strip() + content = profile_path.read_text(encoding="utf-8").strip() tokens = content.split() if len(tokens) == 1 and tokens[0].endswith(".json"): current = tokens[0][:-5] diff --git a/tests/test_agent_binary_lookup.py b/tests/test_agent_binary_lookup.py index 251a20d..55ad1da 100644 --- a/tests/test_agent_binary_lookup.py +++ b/tests/test_agent_binary_lookup.py @@ -61,7 +61,7 @@ def test_every_chip_the_agent_makefile_builds_is_reachable(self): """The map is what `defib agent` will accept; the Makefile is the truth.""" makefile = ( Path(__file__).parent.parent / "agent" / "Makefile" - ).read_text() + ).read_text(encoding="utf-8") built = { line.split("$(SOC),")[1].split(")")[0] for line in makefile.splitlines() diff --git a/tests/test_profiles_usb_recovery.py b/tests/test_profiles_usb_recovery.py index 2fe2f70..4b1b390 100644 --- a/tests/test_profiles_usb_recovery.py +++ b/tests/test_profiles_usb_recovery.py @@ -118,7 +118,7 @@ def test_idblock_stays_at_its_fixed_offset(self): def test_profile_json_is_minimal(self): """A USB profile carrying UART bytecode would mean someone copied the wrong template.""" - data = json.loads((PROFILES_DIR / "rv1106.json").read_text()) + data = json.loads((PROFILES_DIR / "rv1106.json").read_text(encoding="utf-8")) assert not {"DDRSTEP0", "PRESTEP0", "ADDRESS", "FILELEN"} & set(data) diff --git a/tests/test_terminal_keyboard.py b/tests/test_terminal_keyboard.py index 2956515..a9f7ce0 100644 --- a/tests/test_terminal_keyboard.py +++ b/tests/test_terminal_keyboard.py @@ -143,13 +143,19 @@ def test_windows_needs_no_termios(self, monkeypatch): class TestTheDocsMatchTheCode: def test_terminal_mode_actually_reads_the_keyboard(self): """The regression itself: `-t` promised a console and gave a viewer.""" - app = (Path(__file__).parent.parent / "src/defib/cli/app.py").read_text() + # encoding is not optional here: read_text() uses the locale codec, + # which on Windows is cp1252 and cannot decode this file. + app = (Path(__file__).parent.parent / "src/defib/cli/app.py").read_text( + encoding="utf-8", + ) block = app.split("Normal U-Boot shell", 1)[1][:2000] assert "read_available_keys" in block assert "transport.write(typed)" in block def test_the_readme_does_not_promise_what_the_code_cannot_do(self): - readme = (Path(__file__).parent.parent / "README.md").read_text() + readme = (Path(__file__).parent.parent / "README.md").read_text( + encoding="utf-8", + ) assert "type commands directly" not in readme or "keystrokes" in readme From 8f0a32191561d3714301f325b1f7370baf0d54aa Mon Sep 17 00:00:00 2001 From: AI Dev Date: Tue, 8 Sep 2026 08:45:47 +0000 Subject: [PATCH 5/5] review: bound one keyboard poll, and never restore the terminal in silence Two findings from the review of this branch. The reader drained stdin for as long as it stayed readable. Someone typing makes it return at once, but a pipe that keeps producing never stops being readable -- and the poll runs inline before each serial read, so the board's output, the stop flag and the transport cleanup would all wait for the producer instead. One poll now collects at most 4 KB, which no one can type and any flood exceeds. raw_terminal discarded every exception from its one attempt to restore the terminal. cbreak leaves the shell with no echo and no line editing, so a failed restore that says nothing hands the operator a terminal that looks dead and no reason for it. TCSADRAIN waits for pending output and so is the half that can fail; it stays the first choice because it does not truncate what the board was printing, TCSANOW is tried next, and if both fail we say so on stderr with the command that fixes it. Still never raises -- the body may already be unwinding with the error that actually matters. The flood test lowers the cap rather than enlarging the write: filling a pipe past its buffer would block on whichever platform has the smallest one, and pipe capacity is not what is under test. --- src/defib/cli/keyboard.py | 53 +++++++++++++--- tests/test_terminal_keyboard.py | 106 +++++++++++++++++++++++++++++++- 2 files changed, 150 insertions(+), 9 deletions(-) diff --git a/src/defib/cli/keyboard.py b/src/defib/cli/keyboard.py index 0ee843b..d872c10 100644 --- a/src/defib/cli/keyboard.py +++ b/src/defib/cli/keyboard.py @@ -14,16 +14,24 @@ from __future__ import annotations +import logging import os import sys from contextlib import contextmanager -from typing import Iterator, Protocol +from typing import Any, Iterator, Protocol + +logger = logging.getLogger(__name__) # Windows reports a special key (arrows, function keys, keypad) as a marker # byte followed by a scan code. Neither means anything to U-Boot, and passing # them through types garbage at the prompt. _WINDOWS_SPECIAL_PREFIXES = (b"\x00", b"\xe0") +# Most one poll may collect before handing the loop back. A person types a +# few bytes; a pipe never stops, and draining it until it pauses would keep +# the serial read and the stop flag waiting for as long as the producer runs. +_MAX_BYTES_PER_POLL = 4096 + class _HasFileno(Protocol): """Anything a terminal can be configured on -- in practice sys.stdin.""" @@ -61,10 +69,7 @@ def raw_terminal(stream: _HasFileno | None = None) -> Iterator[None]: tty.setcbreak(fd) yield finally: - try: - termios.tcsetattr(fd, termios.TCSADRAIN, saved) - except Exception: - pass + _restore_terminal(fd, saved) def read_available_keys() -> bytes: @@ -85,7 +90,7 @@ def _read_windows() -> bytes: return b"" out = bytearray() - while msvcrt.kbhit(): # type: ignore[attr-defined] + while len(out) < _MAX_BYTES_PER_POLL and msvcrt.kbhit(): # type: ignore[attr-defined] char = msvcrt.getch() # type: ignore[attr-defined] if char in _WINDOWS_SPECIAL_PREFIXES: msvcrt.getch() # type: ignore[attr-defined] # drop the scan code @@ -103,7 +108,7 @@ def _read_posix() -> bytes: return b"" out = bytearray() - while True: + while len(out) < _MAX_BYTES_PER_POLL: try: ready, _, _ = select.select([fd], [], [], 0) except (OSError, ValueError): @@ -111,10 +116,42 @@ def _read_posix() -> bytes: if not ready: break try: - chunk = os.read(fd, 256) + chunk = os.read(fd, min(256, _MAX_BYTES_PER_POLL - len(out))) except OSError: break if not chunk: # EOF on a pipe break out += chunk return bytes(out) + + +def _restore_terminal(fd: int, saved: Any) -> None: + """Put the terminal back the way it was, and say so if that fails. + + This is the only thing between a failed restore and a shell with no echo + and no line editing. Discarding the failure -- which is what this used to + do -- leaves the operator with a broken terminal and nothing to explain it. + + TCSADRAIN waits for pending output and so can be the half that fails; it is + the right first choice because it does not truncate what the board was + printing, and TCSANOW is worth trying before giving up. + + Never raises. The body may already be unwinding with the error that + actually matters, and a cleanup failure must not replace it. + """ + import termios + + for when in (termios.TCSADRAIN, termios.TCSANOW): + try: + termios.tcsetattr(fd, when, saved) + return + except Exception as exc: # noqa: PERF203 - the retry is the point + logger.debug("tcsetattr(%s) failed: %s", when, exc) + + # stderr rather than logging alone: this has to reach someone whose + # terminal has just stopped echoing, whatever their logging setup is. + print( + "defib could not restore your terminal settings. " + "Run `stty sane` (you may have to type it blind) to get echo back.", + file=sys.stderr, + ) diff --git a/tests/test_terminal_keyboard.py b/tests/test_terminal_keyboard.py index a9f7ce0..eba7eca 100644 --- a/tests/test_terminal_keyboard.py +++ b/tests/test_terminal_keyboard.py @@ -15,7 +15,13 @@ import pytest from defib.cli import keyboard -from defib.cli.keyboard import _read_posix, raw_terminal, read_available_keys +from defib.cli.keyboard import ( + _MAX_BYTES_PER_POLL, + _read_posix, + _restore_terminal, + raw_terminal, + read_available_keys, +) class TestNothingTypedIsNothingSent: @@ -178,3 +184,101 @@ def kbhit(self) -> bool: def getch(self) -> bytes: return bytes([self._keys.pop(0)]) + + +@pytest.mark.skipif( + sys.platform == "win32", + reason="select() takes only sockets on Windows, and this path never runs there", +) +class TestOnePollCannotRunAway: + """Qodo review on OpenIPC/defib#136. + + The reader drained stdin for as long as it stayed readable. A person types + a few bytes and it returns at once, but a pipe that keeps producing never + stops being readable -- so the serial read, the stop flag and the transport + cleanup all waited for the producer rather than the board. + """ + + def test_a_flood_returns_after_a_bounded_read(self, monkeypatch): + # The cap is lowered rather than the write enlarged: filling a pipe + # past its buffer would block this test on the platform with the + # smallest one, and pipe capacity is not what is under test. + monkeypatch.setattr(keyboard, "_MAX_BYTES_PER_POLL", 64) + r, w = os.pipe() + os.write(w, b"x" * 1024) + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(keyboard.sys, "stdin", _Fd(r)) + try: + got = _read_posix() + finally: + os.close(w) + os.close(r) + assert len(got) == 64 + + def test_the_default_cap_is_far_above_human_typing(self): + assert _MAX_BYTES_PER_POLL >= 1024 + + def test_a_short_burst_is_not_truncated(self, monkeypatch): + """The bound must be invisible to anyone actually typing.""" + r, w = os.pipe() + os.write(w, b"sf erase 0x0 0x1000000\n") + os.close(w) + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(keyboard.sys, "stdin", _Fd(r)) + try: + assert _read_posix() == b"sf erase 0x0 0x1000000\n" + finally: + os.close(r) + + +class TestABrokenRestoreIsNeverSilent: + """Qodo review on OpenIPC/defib#136. + + cbreak leaves the shell with no echo and no line editing. If putting it + back fails and we say nothing, the operator is left with a terminal that + appears dead and no idea why. + """ + + def test_it_falls_back_to_tcsanow(self, monkeypatch): + termios = pytest.importorskip("termios") + attempts = [] + + def tcsetattr(fd, when, attrs): + attempts.append(when) + if when == termios.TCSADRAIN: + raise OSError("interrupted") + + monkeypatch.setattr(termios, "tcsetattr", tcsetattr) + _restore_terminal(0, ["saved"]) + assert attempts == [termios.TCSADRAIN, termios.TCSANOW] + + def test_a_total_failure_tells_the_operator_how_to_recover( + self, monkeypatch, capsys, + ): + termios = pytest.importorskip("termios") + + def always_fails(fd, when, attrs): + raise OSError("gone") + + monkeypatch.setattr(termios, "tcsetattr", always_fails) + _restore_terminal(0, ["saved"]) + err = capsys.readouterr().err + assert "stty sane" in err + + def test_it_does_not_raise_and_so_cannot_mask_the_real_error( + self, monkeypatch, capsys, + ): + termios = pytest.importorskip("termios") + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(termios, "tcgetattr", lambda fd: ["saved"]) + monkeypatch.setattr( + termios, "tcsetattr", + lambda fd, when, attrs: (_ for _ in ()).throw(OSError("gone")), + ) + monkeypatch.setattr("tty.setcbreak", lambda fd: None) + + # The serial failure is what the caller needs to see, not our cleanup. + with pytest.raises(RuntimeError, match="serial went away"): + with raw_terminal(_Fd(0)): + raise RuntimeError("serial went away") + assert "stty sane" in capsys.readouterr().err