diff --git a/micropython/aiorepl/README.md b/micropython/aiorepl/README.md index c1c08b899..f48c3e35c 100644 --- a/micropython/aiorepl/README.md +++ b/micropython/aiorepl/README.md @@ -94,8 +94,29 @@ Ctrl-D at the asyncio REPL command prompt will terminate the current event loop, The following features are unsupported: -* Tab completion is not supported (also unsupported in `python -m asyncio`). +* Tab completion requires `micropython.repl_autocomplete` (available when firmware is built with `MICROPY_HELPER_REPL`, which is the default for most ports). * Multi-line continuation. However you can do single-line definitions of functions, see demo above. * Exception tracebacks. Only the exception type and message is shown, see demo above. * Emacs shortcuts (e.g. Ctrl-A, Ctrl-E, to move to start/end of line). * Unicode handling for input. + +## Testing + +### Unit tests + +`test_autocomplete.py` tests the `micropython.repl_autocomplete` API contract +that aiorepl depends on. Run on the unix port: + + MICROPY_MICROPYTHON=ports/unix/build-standard/micropython \ + tests/run-tests.py lib/micropython-lib/micropython/aiorepl/test_autocomplete.py + +Or deploy to a device with `mpremote cp` and run directly. + +### Integration test (PTY) + +`test_aiorepl_pty.py` exercises aiorepl interactively via a pseudo-terminal, +testing tab completion, command execution, and terminal mode switching. +Run with CPython against the unix build: + + python3 lib/micropython-lib/micropython/aiorepl/test_aiorepl_pty.py \ + ports/unix/build-standard/micropython diff --git a/micropython/aiorepl/aiorepl.py b/micropython/aiorepl/aiorepl.py index 15026e435..80790dcbc 100644 --- a/micropython/aiorepl/aiorepl.py +++ b/micropython/aiorepl/aiorepl.py @@ -97,6 +97,7 @@ async def task(g=None, prompt="--> "): print("Starting asyncio REPL...") if g is None: g = __import__("__main__").__dict__ + _stdio_raw = getattr(micropython, "stdio_mode_raw", lambda _: None) try: micropython.kbd_intr(-1) s = asyncio.StreamReader(sys.stdin) @@ -106,6 +107,8 @@ async def task(g=None, prompt="--> "): hist_n = 0 # Number of history entries. c = 0 # ord of most recent character. t = 0 # timestamp of most recent character. + _autocomplete = getattr(micropython, "repl_autocomplete", None) + _stdio_raw(True) while True: hist_b = 0 # How far back in the history are we currently. sys.stdout.write(prompt) @@ -144,10 +147,12 @@ async def task(g=None, prompt="--> "): hist_n = min(_HISTORY_LIMIT - 1, hist_n + 1) hist_i = (hist_i + 1) % _HISTORY_LIMIT + _stdio_raw(False) result = await execute(cmd, g, s) if result is not None: sys.stdout.write(repr(result)) sys.stdout.write("\n") + _stdio_raw(True) break elif c == 0x08 or c == 0x7F: # Backspace. @@ -163,7 +168,9 @@ async def task(g=None, prompt="--> "): cmd = cmd[:-1] sys.stdout.write("\x08 \x08") elif c == CHAR_CTRL_A: + _stdio_raw(False) raw_repl(sys.stdin, g) + _stdio_raw(True) break elif c == CHAR_CTRL_B: continue @@ -174,10 +181,12 @@ async def task(g=None, prompt="--> "): break elif c == CHAR_CTRL_D: if paste: + _stdio_raw(False) result = await execute(cmd, g, s) if result is not None: sys.stdout.write(repr(result)) sys.stdout.write("\n") + _stdio_raw(True) break sys.stdout.write("\n") @@ -187,6 +196,31 @@ async def task(g=None, prompt="--> "): elif c == CHAR_CTRL_E: sys.stdout.write("paste mode; Ctrl-C to cancel, Ctrl-D to finish\n===\n") paste = True + elif c == 0x09 and not paste: + # Tab key. + cursor_pos = len(cmd) - curs + if cursor_pos > 0 and cmd[cursor_pos - 1] <= " ": + # Insert 4 spaces for indentation after whitespace. + compl = " " + elif _autocomplete and cursor_pos > 0: + compl = _autocomplete(cmd[:cursor_pos]) + else: + compl = "" + if compl: + # Insert completion at cursor. + if curs: + cmd = "".join((cmd[:-curs], compl, cmd[-curs:])) + sys.stdout.write(cmd[-curs - len(compl) :]) + sys.stdout.write("\x1b[{}D".format(curs)) + else: + sys.stdout.write(compl) + cmd += compl + elif compl is None: + # Multiple matches printed by autocomplete, redraw line. + sys.stdout.write(prompt) + sys.stdout.write(cmd) + if curs: + sys.stdout.write("\x1b[{}D".format(curs)) elif c == 0x1B: # Start of escape sequence. key = await s.read(2) @@ -238,6 +272,7 @@ async def task(g=None, prompt="--> "): sys.stdout.write(b) cmd += b finally: + _stdio_raw(False) micropython.kbd_intr(3) diff --git a/micropython/aiorepl/manifest.py b/micropython/aiorepl/manifest.py index 83802e1c0..6def2aba9 100644 --- a/micropython/aiorepl/manifest.py +++ b/micropython/aiorepl/manifest.py @@ -1,5 +1,5 @@ metadata( - version="0.2.2", + version="0.3.0", description="Provides an asynchronous REPL that can run concurrently with an asyncio, also allowing await expressions.", ) diff --git a/micropython/aiorepl/test_aiorepl_pty.py b/micropython/aiorepl/test_aiorepl_pty.py new file mode 100755 index 000000000..249be5f78 --- /dev/null +++ b/micropython/aiorepl/test_aiorepl_pty.py @@ -0,0 +1,194 @@ +#!/usr/bin/env python3 +"""PTY-based integration test for aiorepl features. + +Requires the unix port of micropython. + +Usage: + python3 test_aiorepl_pty.py [path/to/micropython] + +The micropython binary must have MICROPY_HELPER_REPL and +MICROPY_PY_MICROPYTHON_STDIO_RAW enabled (unix standard build). +MICROPYPATH is set automatically to include frozen modules and this +directory (for aiorepl). +""" + +import os +import pty +import re +import select +import subprocess +import sys + + +MICROPYTHON = sys.argv[1] if len(sys.argv) > 1 else os.environ.get( + "MICROPY_MICROPYTHON", "micropython" +) + + +def get(master, timeout=0.02, required=False): + """Read from PTY master until *timeout* seconds of silence.""" + rv = b"" + while True: + ready = select.select([master], [], [], timeout) + if ready[0] == [master]: + rv += os.read(master, 1024) + else: + if not required or rv: + return rv + + +def send_get(master, data, timeout=0.02): + """Write *data* to PTY master and return response after silence.""" + os.write(master, data) + return get(master, timeout) + + +def strip_ansi(data): + """Remove ANSI escape sequences from bytes.""" + return re.sub(rb"\x1b\[[0-9;]*[A-Za-z]", b"", data) + + +class TestFailure(Exception): + pass + + +def assert_in(needle, haystack, label): + if needle not in haystack: + raise TestFailure( + f"[{label}] expected {needle!r} in output, got: {haystack!r}" + ) + + +def assert_not_in(needle, haystack, label): + if needle in haystack: + raise TestFailure( + f"[{label}] did not expect {needle!r} in output, got: {haystack!r}" + ) + + +def main(): + master, slave = pty.openpty() + + # Build MICROPYPATH: frozen modules + this directory (for aiorepl). + this_dir = os.path.dirname(os.path.abspath(__file__)) + env = os.environ.copy() + env["MICROPYPATH"] = ".frozen:" + this_dir + + p = subprocess.Popen( + [MICROPYTHON], + stdin=slave, + stdout=slave, + stderr=subprocess.STDOUT, + bufsize=0, + env=env, + ) + + passed = 0 + failed = 0 + + try: + # Wait for the standard REPL banner and >>> prompt. + banner = get(master, timeout=0.1, required=True) + if b">>>" not in banner: + raise TestFailure(f"No REPL banner/prompt, got: {banner!r}") + + # --- Test 1: Start aiorepl --- + # Standard REPL readline handles both \r and \n as enter. + resp = send_get( + master, + b"import asyncio, aiorepl; asyncio.run(aiorepl.task())\r", + timeout=0.1, + ) + assert_in(b"Starting asyncio REPL", resp, "startup") + assert_in(b"--> ", resp, "startup prompt") + print("PASS: startup") + passed += 1 + + # Once aiorepl is running, the terminal is in raw mode (ICRNL cleared), + # and aiorepl only handles 0x0A (LF) as enter — not 0x0D (CR). + # All subsequent commands must use \n. + + # --- Test 2: Single tab completion --- + # Type "impo" then tab. Should complete to "import " (suffix "rt "). + resp = send_get(master, b"impo\x09", timeout=0.05) + clean = strip_ansi(resp) + assert_in(b"rt ", clean, "single tab completion") + # Ctrl-C to clear and get fresh prompt. + resp = send_get(master, b"\x03", timeout=0.05) + assert_in(b"--> ", resp, "prompt after ctrl-c") + print("PASS: single tab completion") + passed += 1 + + # --- Test 3: Attribute tab completion --- + # First import sys. + resp = send_get(master, b"import sys\n", timeout=0.1) + # Now type "sys.ver" + tab. Should complete common prefix "sion". + resp = send_get(master, b"sys.ver\x09", timeout=0.05) + clean = strip_ansi(resp) + assert_in(b"sion", clean, "attribute tab completion") + # Ctrl-C to clear. + resp = send_get(master, b"\x03", timeout=0.05) + print("PASS: attribute tab completion") + passed += 1 + + # --- Test 4: Multiple-match completion --- + # Create two globals sharing prefix "_tc". + resp = send_get(master, b"_tca=1;_tcb=2\n", timeout=0.1) + # Type "_tc" + tab. Multiple matches -> candidates printed, returns None. + resp = send_get(master, b"_tc\x09", timeout=0.05) + clean = strip_ansi(resp) + assert_in(b"_tca", clean, "multi-match candidates") + assert_in(b"_tcb", clean, "multi-match candidates") + # The prompt and partial input should be redrawn. + assert_in(b"--> ", clean, "multi-match prompt redraw") + # Ctrl-C to clear. + resp = send_get(master, b"\x03", timeout=0.05) + print("PASS: multiple-match completion") + passed += 1 + + # --- Test 5: Command execution --- + resp = send_get(master, b"print(42)\n", timeout=0.1) + assert_in(b"42", resp, "command execution") + print("PASS: command execution") + passed += 1 + + # --- Test 6: Terminal mode verification --- + # aiorepl calls _stdio_raw(False) before execute(), so during execution + # the terminal should be in original (non-raw) mode with LFLAG non-zero. + resp = send_get( + master, + b"import termios; print('lflag:', termios.tcgetattr(0)[3] != 0)\n", + timeout=0.1, + ) + assert_in(b"lflag: True", resp, "terminal mode") + print("PASS: terminal mode verification") + passed += 1 + + # --- Test 7: Ctrl-D exit --- + # Ctrl-D on empty line should exit aiorepl and return to standard REPL. + resp = send_get(master, b"\x04", timeout=0.1) + assert_in(b">>>", resp, "ctrl-d exit") + print("PASS: ctrl-d exit") + passed += 1 + + except TestFailure as e: + print(f"FAIL: {e}") + failed += 1 + except Exception as e: + print(f"ERROR: {type(e).__name__}: {e}") + failed += 1 + finally: + try: + p.kill() + except ProcessLookupError: + pass + p.wait() + os.close(master) + os.close(slave) + + print(f"\n{passed} passed, {failed} failed") + sys.exit(1 if failed else 0) + + +if __name__ == "__main__": + main() diff --git a/micropython/aiorepl/test_autocomplete.py b/micropython/aiorepl/test_autocomplete.py new file mode 100644 index 000000000..8db497c8c --- /dev/null +++ b/micropython/aiorepl/test_autocomplete.py @@ -0,0 +1,47 @@ +# Test tab completion logic used by aiorepl. +import sys +import micropython +import __main__ + +try: + micropython.repl_autocomplete +except AttributeError: + print("SKIP") + raise SystemExit + +# Test the autocomplete API contract that aiorepl depends on. + +# Single completion: keyword "import" +result = micropython.repl_autocomplete("impo") +print(repr(result)) + +# No match: returns empty string +result = micropython.repl_autocomplete("xyz_no_match_zzz") +print(repr(result)) + +# Multiple matches: returns None (candidates printed to stdout by C code). +# Create two globals sharing a prefix so autocomplete finds multiple matches. +__main__.tvar_alpha = 1 +__main__.tvar_beta = 2 +result = micropython.repl_autocomplete("tvar_") +del __main__.tvar_alpha +del __main__.tvar_beta +print("multiple:", repr(result)) + +# Attribute completion: "sys.ver" matches sys.version and sys.version_info. +# Common prefix is "version" so completion suffix is "sion". +result = micropython.repl_autocomplete("sys.ver") +print("attr:", repr(result)) + +# Test the whitespace-before-cursor logic used for tab-as-indentation. +# This validates the condition: cursor_pos > 0 and cmd[cursor_pos - 1] <= " " +test_cases = [ + ("x ", True), # space before cursor + ("x", False), # non-whitespace before cursor + ("\n", True), # newline counts as whitespace + ("", False), # empty line (cursor_pos == 0) +] +for cmd, expected in test_cases: + cursor_pos = len(cmd) + is_whitespace = cursor_pos > 0 and cmd[cursor_pos - 1] <= " " + print(cmd.encode(), is_whitespace == expected) diff --git a/micropython/aiorepl/test_autocomplete.py.exp b/micropython/aiorepl/test_autocomplete.py.exp new file mode 100644 index 000000000..f2ce95607 --- /dev/null +++ b/micropython/aiorepl/test_autocomplete.py.exp @@ -0,0 +1,10 @@ +'rt ' +'' + +tvar_alpha tvar_beta +multiple: None +attr: 'sion' +b'x ' True +b'x' True +b'\n' True +b'' True