From 453b78a235f867559cbeed4399626277e975729d Mon Sep 17 00:00:00 2001 From: David Samson Date: Wed, 27 May 2026 15:11:59 -0400 Subject: [PATCH 1/4] initial stab at reworking in pure python to support multiple OSs (instead of just linux). Also improved multi-line inputs to have better navigation control --- README.md | 66 ++++++++++--- easyrepl/__init__.py | 4 +- easyrepl/buffer.py | 201 ++++++++++++++++++++++++++++++++++++++ easyrepl/editor.py | 228 +++++++++++++++++++++++++++++++++++++++++++ easyrepl/history.py | 106 ++++++++++++++++++++ easyrepl/keys.py | 162 ++++++++++++++++++++++++++++++ easyrepl/render.py | 107 ++++++++++++++++++++ easyrepl/repl.py | 204 ++++++++++++++++---------------------- easyrepl/terminal.py | 95 ++++++++++++++++++ pyproject.toml | 36 ++++--- uv.lock | 36 +++++++ 11 files changed, 1096 insertions(+), 149 deletions(-) create mode 100644 easyrepl/buffer.py create mode 100644 easyrepl/editor.py create mode 100644 easyrepl/history.py create mode 100644 easyrepl/keys.py create mode 100644 easyrepl/render.py create mode 100644 easyrepl/terminal.py create mode 100644 uv.lock diff --git a/README.md b/README.md index 9b9dd99..f52f86b 100644 --- a/README.md +++ b/README.md @@ -3,13 +3,21 @@ A simple python class for creating Read Evaluate Print Line (REPL) interfaces. ## Requirements -This module requires Python 3.7 or higher. +This module requires Python 3.8 or higher. -Additionally this library makes use of the [GNU readline Interface](https://docs.python.org/3/library/readline.html), so it will only work on Unix based systems. +EasyREPL is a pure-Python line editor with no dependency on GNU readline. It works +on Linux, macOS, and Windows (Windows uses [colorama](https://pypi.org/project/colorama/) +to enable ANSI escape-sequence processing on the console). + +## Installation + +```bash +pip install easyrepl +``` ## Usage -This module exposes the `REPL` class which can be used to quickly create a REPL interface. REPL will read in a line of user input via a custom input function that allows you to edit the text by moving the cursor with the arrow keys, as well as view the history of previous inputs. +This module exposes the `REPL` class which can be used to quickly create a REPL interface. `REPL` reads a line of user input via a built-in line editor that allows you to move the cursor with the arrow keys, navigate previous inputs, and edit multi-line input. ```python from easyrepl import REPL @@ -29,25 +37,28 @@ world >>> ``` -The input allows common terminal shortcuts like: -- **Ctrl-D**: exit REPL +The input supports the following key bindings: +- **Ctrl-D**: exit REPL (on an empty buffer); otherwise forward-delete - **Ctrl-L**: clear screen -- **Ctrl-R**: search history +- **Ctrl-R**: reverse-i-search through history - **Left/Right Arrow**: move cursor left/right -- **Up/Down Arrow**: previous/next history -- **Ctrl-A**: move cursor to beginning of line -- **Ctrl-E**: move cursor to end of line -- **Alt-B**: move cursor backward one word -- **Alt-F**: move cursor forward one word +- **Up/Down Arrow**: move up/down within a multi-line input; at the top/bottom edge, navigates previous/next history entry +- **Home / Ctrl-A**: move cursor to beginning of line +- **End / Ctrl-E**: move cursor to end of line +- **Alt-B / Ctrl-Left**: move cursor backward one word +- **Alt-F / Ctrl-Right**: move cursor forward one word - **Ctrl-K**: delete from cursor to end of line - **Ctrl-U**: delete from cursor to beginning of line - **Ctrl-W**: delete from cursor to beginning of word - **Alt-D**: delete from cursor to end of word -- **Ctrl-C**: no operation -- etc. +- **Ctrl-C**: abort current input (or quit, if `ctrl_c_quit=True`) -Additionally, multi-line input can be achieved by starting a line with triple quotes (`"""` or `'''`), and ending the final line with a matching triple quote. Triple quotes in the middle of a line will have no effect. +## Multi-line input + +Multi-line input can be entered by starting a line with triple quotes (`"""` or `'''`), +and ending the final line with a matching triple quote. Triple quotes in the middle +of a line have no effect. ``` @@ -64,4 +75,29 @@ at the end of a line >>> ``` -Note that a single newline will be stripped from the beginning and end of the input if present. \ No newline at end of file +A single newline is stripped from the beginning and end of the result if present. + +Within a multi-line block, the Up and Down arrow keys move the cursor between +lines of the buffer. When the cursor is on the first row, Up navigates to the +previous history entry; when it's on the last row, Down navigates to the next +history entry. Pressing Enter inserts a newline unless the cursor is at the end +of the buffer and the block is closed (the line then gets submitted). + +## API + +```python +REPL(*, prompt='>>> ', continuation_prompt='... ', history=None, + dedup_history=True, ctrl_c_quit=False) +``` + +- `prompt`: prompt rendered before the first line of each input. +- `continuation_prompt`: prompt rendered on every line of a multi-line input after the first. +- `history`: path to a file used to load and persist history; `None` keeps history in-memory only. +- `dedup_history`: when True, appending an entry removes any earlier identical entries. +- `ctrl_c_quit`: when True, Ctrl-C re-raises `KeyboardInterrupt` to terminate the REPL. + +```python +readl(*, prompt='', ctrl_c_quit=True, **kwargs) -> str +``` + +Read a single line using the REPL editor. diff --git a/easyrepl/__init__.py b/easyrepl/__init__.py index b26461a..0aaf829 100644 --- a/easyrepl/__init__.py +++ b/easyrepl/__init__.py @@ -1 +1,3 @@ -from .repl import REPL, readl \ No newline at end of file +from .repl import REPL, readl + +__all__ = ['REPL', 'readl'] diff --git a/easyrepl/buffer.py b/easyrepl/buffer.py new file mode 100644 index 0000000..eaa75ca --- /dev/null +++ b/easyrepl/buffer.py @@ -0,0 +1,201 @@ +from dataclasses import dataclass, field +from typing import List + + +def _is_word_char(c: str) -> bool: + return c.isalnum() or c == '_' + + +@dataclass +class Buffer: + """A multi-line text buffer with a cursor. + + `lines` always has at least one entry. `row`/`col` index into it. `desired_col` + tracks the column the cursor "wants" to be in across vertical movement, so that + moving up from a short line into a long line preserves the original column. + """ + + lines: List[str] = field(default_factory=lambda: [""]) + row: int = 0 + col: int = 0 + desired_col: int = 0 + + @property + def text(self) -> str: + return '\n'.join(self.lines) + + @classmethod + def from_text(cls, text: str) -> 'Buffer': + lines = text.split('\n') if text else [""] + buf = cls(lines=lines) + buf.move_end_of_buffer() + return buf + + def replace_with(self, other: 'Buffer') -> None: + self.lines = list(other.lines) + self.row = other.row + self.col = other.col + self.desired_col = other.desired_col + + def insert_char(self, c: str) -> None: + line = self.lines[self.row] + self.lines[self.row] = line[:self.col] + c + line[self.col:] + self.col += len(c) + self.desired_col = self.col + + def insert_newline(self) -> None: + line = self.lines[self.row] + before = line[:self.col] + after = line[self.col:] + self.lines[self.row] = before + self.lines.insert(self.row + 1, after) + self.row += 1 + self.col = 0 + self.desired_col = 0 + + def backspace(self) -> None: + if self.col > 0: + line = self.lines[self.row] + self.lines[self.row] = line[:self.col - 1] + line[self.col:] + self.col -= 1 + elif self.row > 0: + prev = self.lines[self.row - 1] + cur = self.lines[self.row] + self.col = len(prev) + self.lines[self.row - 1] = prev + cur + del self.lines[self.row] + self.row -= 1 + self.desired_col = self.col + + def delete(self) -> None: + line = self.lines[self.row] + if self.col < len(line): + self.lines[self.row] = line[:self.col] + line[self.col + 1:] + elif self.row < len(self.lines) - 1: + self.lines[self.row] = line + self.lines[self.row + 1] + del self.lines[self.row + 1] + self.desired_col = self.col + + def move_left(self) -> bool: + if self.col > 0: + self.col -= 1 + elif self.row > 0: + self.row -= 1 + self.col = len(self.lines[self.row]) + else: + return False + self.desired_col = self.col + return True + + def move_right(self) -> bool: + line = self.lines[self.row] + if self.col < len(line): + self.col += 1 + elif self.row < len(self.lines) - 1: + self.row += 1 + self.col = 0 + else: + return False + self.desired_col = self.col + return True + + def move_up(self) -> bool: + if self.row > 0: + self.row -= 1 + self.col = min(self.desired_col, len(self.lines[self.row])) + return True + return False + + def move_down(self) -> bool: + if self.row < len(self.lines) - 1: + self.row += 1 + self.col = min(self.desired_col, len(self.lines[self.row])) + return True + return False + + def move_bol(self) -> None: + self.col = 0 + self.desired_col = 0 + + def move_eol(self) -> None: + self.col = len(self.lines[self.row]) + self.desired_col = self.col + + def move_end_of_buffer(self) -> None: + self.row = len(self.lines) - 1 + self.col = len(self.lines[self.row]) + self.desired_col = self.col + + def is_at_end_of_buffer(self) -> bool: + return self.row == len(self.lines) - 1 and self.col == len(self.lines[-1]) + + def _word_left_col(self) -> int: + line = self.lines[self.row] + i = self.col + while i > 0 and not _is_word_char(line[i - 1]): + i -= 1 + while i > 0 and _is_word_char(line[i - 1]): + i -= 1 + return i + + def _word_right_col(self) -> int: + line = self.lines[self.row] + i = self.col + while i < len(line) and not _is_word_char(line[i]): + i += 1 + while i < len(line) and _is_word_char(line[i]): + i += 1 + return i + + def move_word_left(self) -> None: + if self.col == 0 and self.row > 0: + self.row -= 1 + self.col = len(self.lines[self.row]) + else: + self.col = self._word_left_col() + self.desired_col = self.col + + def move_word_right(self) -> None: + line = self.lines[self.row] + if self.col == len(line) and self.row < len(self.lines) - 1: + self.row += 1 + self.col = 0 + else: + self.col = self._word_right_col() + self.desired_col = self.col + + def delete_word_left(self) -> None: + if self.col == 0: + if self.row > 0: + self.backspace() + return + new_col = self._word_left_col() + line = self.lines[self.row] + self.lines[self.row] = line[:new_col] + line[self.col:] + self.col = new_col + self.desired_col = self.col + + def delete_word_right(self) -> None: + line = self.lines[self.row] + if self.col == len(line): + if self.row < len(self.lines) - 1: + self.delete() + return + new_col = self._word_right_col() + self.lines[self.row] = line[:self.col] + line[new_col:] + self.desired_col = self.col + + def kill_to_eol(self) -> None: + line = self.lines[self.row] + if self.col < len(line): + self.lines[self.row] = line[:self.col] + elif self.row < len(self.lines) - 1: + self.lines[self.row] = line + self.lines[self.row + 1] + del self.lines[self.row + 1] + self.desired_col = self.col + + def kill_to_bol(self) -> None: + line = self.lines[self.row] + self.lines[self.row] = line[self.col:] + self.col = 0 + self.desired_col = 0 diff --git a/easyrepl/editor.py b/easyrepl/editor.py new file mode 100644 index 0000000..2a3dfa1 --- /dev/null +++ b/easyrepl/editor.py @@ -0,0 +1,228 @@ +from typing import Optional + +from .buffer import Buffer +from .history import History +from .keys import Key, read_key +from .render import Renderer +from .terminal import Terminal + + +class _Submit(Exception): + pass + + +class _Abort(Exception): + pass + + +class _Eof(Exception): + pass + + +class LineEditor: + """Interactive multi-line line editor over a Terminal. + + Reads one user input at a time. Returns the finished text from `edit()`, + or raises KeyboardInterrupt on Ctrl-C / EOFError on Ctrl-D in empty buffer. + """ + + def __init__( + self, + terminal: Terminal, + history: History, + prompt: str, + continuation_prompt: str, + ): + self.terminal = terminal + self.history = history + self.renderer = Renderer(terminal, prompt, continuation_prompt) + + def edit(self) -> str: + buf = Buffer() + self.history.reset() + self.renderer.reset() + + while True: + self.renderer.render(buf) + key = read_key(self.terminal) + + try: + self._dispatch(key, buf) + except _Submit: + text = self._finalize_text(buf) + if text is None: + buf.insert_newline() + continue + self.renderer.finalize() + return text + except _Abort: + self.renderer.finalize() + raise KeyboardInterrupt + except _Eof: + self.renderer.finalize() + raise EOFError + + def _finalize_text(self, buf: Buffer) -> Optional[str]: + """Return the submission string if the buffer is complete, else None. + + Handles the triple-quote multi-line stripping rules from the legacy + implementation: the surrounding triple quotes are removed, then up to + one leading newline and one trailing newline are stripped. + """ + text = buf.text + if text.startswith('"""') or text.startswith("'''"): + delim = text[:3] + if len(text) < 6 or not text.endswith(delim): + return None + body = text[3:-3] + if body.startswith('\n'): + body = body[1:] + if body.endswith('\n'): + body = body[:-1] + return body + return text + + def _is_complete(self, buf: Buffer) -> bool: + text = buf.text + if text.startswith('"""') or text.startswith("'''"): + delim = text[:3] + return len(text) >= 6 and text.endswith(delim) + return True + + def _replace_buf(self, buf: Buffer, text: str) -> None: + buf.replace_with(Buffer.from_text(text)) + + def _dispatch(self, key: Key, buf: Buffer) -> None: + name = key.name + + if name == 'char': + buf.insert_char(key.char) + return + if name == 'enter': + if self._is_complete(buf) and buf.is_at_end_of_buffer(): + raise _Submit + buf.insert_newline() + return + if name == 'backspace': + buf.backspace() + return + if name == 'delete': + buf.delete() + return + if name in ('left', 'ctrl-b'): + buf.move_left() + return + if name in ('right', 'ctrl-f'): + buf.move_right() + return + if name == 'up': + if not buf.move_up(): + entry = self.history.prev(buf.text) + if entry is not None: + self._replace_buf(buf, entry) + return + if name == 'down': + if not buf.move_down(): + entry = self.history.next(buf.text) + if entry is not None: + self._replace_buf(buf, entry) + return + if name in ('home', 'ctrl-a'): + buf.move_bol() + return + if name in ('end', 'ctrl-e'): + buf.move_eol() + return + if name in ('ctrl-left', 'alt-b'): + buf.move_word_left() + return + if name in ('ctrl-right', 'alt-f'): + buf.move_word_right() + return + if name == 'ctrl-w': + buf.delete_word_left() + return + if name in ('alt-d', 'ctrl-delete'): + buf.delete_word_right() + return + if name == 'ctrl-k': + buf.kill_to_eol() + return + if name == 'ctrl-u': + buf.kill_to_bol() + return + if name == 'ctrl-l': + self.renderer.clear_screen() + return + if name == 'ctrl-c': + raise _Abort + if name == 'ctrl-d': + if not buf.text: + raise _Eof + buf.delete() + return + if name == 'ctrl-r': + self._search_mode(buf) + return + + def _search_mode(self, buf: Buffer) -> None: + """Reverse-i-search sub-mode. + + Saves the current buffer; types-pattern extends or shrinks the search + string; Ctrl-R steps to the previous match; Enter accepts the current + match and submits; Esc / Ctrl-G / Ctrl-C cancels and restores the buffer. + """ + saved = Buffer( + lines=list(buf.lines), + row=buf.row, + col=buf.col, + desired_col=buf.desired_col, + ) + pattern = "" + match_idx: Optional[int] = None + search_from: Optional[int] = None + + def apply_match() -> None: + if match_idx is not None: + self._replace_buf(buf, self.history.entries[match_idx]) + else: + buf.replace_with(saved) + + def render_search() -> None: + shown = pattern + self.renderer.render(buf, footer=f"(reverse-i-search)`{shown}': ") + + while True: + render_search() + key = read_key(self.terminal) + name = key.name + + if name == 'char': + pattern += key.char + match_idx = self.history.search_back(pattern) + search_from = match_idx + apply_match() + continue + if name == 'backspace': + pattern = pattern[:-1] + if pattern: + match_idx = self.history.search_back(pattern) + else: + match_idx = None + search_from = match_idx + apply_match() + continue + if name == 'ctrl-r': + if pattern and search_from is not None: + new_match = self.history.search_back(pattern, before=search_from) + if new_match is not None: + match_idx = new_match + search_from = new_match + apply_match() + continue + if name in ('escape', 'ctrl-g', 'ctrl-c'): + buf.replace_with(saved) + return + if name == 'enter': + raise _Submit + return diff --git a/easyrepl/history.py b/easyrepl/history.py new file mode 100644 index 0000000..ccc1ca6 --- /dev/null +++ b/easyrepl/history.py @@ -0,0 +1,106 @@ +import json +from pathlib import Path +from os import PathLike +from typing import List, Optional, Union + + +class History: + """In-memory command history with optional JSON-lines persistence. + + The cursor behavior mirrors readline: a fresh entry is added at the bottom, + `prev()` walks backwards through past entries, and `next()` walks forward, + returning to the user's pending unsubmitted text once past the most recent. + """ + + def __init__( + self, + path: Optional[Union[str, PathLike]] = None, + dedup: bool = True, + ): + self.path: Optional[Path] = None + self.entries: List[str] = [] + self.dedup = dedup + self.cursor: Optional[int] = None + self.pending: str = "" + + if path is not None: + self.path = Path(path).expanduser().resolve() + self.path.parent.mkdir(parents=True, exist_ok=True) + self._load() + + def _load(self) -> None: + if not self.path.exists(): + return + try: + text = self.path.read_text(encoding='utf-8') + except OSError: + return + for raw_line in text.splitlines(): + if not raw_line: + continue + try: + entry = json.loads(raw_line) + if isinstance(entry, str): + self.entries.append(entry) + continue + except json.JSONDecodeError: + pass + # legacy format: each file line is a single history entry + self.entries.append(raw_line) + + def _save(self) -> None: + if self.path is None: + return + try: + lines = [json.dumps(e) for e in self.entries] + self.path.write_text('\n'.join(lines) + ('\n' if lines else ''), encoding='utf-8') + except OSError: + pass + + def append(self, entry: str) -> None: + if not entry: + return + if self.dedup: + self.entries = [e for e in self.entries if e != entry] + self.entries.append(entry) + self._save() + self.reset() + + def reset(self) -> None: + self.cursor = None + self.pending = "" + + def _start(self, current: str) -> None: + if self.cursor is None: + self.pending = current + self.cursor = len(self.entries) + + def prev(self, current: str) -> Optional[str]: + if not self.entries: + return None + self._start(current) + if self.cursor > 0: + self.cursor -= 1 + return self.entries[self.cursor] + return None + + def next(self, current: str) -> Optional[str]: + if self.cursor is None: + return None + if self.cursor < len(self.entries) - 1: + self.cursor += 1 + return self.entries[self.cursor] + if self.cursor == len(self.entries) - 1: + self.cursor = len(self.entries) + return self.pending + return None + + def search_back(self, pattern: str, before: Optional[int] = None) -> Optional[int]: + if not pattern: + return None + if before is None: + before = len(self.entries) + for i in range(before - 1, -1, -1): + if pattern in self.entries[i]: + return i + return None diff --git a/easyrepl/keys.py b/easyrepl/keys.py new file mode 100644 index 0000000..ad5a251 --- /dev/null +++ b/easyrepl/keys.py @@ -0,0 +1,162 @@ +from typing import NamedTuple, Optional +from .terminal import Terminal, IS_WINDOWS + + +class Key(NamedTuple): + name: str + char: Optional[str] + ctrl: bool + alt: bool + + +_WIN_SPECIAL = { + 'H': 'up', + 'P': 'down', + 'K': 'left', + 'M': 'right', + 'G': 'home', + 'O': 'end', + 'S': 'delete', + 'I': 'page_up', + 'Q': 'page_down', + 's': 'ctrl-left', + 't': 'ctrl-right', + '\x8d': 'ctrl-up', + '\x91': 'ctrl-down', + '\x93': 'ctrl-delete', + '\x77': 'ctrl-home', + '\x75': 'ctrl-end', + '0': 'alt-b', + '!': 'alt-n', + '"': 'alt-m', + ' ': 'alt-d', +} + + +def _read_key_posix(term: Terminal) -> Key: + c = term.read_char() + b = ord(c) if len(c) == 1 else None + + if b == 0x1b: + if not term.peek_ready(0.05): + return Key('escape', None, False, False) + c2 = term.read_char() + if c2 == '[' or c2 == 'O': + seq = c2 + while True: + ch = term.read_char() + seq += ch + if ch.isalpha() or ch == '~': + break + if len(seq) > 16: + break + return _parse_csi(seq) + return Key('alt-' + c2.lower(), c2, False, True) + + if b == 0x0d or b == 0x0a: + return Key('enter', None, False, False) + if b == 0x09: + return Key('tab', None, False, False) + if b == 0x7f or b == 0x08: + return Key('backspace', None, False, False) + if b is not None and 0x01 <= b <= 0x1a: + letter = chr(b + 0x60) + return Key('ctrl-' + letter, None, True, False) + if b == 0x00: + return Key('ctrl-space', None, True, False) + + return Key('char', c, False, False) + + +def _parse_csi(seq: str) -> Key: + mapping = { + '[A': 'up', + '[B': 'down', + '[C': 'right', + '[D': 'left', + '[H': 'home', + '[F': 'end', + 'OH': 'home', + 'OF': 'end', + '[3~': 'delete', + '[5~': 'page_up', + '[6~': 'page_down', + '[1~': 'home', + '[4~': 'end', + '[7~': 'home', + '[8~': 'end', + '[1;5A': 'ctrl-up', + '[1;5B': 'ctrl-down', + '[1;5C': 'ctrl-right', + '[1;5D': 'ctrl-left', + '[1;3A': 'alt-up', + '[1;3B': 'alt-down', + '[1;3C': 'alt-right', + '[1;3D': 'alt-left', + '[5C': 'ctrl-right', + '[5D': 'ctrl-left', + '[3;5~': 'ctrl-delete', + } + if seq in mapping: + name = mapping[seq] + ctrl = name.startswith('ctrl-') + alt = name.startswith('alt-') + return Key(name, None, ctrl, alt) + return Key('unknown', None, False, False) + + +def _read_key_windows(term: Terminal) -> Key: + c = term.read_char() + if c == '\x00' or c == '\xe0': + c2 = term.read_char() + name = _WIN_SPECIAL.get(c2) + if name is None: + return Key('unknown', None, False, False) + ctrl = name.startswith('ctrl-') + alt = name.startswith('alt-') + return Key(name, None, ctrl, alt) + + if c == '\r' or c == '\n': + return Key('enter', None, False, False) + if c == '\t': + return Key('tab', None, False, False) + if c == '\x08' or c == '\x7f': + return Key('backspace', None, False, False) + if c == '\x1b': + if term.peek_ready(0.05): + c2 = term.read_char() + if c2 == '[' or c2 == 'O': + seq = c2 + while True: + ch = term.read_char() + seq += ch + if ch.isalpha() or ch == '~': + break + if len(seq) > 16: + break + return _parse_csi(seq) + return Key('alt-' + c2.lower(), c2, False, True) + return Key('escape', None, False, False) + if c == '\x03': + return Key('ctrl-c', None, True, False) + + b = ord(c) if len(c) == 1 else None + if b is not None and 0x01 <= b <= 0x1a: + letter = chr(b + 0x60) + return Key('ctrl-' + letter, None, True, False) + + return Key('char', c, False, False) + + +def read_key(term: Terminal) -> Key: + """Read one key event from the terminal. + + On Windows, msvcrt converts Ctrl-C into a KeyboardInterrupt at the runtime + level instead of returning '\\x03'; we catch that and synthesize a ctrl-c key. + """ + try: + if IS_WINDOWS: + return _read_key_windows(term) + return _read_key_posix(term) + except KeyboardInterrupt: + return Key('ctrl-c', None, True, False) diff --git a/easyrepl/render.py b/easyrepl/render.py new file mode 100644 index 0000000..f3e519d --- /dev/null +++ b/easyrepl/render.py @@ -0,0 +1,107 @@ +from typing import List +from wcwidth import wcswidth + +from .buffer import Buffer +from .terminal import Terminal + + +class Renderer: + """Differential renderer. + + Tracks the position the cursor was left in after the previous render and + rewrites the prompt area on each call by moving back to the start of the + prompt area, erasing to end of screen, and emitting fresh content. + """ + + def __init__(self, terminal: Terminal, prompt: str, continuation_prompt: str): + self.terminal = terminal + self.prompt = prompt + self.continuation_prompt = continuation_prompt + self.last_cursor_row = 0 + self.last_total_rows = 0 + + def reset(self) -> None: + self.last_cursor_row = 0 + self.last_total_rows = 0 + + def _prompt_for_row(self, row: int) -> str: + return self.prompt if row == 0 else self.continuation_prompt + + @staticmethod + def _width(s: str) -> int: + w = wcswidth(s) + return w if w >= 0 else len(s) + + def render(self, buf: Buffer, footer: str = "") -> None: + term_width = max(1, self.terminal.size().columns) + out: List[str] = [] + + if self.last_cursor_row > 0: + out.append(f'\x1b[{self.last_cursor_row}A') + out.append('\r') + out.append('\x1b[J') + + rendered_rows: List[int] = [] + for i, line in enumerate(buf.lines): + if i > 0: + out.append('\r\n') + prefix = self._prompt_for_row(i) + full = prefix + line + out.append(full) + line_width = self._width(full) + base_rows = max(1, (line_width + term_width - 1) // term_width) + if line_width > 0 and line_width % term_width == 0: + out.append(' \b') + base_rows += 1 + rendered_rows.append(base_rows) + + footer_rows = 0 + if footer: + out.append('\r\n' + footer) + fw = self._width(footer) + footer_rows = max(1, (fw + term_width - 1) // term_width) + + target_line = buf.lines[buf.row] + target_prefix = self._prompt_for_row(buf.row) + before_cursor_width = self._width(target_prefix + target_line[:buf.col]) + target_visual_row = before_cursor_width // term_width + target_visual_col = before_cursor_width % term_width + + rows_before = sum(rendered_rows[:buf.row]) + target_row_from_start = rows_before + target_visual_row + + current_row_from_start = sum(rendered_rows) - 1 + if footer: + current_row_from_start = sum(rendered_rows) + footer_rows - 1 + + delta = current_row_from_start - target_row_from_start + if delta > 0: + out.append(f'\x1b[{delta}A') + elif delta < 0: + out.append(f'\x1b[{-delta}B') + out.append('\r') + if target_visual_col > 0: + out.append(f'\x1b[{target_visual_col}C') + + self.last_total_rows = sum(rendered_rows) + footer_rows + self.last_cursor_row = target_row_from_start + + self.terminal.write(''.join(out)) + + def finalize(self) -> None: + """Move the cursor past the rendered area and emit a final newline.""" + out: List[str] = [] + if self.last_total_rows == 0: + return + delta = (self.last_total_rows - 1) - self.last_cursor_row + if delta > 0: + out.append(f'\x1b[{delta}B') + out.append('\r\n') + self.terminal.write(''.join(out)) + self.last_cursor_row = 0 + self.last_total_rows = 0 + + def clear_screen(self) -> None: + self.terminal.write('\x1b[2J\x1b[H') + self.last_cursor_row = 0 + self.last_total_rows = 0 diff --git a/easyrepl/repl.py b/easyrepl/repl.py index 95275b5..edad146 100644 --- a/easyrepl/repl.py +++ b/easyrepl/repl.py @@ -1,145 +1,113 @@ -import readline -from tempfile import NamedTemporaryFile -from pathlib import Path +import sys from os import PathLike -from typing import Union +from typing import Iterator, Optional, Union + +from .editor import LineEditor +from .history import History +from .terminal import Terminal + class REPL: - """ - simple python class for creating custom REPLs. - manages receiving input, cursor position, and history, while the library user + """Generator-based Read Evaluate Print Loop with a built-in line editor. Args: - prompt (str, optional): prompt to display before each line. Defaults to '>>> '. - history_file (Union[str,Path,None], optional): file to store history. Defaults to a temporary file. - dedup_history (bool, optional): remove duplicates from history. Defaults to True. - ctrl_c_quit (bool, optional): raise KeyboardInterrupt on Ctrl-C. Defaults to False. + prompt: prompt rendered before the first line of each input. Defaults to '>>> '. + continuation_prompt: prompt rendered on every line of a multi-line input + after the first. Defaults to '... '. + history: file path to load/save history. If None, history is in-memory only. + dedup_history: when True, appending an entry removes any prior identical entries. + ctrl_c_quit: when True, Ctrl-C re-raises KeyboardInterrupt to terminate the REPL. Yields: - str: each line of input from the user + Each non-empty submission as a string. Usage: - ```python - for line in REPL(): - # do something with line - print(line) - ``` + ```python + for line in REPL(): + print(line) + ``` """ - def __init__(self, *, prompt:str='>>> ', history:Union[PathLike,None]=None, dedup_history:bool=True, ctrl_c_quit:bool=False): + def __init__( + self, + *, + prompt: str = '>>> ', + continuation_prompt: str = '... ', + history: Optional[Union[str, PathLike]] = None, + dedup_history: bool = True, + ctrl_c_quit: bool = False, + ): self.prompt = prompt - self.external_history_file = NamedTemporaryFile() - self.external_history = self.external_history_file.name - self.dedup_history = dedup_history + self.continuation_prompt = continuation_prompt + self.history = History(path=history, dedup=dedup_history) self.ctrl_c_quit = ctrl_c_quit - # let easyrepl manually manage history - readline.set_auto_history(False) - - # If set, ensure that the regular history directory and file exists. - # Otherwise, create a temporary file - if history is None: - self.history_file_ref = NamedTemporaryFile() - self.history_file = self.history_file_ref.name - else: - history = Path(history).expanduser().resolve() - history.parent.mkdir(parents=True, exist_ok=True) # ensure the directory exists - self.history_file = str(history) - # a prior history file may or may not exist at this point. restore_history gracefully handles loading it, and replacing it with empty if it is missing or ill formed - - self.restore_history() - + def __iter__(self) -> Iterator[str]: + if not sys.stdin.isatty(): + yield from self._fallback_iter() + return - def stash_history(self): - """ - stash the current history and restore the external history - This should be called when the line is yielded to the user - It ensures that easyrepl's internal history doesn't interfere - with any other processes that might use readline, e.g. pdb - """ - readline.write_history_file(self.history_file) - readline.clear_history() - readline.set_auto_history(True) - readline.read_history_file(self.external_history) - - def restore_history(self): - """ - stash the external history and restore the current history - This should be called after control comes back from the yielded line - """ - readline.write_history_file(self.external_history) - readline.set_auto_history(False) - readline.clear_history() - try: - readline.read_history_file(self.history_file) - except Exception as e: - ... # any issues reading the file means we'll just clobber it next time we stash - - def __iter__(self): while True: try: - line = input(self.prompt) - - if line.startswith('"""') or line.startswith("'''"): - # If the first line starts with a triple quote, continue to read input - # until the closing triple quote is encountered - delimiter, line = line[0:3], line[3:] - - lines = [] - while True: - lines.append(line) - if line.endswith(delimiter): - break - line = input('... ') - - # join the lines together, removing the trailing triple quote - line = '\n'.join(lines) - line = line[:-3] - - # remove up to one newline from the beginning and end of the line - if line[0] == '\n': - line = line[1:] - if line[-1] == '\n': - line = line[:-1] - - if line: - if self.dedup_history: - # append without duplicates - i = 0 - while i < readline.get_current_history_length(): - if readline.get_history_item(i+1) == line: - readline.remove_history_item(i) - else: - i += 1 - - # append to history - readline.add_history(line) - - # yield line as next item in iteration, allowing the user to process it - # stash this history and restore any external history, in case pdb/etc. is used, they won't overlap - self.stash_history() - yield line - self.restore_history() - - except KeyboardInterrupt as e: + with Terminal() as term: + editor = LineEditor( + term, + self.history, + self.prompt, + self.continuation_prompt, + ) + line = editor.edit() + except KeyboardInterrupt: if self.ctrl_c_quit: - raise e from None - print() - print(KeyboardInterrupt.__name__) - + raise + print('KeyboardInterrupt') + continue except EOFError: - break + return - # save history at the end of the REPL - self.stash_history() + if line: + self.history.append(line) + yield line - -def readl(*, prompt='', ctrl_c_quit=True, **kwargs): - """read a single line using the REPL""" + def _fallback_iter(self) -> Iterator[str]: + """Plain `input()` loop for non-TTY stdin (pipes, redirects).""" + while True: + try: + line = input(self.prompt) + except EOFError: + return + except KeyboardInterrupt: + if self.ctrl_c_quit: + raise + print('KeyboardInterrupt') + continue + if line.startswith('"""') or line.startswith("'''"): + delim = line[:3] + rest = line[3:] + parts = [rest] + while True: + if parts[-1].endswith(delim): + break + try: + parts.append(input(self.continuation_prompt)) + except EOFError: + return + joined = '\n'.join(parts) + line = joined[:-3] + if line.startswith('\n'): + line = line[1:] + if line.endswith('\n'): + line = line[:-1] + if line: + self.history.append(line) + yield line + + +def readl(*, prompt: str = '', ctrl_c_quit: bool = True, **kwargs) -> str: + """Read a single line via the REPL editor, returning its contents.""" return next(iter(REPL(prompt=prompt, ctrl_c_quit=ctrl_c_quit, **kwargs))) if __name__ == '__main__': - # simple echo REPL for line in REPL(history='history.txt'): print(line) diff --git a/easyrepl/terminal.py b/easyrepl/terminal.py new file mode 100644 index 0000000..fb0fd7d --- /dev/null +++ b/easyrepl/terminal.py @@ -0,0 +1,95 @@ +import sys +import os +import platform +import shutil +from typing import Optional + +_IS_WINDOWS = platform.system() == 'Windows' + +if _IS_WINDOWS: + import msvcrt + import time + import colorama + colorama.just_fix_windows_console() +else: + import termios + import tty + import select + + +class Terminal: + """Cross-platform raw-mode terminal context manager. + + On POSIX, switches stdin into cbreak/raw mode for the duration of the + context and restores the previous termios state on exit. + On Windows, raw input is handled by msvcrt directly; colorama is + initialized at module import to enable VT-mode output processing. + """ + + def __enter__(self) -> 'Terminal': + if _IS_WINDOWS: + self.old_settings = None + else: + self.fd = sys.stdin.fileno() + self.old_settings = termios.tcgetattr(self.fd) + tty.setraw(self.fd) + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if not _IS_WINDOWS and self.old_settings is not None: + termios.tcsetattr(self.fd, termios.TCSADRAIN, self.old_settings) + self.old_settings = None + + def read_char(self) -> str: + """Read one logical character (decoding UTF-8 multibyte sequences on POSIX). + + Raises EOFError if the underlying stream is closed. + """ + if _IS_WINDOWS: + ch = msvcrt.getwch() + if ch == '': + raise EOFError + return ch + + b = os.read(self.fd, 1) + if not b: + raise EOFError + b0 = b[0] + if b0 < 0x80: + return b.decode('latin-1') + if b0 >= 0xf0: + n = 4 + elif b0 >= 0xe0: + n = 3 + elif b0 >= 0xc0: + n = 2 + else: + return b.decode('utf-8', errors='replace') + for _ in range(n - 1): + more = os.read(self.fd, 1) + if not more: + break + b += more + return b.decode('utf-8', errors='replace') + + def peek_ready(self, timeout: float = 0.05) -> bool: + """Return True if a character is available within `timeout` seconds.""" + if _IS_WINDOWS: + end = time.monotonic() + timeout + while time.monotonic() < end: + if msvcrt.kbhit(): + return True + time.sleep(0.001) + return False + r, _, _ = select.select([self.fd], [], [], timeout) + return bool(r) + + def write(self, s: str) -> None: + sys.stdout.write(s) + sys.stdout.flush() + + def size(self) -> os.terminal_size: + return shutil.get_terminal_size() + + +IS_WINDOWS = _IS_WINDOWS diff --git a/pyproject.toml b/pyproject.toml index 1c4978d..0fcf792 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,24 +1,30 @@ -[tool.poetry] +[project] name = "easyrepl" -version = "0.2.0" +version = "0.3.0" description = "A simple python class for creating Read Evaluate Print Line (REPL) interfaces" -authors = ["David Samson "] -license = "MIT" readme = "README.md" -homepage = "https://github.com/david-andrew/EasyREPL" - +license = { text = "MIT" } +authors = [ + { name = "David Samson", email = "david.andrew.engineer@gmail.com" }, +] +requires-python = ">=3.8" +dependencies = [ + "wcwidth", + "colorama", +] classifiers = [ - "Operating System :: POSIX" + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "License :: OSI Approved :: MIT License", ] -[tool.poetry.urls] +[project.urls] +Homepage = "https://github.com/david-andrew/EasyREPL" "Bug Tracker" = "https://github.com/david-andrew/EasyREPL/issues" - -[tool.poetry.dependencies] -python = ">=3.7" - - [build-system] -requires = ["poetry-core"] -build-backend = "poetry.core.masonry.api" +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["easyrepl"] diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..8c50cfd --- /dev/null +++ b/uv.lock @@ -0,0 +1,36 @@ +version = 1 +revision = 3 +requires-python = ">=3.8" + +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + +[[package]] +name = "easyrepl" +version = "0.3.0" +source = { editable = "." } +dependencies = [ + { name = "colorama" }, + { name = "wcwidth" }, +] + +[package.metadata] +requires-dist = [ + { name = "colorama" }, + { name = "wcwidth" }, +] + +[[package]] +name = "wcwidth" +version = "0.7.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/2c/ee/afaf0f85a9a18fe47a67f1e4422ed6cf1fe642f0ae0a2f81166231303c52/wcwidth-0.7.0.tar.gz", hash = "sha256:90e3a7ea092341c44b99562e75d09e4d5160fe7a3974c6fb842a101a95e7eed0", size = 182132, upload-time = "2026-05-02T16:04:12.653Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/41/52/e465037f5375f43533d1a80b6923955201596a99142ed524d77b571a1418/wcwidth-0.7.0-py3-none-any.whl", hash = "sha256:5d69154c429a82910e241c738cd0e2976fac8a2dd47a1a805f4afed1c0f136f2", size = 110825, upload-time = "2026-05-02T16:04:11.033Z" }, +] From 949bbb1faf93a0cf6cddb685c0f1f4fa6c18ec81 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 27 May 2026 19:46:06 +0000 Subject: [PATCH 2/4] Add multi-OS unit tests and real TTY smoke script for CI - pytest suite with FakeTerminal editor/key tests - scripts/tty_smoke_test.py drives readl through a PTY - GitHub Actions matrix on ubuntu, macos, and windows Co-authored-by: David-Andrew Samson --- .github/workflows/test-multi-os.yml | 42 ++++++++++++++++++ pyproject.toml | 6 +++ scripts/tty_smoke_test.py | 67 +++++++++++++++++++++++++++++ tests/conftest.py | 27 ++++++++++++ tests/test_buffer.py | 37 ++++++++++++++++ tests/test_editor.py | 28 ++++++++++++ tests/test_history.py | 29 +++++++++++++ tests/test_keys.py | 44 +++++++++++++++++++ tests/test_repl.py | 58 +++++++++++++++++++++++++ 9 files changed, 338 insertions(+) create mode 100644 .github/workflows/test-multi-os.yml create mode 100644 scripts/tty_smoke_test.py create mode 100644 tests/conftest.py create mode 100644 tests/test_buffer.py create mode 100644 tests/test_editor.py create mode 100644 tests/test_history.py create mode 100644 tests/test_keys.py create mode 100644 tests/test_repl.py diff --git a/.github/workflows/test-multi-os.yml b/.github/workflows/test-multi-os.yml new file mode 100644 index 0000000..4f10ed6 --- /dev/null +++ b/.github/workflows/test-multi-os.yml @@ -0,0 +1,42 @@ +name: Multi-OS tests + +on: + push: + branches: [multi-os-revamp, 'cursor/**'] + pull_request: + branches: [multi-os-revamp, master] + workflow_dispatch: + +jobs: + unit-tests: + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + python-version: ['3.12'] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install package and test deps + run: pip install -e ".[dev]" + - name: Run unit tests + run: pytest -v + + tty-smoke: + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - name: Install package + run: pip install -e . + - name: Real TTY smoke test (PTY) + run: python scripts/tty_smoke_test.py diff --git a/pyproject.toml b/pyproject.toml index 0fcf792..a74ddda 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,9 @@ classifiers = [ "License :: OSI Approved :: MIT License", ] +[project.optional-dependencies] +dev = ["pytest>=8.0"] + [project.urls] Homepage = "https://github.com/david-andrew/EasyREPL" "Bug Tracker" = "https://github.com/david-andrew/EasyREPL/issues" @@ -28,3 +31,6 @@ build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["easyrepl"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/scripts/tty_smoke_test.py b/scripts/tty_smoke_test.py new file mode 100644 index 0000000..8483830 --- /dev/null +++ b/scripts/tty_smoke_test.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +"""Drive easyrepl through a real TTY (PTY when available). + +Exits 0 when a line can be typed and read back via readl(); non-zero on failure. +""" + +from __future__ import annotations + +import os +import select +import subprocess +import sys +import time + + +def _pty_smoke() -> int: + master, slave = os.openpty() + child = subprocess.Popen( + [ + sys.executable, + '-c', + "from easyrepl import readl; print('RESULT:' + repr(readl(prompt='>>> ')))", + ], + stdin=slave, + stdout=slave, + stderr=slave, + close_fds=True, + ) + os.close(slave) + time.sleep(0.25) + os.write(master, b'smoke-line\r') + captured = b'' + deadline = time.time() + 10 + while time.time() < deadline: + if b"RESULT:'smoke-line'" in captured: + break + if child.poll() is not None and b'RESULT:' in captured: + break + ready, _, _ = select.select([master], [], [], 0.1) + if ready: + chunk = os.read(master, 4096) + if not chunk: + break + captured += chunk + os.close(master) + child.wait(timeout=5) + if child.returncode != 0: + print(f'child exit {child.returncode}', file=sys.stderr) + print(captured.decode('utf-8', errors='replace'), file=sys.stderr) + return 1 + if b"RESULT:'smoke-line'" not in captured: + print('missing expected RESULT in TTY output', file=sys.stderr) + print(captured.decode('utf-8', errors='replace'), file=sys.stderr) + return 1 + print('TTY smoke OK') + return 0 + + +def main() -> int: + if not hasattr(os, 'openpty'): + print('openpty not available on this platform', file=sys.stderr) + return 1 + return _pty_smoke() + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..c0c22ae --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,27 @@ +import os +from dataclasses import dataclass, field + + +@dataclass +class FakeTerminal: + """Scripted terminal for driving LineEditor without a real TTY.""" + + chars: list[str] = field(default_factory=list) + _index: int = 0 + written: list[str] = field(default_factory=list) + + def read_char(self) -> str: + if self._index >= len(self.chars): + raise EOFError + ch = self.chars[self._index] + self._index += 1 + return ch + + def peek_ready(self, timeout: float = 0.05) -> bool: + return self._index < len(self.chars) + + def write(self, s: str) -> None: + self.written.append(s) + + def size(self) -> os.terminal_size: + return os.terminal_size((80, 24)) diff --git a/tests/test_buffer.py b/tests/test_buffer.py new file mode 100644 index 0000000..a038571 --- /dev/null +++ b/tests/test_buffer.py @@ -0,0 +1,37 @@ +from easyrepl.buffer import Buffer + + +def test_insert_and_backspace(): + buf = Buffer() + for c in 'hi': + buf.insert_char(c) + assert buf.text == 'hi' + buf.backspace() + assert buf.text == 'h' + + +def test_multiline_newline(): + buf = Buffer.from_text('ab') + buf.move_end_of_buffer() + buf.insert_newline() + buf.insert_char('c') + assert buf.text == 'ab\nc' + assert buf.row == 1 and buf.col == 1 + + +def test_word_motion(): + buf = Buffer.from_text('foo bar_baz') + buf.move_end_of_buffer() + buf.move_word_left() + assert buf.col == 4 + buf.move_word_left() + assert buf.col == 0 + + +def test_kill_line(): + buf = Buffer.from_text('hello world') + buf.move_bol() + buf.move_right() + buf.move_right() + buf.kill_to_eol() + assert buf.text == 'he' diff --git a/tests/test_editor.py b/tests/test_editor.py new file mode 100644 index 0000000..f343fd4 --- /dev/null +++ b/tests/test_editor.py @@ -0,0 +1,28 @@ +from easyrepl.editor import LineEditor +from easyrepl.history import History +from tests.conftest import FakeTerminal + + +def test_simple_line_submit(): + term = FakeTerminal(chars=list('hello') + ['\r']) + editor = LineEditor(term, History(), '>>> ', '... ') + assert editor.edit() == 'hello' + + +def test_triple_quote_multiline(): + """Closing delimiter on last line submits the block.""" + chars = list('"""') + ['\r'] + chars += list('line one') + ['\r'] + chars += list('line two') + ['\r'] + chars += list('"""') + ['\r'] + term = FakeTerminal(chars=chars) + editor = LineEditor(term, History(), '>>> ', '... ') + assert editor.edit() == 'line one\nline two' + + +def test_history_navigation(): + hist = History() + hist.append('previous') + term = FakeTerminal(chars=['\x1b', '[', 'A', '\r']) + editor = LineEditor(term, hist, '>>> ', '... ') + assert editor.edit() == 'previous' diff --git a/tests/test_history.py b/tests/test_history.py new file mode 100644 index 0000000..f015124 --- /dev/null +++ b/tests/test_history.py @@ -0,0 +1,29 @@ +from pathlib import Path + +from easyrepl.history import History + + +def test_prev_next_roundtrip(tmp_path: Path): + h = History() + h.append('one') + h.append('two') + assert h.prev('draft') == 'two' + assert h.prev('two') == 'one' + assert h.next('one') == 'two' + assert h.next('two') == 'draft' + + +def test_dedup(tmp_path: Path): + path = tmp_path / 'hist.jsonl' + h = History(path=path, dedup=True) + h.append('a') + h.append('b') + h.append('a') + assert h.entries == ['b', 'a'] + + +def test_load_legacy_lines(tmp_path: Path): + path = tmp_path / 'hist.txt' + path.write_text('legacy-one\nlegacy-two\n', encoding='utf-8') + h = History(path=path) + assert h.entries == ['legacy-one', 'legacy-two'] diff --git a/tests/test_keys.py b/tests/test_keys.py new file mode 100644 index 0000000..9e31a68 --- /dev/null +++ b/tests/test_keys.py @@ -0,0 +1,44 @@ +import pytest + +from easyrepl.keys import Key, _read_key_posix, _read_key_windows, read_key +from tests.conftest import FakeTerminal + + +def test_posix_arrow_and_enter(): + term = FakeTerminal(chars=['\x1b', '[', 'A', 'x', '\r']) + assert _read_key_posix(term).name == 'up' + key = _read_key_posix(term) + assert key.name == 'char' and key.char == 'x' + assert _read_key_posix(term).name == 'enter' + + +def test_posix_ctrl_letter(): + term = FakeTerminal(chars=['\x01']) + assert _read_key_posix(term) == Key('ctrl-a', None, True, False) + + +def test_windows_arrow_prefix(): + term = FakeTerminal(chars=['\x00', 'H', 'a', '\r']) + assert _read_key_windows(term).name == 'up' + key = _read_key_windows(term) + assert key.name == 'char' and key.char == 'a' + assert _read_key_windows(term).name == 'enter' + + +def test_windows_ctrl_c_synthetic(monkeypatch): + monkeypatch.setattr('easyrepl.keys.IS_WINDOWS', True) + term = FakeTerminal() + + def raise_interrupt(): + raise KeyboardInterrupt + + monkeypatch.setattr(term, 'read_char', raise_interrupt) + assert read_key(term) == Key('ctrl-c', None, True, False) + + +@pytest.mark.parametrize('is_windows', [False, True]) +def test_read_key_dispatches(monkeypatch, is_windows: bool): + monkeypatch.setattr('easyrepl.keys.IS_WINDOWS', is_windows) + term = FakeTerminal(chars=['\x05']) + key = read_key(term) + assert key.name == 'ctrl-e' diff --git a/tests/test_repl.py b/tests/test_repl.py new file mode 100644 index 0000000..df109c7 --- /dev/null +++ b/tests/test_repl.py @@ -0,0 +1,58 @@ +import os +import platform +import select +import subprocess +import sys +import time + +import pytest + + +def test_pipe_mode_integration(): + script = """ +from easyrepl import REPL +for line in REPL(prompt='> '): + print('OUT:' + line) +""" + proc = subprocess.run( + [sys.executable, '-c', script], + input='first\nsecond\n', + text=True, + capture_output=True, + timeout=10, + ) + assert proc.returncode == 0 + assert 'OUT:first' in proc.stdout + assert 'OUT:second' in proc.stdout + + +@pytest.mark.skipif(not hasattr(os, 'openpty'), reason='requires os.openpty') +def test_pty_repl_submits_line(): + master, slave = os.openpty() + proc = subprocess.Popen( + [sys.executable, '-c', 'from easyrepl import readl; print(readl(prompt=">>> "))'], + stdin=slave, + stdout=slave, + stderr=slave, + close_fds=True, + ) + os.close(slave) + time.sleep(0.25) + os.write(master, b'pty-test\r') + out = b'' + deadline = time.time() + 5 + while time.time() < deadline: + if proc.poll() is not None: + break + ready, _, _ = select.select([master], [], [], 0.1) + if ready: + chunk = os.read(master, 4096) + if not chunk: + break + out += chunk + if b'pty-test' in out: + break + os.close(master) + proc.wait(timeout=5) + assert proc.returncode == 0, (proc.returncode, out) + assert b'pty-test' in out From 4732c5c8d69ac1f3660bceb42a29960b66fe556b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 27 May 2026 19:49:02 +0000 Subject: [PATCH 3/4] Fix CI: drop flaky PTY pytest, add Windows winpty TTY smoke Co-authored-by: David-Andrew Samson --- .github/workflows/test-multi-os.yml | 5 +- scripts/tty_smoke_test.py | 73 ++++++++++++++++++++++------- tests/test_repl.py | 38 --------------- 3 files changed, 61 insertions(+), 55 deletions(-) diff --git a/.github/workflows/test-multi-os.yml b/.github/workflows/test-multi-os.yml index 4f10ed6..39219ca 100644 --- a/.github/workflows/test-multi-os.yml +++ b/.github/workflows/test-multi-os.yml @@ -38,5 +38,8 @@ jobs: python-version: '3.12' - name: Install package run: pip install -e . - - name: Real TTY smoke test (PTY) + - name: Install Windows PTY helper + if: runner.os == 'Windows' + run: pip install pywinpty + - name: Real TTY smoke test run: python scripts/tty_smoke_test.py diff --git a/scripts/tty_smoke_test.py b/scripts/tty_smoke_test.py index 8483830..745c908 100644 --- a/scripts/tty_smoke_test.py +++ b/scripts/tty_smoke_test.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Drive easyrepl through a real TTY (PTY when available). +"""Drive easyrepl through a real TTY (PTY on POSIX, winpty on Windows). Exits 0 when a line can be typed and read back via readl(); non-zero on failure. """ @@ -7,20 +7,22 @@ from __future__ import annotations import os +import platform import select import subprocess import sys import time +_CHILD = ( + "from easyrepl import readl; print('RESULT:' + repr(readl(prompt='>>> ')))" +) +_EXPECTED = b"RESULT:'smoke-line'" -def _pty_smoke() -> int: + +def _posix_smoke() -> int: master, slave = os.openpty() child = subprocess.Popen( - [ - sys.executable, - '-c', - "from easyrepl import readl; print('RESULT:' + repr(readl(prompt='>>> ')))", - ], + [sys.executable, '-c', _CHILD], stdin=slave, stdout=slave, stderr=slave, @@ -29,12 +31,47 @@ def _pty_smoke() -> int: os.close(slave) time.sleep(0.25) os.write(master, b'smoke-line\r') + captured = _read_until_done(child, master) + os.close(master) + child.wait(timeout=5) + return _check(child.returncode, captured) + + +def _windows_smoke() -> int: + try: + from winpty import PtyProcess + except ImportError: + print('pywinpty required on Windows: pip install pywinpty', file=sys.stderr) + return 1 + + proc = PtyProcess.spawn(f'{sys.executable} -c "{_CHILD}"') + time.sleep(0.5) + proc.write('smoke-line\r') captured = b'' deadline = time.time() + 10 while time.time() < deadline: - if b"RESULT:'smoke-line'" in captured: + if not proc.isalive() and not proc.iseof(): + break + try: + chunk = proc.read(4096) + except EOFError: break - if child.poll() is not None and b'RESULT:' in captured: + if chunk: + captured += chunk.encode('utf-8', errors='replace') + if _EXPECTED in captured: + break + time.sleep(0.05) + proc.close() + return _check(proc.getexitcode(), captured) + + +def _read_until_done(child: subprocess.Popen, master: int) -> bytes: + captured = b'' + deadline = time.time() + 10 + while time.time() < deadline: + if _EXPECTED in captured: + break + if child.poll() is not None and _EXPECTED in captured: break ready, _, _ = select.select([master], [], [], 0.1) if ready: @@ -42,25 +79,29 @@ def _pty_smoke() -> int: if not chunk: break captured += chunk - os.close(master) - child.wait(timeout=5) - if child.returncode != 0: - print(f'child exit {child.returncode}', file=sys.stderr) + return captured + + +def _check(returncode: int | None, captured: bytes) -> int: + if returncode != 0: + print(f'child exit {returncode}', file=sys.stderr) print(captured.decode('utf-8', errors='replace'), file=sys.stderr) return 1 - if b"RESULT:'smoke-line'" not in captured: + if _EXPECTED not in captured: print('missing expected RESULT in TTY output', file=sys.stderr) print(captured.decode('utf-8', errors='replace'), file=sys.stderr) return 1 - print('TTY smoke OK') + print(f'{platform.system()} TTY smoke OK') return 0 def main() -> int: + if platform.system() == 'Windows': + return _windows_smoke() if not hasattr(os, 'openpty'): print('openpty not available on this platform', file=sys.stderr) return 1 - return _pty_smoke() + return _posix_smoke() if __name__ == '__main__': diff --git a/tests/test_repl.py b/tests/test_repl.py index df109c7..91c0a2b 100644 --- a/tests/test_repl.py +++ b/tests/test_repl.py @@ -1,11 +1,5 @@ -import os -import platform -import select import subprocess import sys -import time - -import pytest def test_pipe_mode_integration(): @@ -24,35 +18,3 @@ def test_pipe_mode_integration(): assert proc.returncode == 0 assert 'OUT:first' in proc.stdout assert 'OUT:second' in proc.stdout - - -@pytest.mark.skipif(not hasattr(os, 'openpty'), reason='requires os.openpty') -def test_pty_repl_submits_line(): - master, slave = os.openpty() - proc = subprocess.Popen( - [sys.executable, '-c', 'from easyrepl import readl; print(readl(prompt=">>> "))'], - stdin=slave, - stdout=slave, - stderr=slave, - close_fds=True, - ) - os.close(slave) - time.sleep(0.25) - os.write(master, b'pty-test\r') - out = b'' - deadline = time.time() + 5 - while time.time() < deadline: - if proc.poll() is not None: - break - ready, _, _ = select.select([master], [], [], 0.1) - if ready: - chunk = os.read(master, 4096) - if not chunk: - break - out += chunk - if b'pty-test' in out: - break - os.close(master) - proc.wait(timeout=5) - assert proc.returncode == 0, (proc.returncode, out) - assert b'pty-test' in out From 1a801d5344bb8248222816b48d9764492fbcd19f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 27 May 2026 19:51:48 +0000 Subject: [PATCH 4/4] Fix Windows winpty exitstatus handling in TTY smoke Co-authored-by: David-Andrew Samson --- scripts/tty_smoke_test.py | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/scripts/tty_smoke_test.py b/scripts/tty_smoke_test.py index 745c908..fecb09a 100644 --- a/scripts/tty_smoke_test.py +++ b/scripts/tty_smoke_test.py @@ -44,25 +44,28 @@ def _windows_smoke() -> int: print('pywinpty required on Windows: pip install pywinpty', file=sys.stderr) return 1 - proc = PtyProcess.spawn(f'{sys.executable} -c "{_CHILD}"') + proc = PtyProcess.spawn([sys.executable, '-c', _CHILD]) time.sleep(0.5) proc.write('smoke-line\r') captured = b'' - deadline = time.time() + 10 + deadline = time.time() + 15 while time.time() < deadline: - if not proc.isalive() and not proc.iseof(): + if _EXPECTED in captured: break try: chunk = proc.read(4096) except EOFError: break if chunk: - captured += chunk.encode('utf-8', errors='replace') - if _EXPECTED in captured: + if isinstance(chunk, bytes): + captured += chunk + else: + captured += chunk.encode('utf-8', errors='replace') + if not proc.isalive(): break time.sleep(0.05) - proc.close() - return _check(proc.getexitcode(), captured) + proc.wait() + return _check(proc.exitstatus, captured) def _read_until_done(child: subprocess.Popen, master: int) -> bytes: @@ -83,14 +86,14 @@ def _read_until_done(child: subprocess.Popen, master: int) -> bytes: def _check(returncode: int | None, captured: bytes) -> int: - if returncode != 0: - print(f'child exit {returncode}', file=sys.stderr) - print(captured.decode('utf-8', errors='replace'), file=sys.stderr) - return 1 if _EXPECTED not in captured: print('missing expected RESULT in TTY output', file=sys.stderr) print(captured.decode('utf-8', errors='replace'), file=sys.stderr) return 1 + if returncode not in (None, 0): + print(f'child exit {returncode}', file=sys.stderr) + print(captured.decode('utf-8', errors='replace'), file=sys.stderr) + return 1 print(f'{platform.system()} TTY smoke OK') return 0