aiorepl: Add tab completion support.

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

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)
 
 

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.",
 )
 

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()

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)

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