#!/usr/bin/env python3 """repl_edit_check.py - 验证 REPL 的行编辑行为 (方向键、Home/End、退格、历史) 只抓字节流是不够的: 光标移动是靠转义序列实现的, 必须把输出回放到一个 终端模型里才能知道"屏幕上最终长什么样". 所以这里带一个极简终端模拟器, 只实现本编辑器用到的那几条控制序列. """ import os import pty import re import select import sys import tempfile import time COLS = 80 ROWS = 200 class Term: """极简终端模型: 只实现 \\r \\n \\b、CSI 光标移动、CSI J/K 清屏""" def __init__(self, cols=COLS, rows=ROWS): self.cols, self.rows = cols, rows self.grid = [[' '] * cols for _ in range(rows)] self.row = self.col = 0 def _clear_from_cursor(self): for c in range(self.col, self.cols): self.grid[self.row][c] = ' ' for r in range(self.row + 1, self.rows): for c in range(self.cols): self.grid[r][c] = ' ' def feed(self, data): i, n = 0, len(data) while i < n: ch = data[i] if ch == '\x1b': j = i + 1 if j < n and data[j] in '[O': j += 1 start = j # CSI 的参数可以包含 ? < > ! 等私有标记, 一律扫到终结符为止 while j < n and not ('@' <= data[j] <= '~'): j += 1 params = data[start:j] final = data[j] if j < n else '' num = int(params) if params.isdigit() else 1 if final == 'A': self.row = max(0, self.row - num) elif final == 'B': self.row = min(self.rows - 1, self.row + num) elif final == 'C': self.col = min(self.cols - 1, self.col + num) elif final == 'D': self.col = max(0, self.col - num) elif final == 'J': self._clear_from_cursor() elif final == 'K': for c in range(self.col, self.cols): self.grid[self.row][c] = ' ' i = j + 1 continue i += 1 continue if ch == '\r': self.col = 0 elif ch == '\n': self.row = min(self.rows - 1, self.row + 1) self.col = 0 elif ch == '\b': self.col = max(0, self.col - 1) elif ch == '\x07': pass elif ch == '\t': self.col = min(self.cols - 1, (self.col // 8 + 1) * 8) else: self.grid[self.row][self.col] = ch self.col += 1 if self.col >= self.cols: self.col = 0 self.row = min(self.rows - 1, self.row + 1) i += 1 def lines(self): return [''.join(r).rstrip() for r in self.grid] def nonempty(self): ls = self.lines() last = max((i for i, l in enumerate(ls) if l), default=-1) return ls[:last + 1] def last_prompt_line(self): for line in reversed(self.nonempty()): if line.startswith('ligent> '): return line return '' def to_bytes(s): # 把源码里写的十六进制转义与真实非 ASCII 字符一起转成字节. # 不能直接用 unicode_escape: 一条串里既有转义又有中文时, latin-1 那步会报错. if isinstance(s, bytes): return s out = bytearray() i = 0 while i < len(s): if s[i] == '\\' and i + 1 < len(s): nxt = s[i + 1] if nxt == 'x' and i + 4 <= len(s): try: out.append(int(s[i + 2:i + 4], 16)) i += 4 continue except ValueError: pass simple = {'n': 10, 'r': 13, 't': 9, '\\': 92} if nxt in simple: out.append(simple[nxt]) i += 2 continue out.extend(s[i].encode('utf-8')) i += 1 return bytes(out) def run(binary, args, keys, keyset_settle=0.10, settle=0.45, idle=0.03): pid, fd = pty.fork() if pid == 0: os.environ['TERM'] = 'xterm-256color' os.execv(binary, [binary] + args) os._exit(127) out = bytearray() # 行编辑器每次回到提示符都会打开括号粘贴 —— 用它当"可以开始敲了"的信号 PROMPT_READY = b"\x1b[?2004h" ready_seen = 0 tail = b"" def note(chunk): nonlocal tail, ready_seen buf = tail + chunk ready_seen += buf.count(PROMPT_READY) tail = buf[-(len(PROMPT_READY) - 1):] def drain(timeout, idle=idle, until_ready=False): """安静下来就返回, 不把整个 timeout 睡满. idle 取得很小 (0.03s)是有依据的: 行编辑器每次按键都在一个 fflush 里把整屏重绘写完, 所以一次 read 之后基本就没了. 而提交输入触发请求 时, 等待动画每 250ms 一帧, 永远不会安静, 于是照样等到 timeout —— 那条路径的行为和以前一致, 不会因为 idle 调小而抢跑. """ want = ready_seen + 1 if until_ready else None end = time.time() + timeout last = time.time() while time.time() < end: r, _, _ = select.select([fd], [], [], 0.02) if r: try: d = os.read(fd, 65536) except OSError: return if not d: return out.extend(d) note(d) last = time.time() if want is not None and ready_seen >= want: return elif time.time() - last > idle: return drain(settle, until_ready=True) for k in keys: os.write(fd, to_bytes(k)) drain(keyset_settle) drain(settle) # 刻意不发 Ctrl-C (它会在光标处打出 ^C 标记并覆盖两个字), # 直接关掉 pty: 最后那次重绘就是缓冲区的真值 try: os.close(fd) except OSError: pass try: os.waitpid(pid, 0) except ChildProcessError: pass return bytes(out) def wait_ready(fd, timeout=3.0): """等到行编辑器回到提示符 (它会打开括号粘贴). 裸 pty 会话里不要用 time.sleep 猜时间: 猜短了会 flaky, 猜长了每处都在 白等. 进程退出 (EOF)也算等到. """ READY = b"\x1b[?2004h" end = time.time() + timeout tail = b"" while time.time() < end: r, _, _ = select.select([fd], [], [], 0.02) if not r: continue try: d = os.read(fd, 65536) except OSError: return False if not d: return False buf = tail + d if READY in buf: return True tail = buf[-(len(READY) - 1):] return False LEFT, RIGHT = '\\x1b[D', '\\x1b[C' UP, DOWN = '\\x1b[A', '\\x1b[B' HOME_END = ('\\x1b[H', '\\x1b[F') BS, CTRLC = '\\x7f', '\\x03' def main(): binary = sys.argv[1] if len(sys.argv) > 1 else './ligent' base = sys.argv[2] if len(sys.argv) > 2 else None # 编辑测试不会真的发请求, 但程序在缺 key 时会直接退出, 所以必须给一个 # 默认关掉确认与自动存档: 它们会打断被测行为. # 专门测确认的用例用 --extra 显式打开. args = ['--no-color', '--no-confirm', '--no-auto-save', '--api-key', 'test', '--base-url', base or 'http://127.0.0.1:1/v1'] # 被驱动的实例会读写输入历史. 整轮都指到临时文件: 既不往用户真实的 # ligent.history 里写测试垃圾, 方向键用例的结果也不会取决于开发者机器 # 上碰巧有什么历史. scratch = os.path.join(tempfile.gettempdir(), "ligent_edit_hist_%d" % os.getpid()) os.environ["LIGENT_HISTORY"] = scratch cases = [] # 方向键左移后插入: 光标必须落在中间 cases.append(("左方向键移动光标后插入", ['hello world'] + [LEFT] * 5 + ['X'], 'ligent> hello Xworld')) cases.append(("右方向键回到末尾", ['hello world'] + [LEFT] * 5 + [RIGHT, RIGHT, 'X'], 'ligent> hello woXrld')) cases.append(("Home 回到行首后插入", ['world', '\\x01', 'hello '], 'ligent> hello world')) cases.append(("CSI Home 序列 (\\x1b[H)", ['world', HOME_END[0], 'hello '], 'ligent> hello world')) cases.append(("End 序列 (\\x1b[F)回到末尾", ['world', HOME_END[0], 'hello ', HOME_END[1], '!'], 'ligent> hello world!')) cases.append(("行首再左移不越界", ['abc', HOME_END[0], LEFT, LEFT, 'Z'], 'ligent> Zabc')) cases.append(("退格删除", ['helloX', BS], 'ligent> hello')) cases.append(("中间退格删的是光标前的字符", ['abcd', LEFT, LEFT, BS], 'ligent> acd')) cases.append(("Delete 键删的是光标处的字符", ['abcd', LEFT, LEFT, '\\x1b[3~'], 'ligent> abd')) cases.append(("Ctrl-K 删到行尾", ['abcd', LEFT, LEFT, '\\x0b'], 'ligent> ab')) cases.append(("Ctrl-U 删到行首", ['abcd', LEFT, LEFT, '\\x15'], 'ligent> cd')) PASTE = '\\x1b[200~' PASTE_END = '\\x1b[201~' cases.append(("粘贴多行: 整段插入, 不逐行提交", [PASTE + 'line1\nline2\nline3' + PASTE_END], ['ligent> line1', 'line2', 'line3'])) cases.append(("粘贴后可继续编辑", [PASTE + 'aaa\nbbb' + PASTE_END, 'X'], ['ligent> aaa', 'bbbX'])) cases.append(("粘贴内容里含中文", [PASTE + '\u4e2d\u6587\n\u7b2c\u4e8c\u884c' + PASTE_END], ['ligent> 中文', '第二行'])) cases.append(("粘贴多行后 Home 只回到本行行首", [PASTE + 'aaa\nbbb\nccc' + PASTE_END, '\\x01', 'X'], ['ligent> aaa', 'bbb', 'Xccc'])) cases.append(("粘贴多行后 End 只到本行行尾", [PASTE + 'aaa\nbbb\nccc' + PASTE_END, '\\x01', '\\x05', 'Y'], ['ligent> aaa', 'bbb', 'cccY'])) cases.append(("中文按字符移动光标", ['中文abc'] + [LEFT] * 3 + ['X'], 'ligent> 中文Xabc')) # 粘贴进来的原始字节里可能夹着控制字符. read_paste 是原样收下的 (不走 # classify), 所以必须清洗: 终端按制表位展开 \t, 编辑器只当 1 列; # ESC 会被终端真的执行掉; \r 把光标拉回行首. cases.append(("粘贴里的制表符按制表位展开", [PASTE + 'a\\tb' + PASTE_END, 'X'], 'ligent> ' + 'a' + ' ' * 7 + 'bX')) cases.append(("制表符展开后光标位置正确 (左移再插入)", [PASTE + 'a\\tb' + PASTE_END, LEFT, 'X'], 'ligent> ' + 'a' + ' ' * 7 + 'Xb')) cases.append(("粘贴里的 ESC 序列被丢弃, 不让它执行", [PASTE + 'before\\x1b[2Jafter' + PASTE_END], 'ligent> beforeafter')) cases.append(("粘贴里的 CRLF 归一成 LF", [PASTE + 'a\\r\\nb' + PASTE_END], ['ligent> a', 'b'])) fail = 0 for name, keys, want in cases: raw = run(binary, args, keys) term = Term() term.feed(raw.decode('utf-8', 'replace')) if isinstance(want, list): tail = term.nonempty()[-len(want):] got = '\n'.join(tail) ok = (tail == want) want = '\n'.join(want) else: got = term.last_prompt_line() ok = (got == want) # 旧实现会把方向键回显成字面量 ^[[D literal_escapes = '^[[' in raw.decode('utf-8', 'replace') if not ok or literal_escapes: fail += 1 print(" ✗ %s" % name) print(" 期望: %r" % want) print(" 实际: %r" % got) if literal_escapes: print(" 输出里出现了字面量转义文本 ^[[ (方向键没被识别)") else: print(" ✓ %s" % name) # 提示符带颜色时, 那串转义序列不能占列宽 —— 光标定位必须和不着色时一致. # 上面的用例都带 --no-color, 所以这里单独补一组彩色的. color_args = [a for a in args if a != '--no-color'] + ['--color'] color_cases = [ ("彩色提示符: 左移后插入", ['hello world'] + [LEFT] * 5 + ['X'], 'ligent> hello Xworld'), ("彩色提示符: Home 后插入", ['world', '\\x01', 'hello '], 'ligent> hello world'), ("彩色提示符: End 后追加", ['abc', '\\x05', '!'], 'ligent> abc!'), ("彩色提示符: 中文按字符移动", ['中文abc'] + [LEFT] * 3 + ['X'], 'ligent> 中文Xabc'), ("彩色提示符: 退格", ['abcd', '\\x7f'], 'ligent> abc'), ] for name, keys, want in color_cases: raw = run(binary, color_args, keys) term = Term() term.feed(raw.decode('utf-8', 'replace')) got = term.last_prompt_line() if got == want: print(" ✓ %s" % name) else: fail += 1 print(" ✗ %s" % name) print(" 期望: %r" % want) print(" 实际: %r" % got) # 上面那些用例只比屏幕内容, 而颜色在终端模型里是被剥掉的 —— 包没包住 # 输入的字看不出来. 所以这里直接查字节: 绿色区间必须一直延续到用户敲的 # 内容之后. 只包提示符的话, 提示符自带的 reset 会让输入掉回默认色. raw = run(binary, color_args, ['greenme']) txt = raw.decode('utf-8', 'replace') start = txt.rfind('\x1b[32m') tail = txt[start:] if start >= 0 else '' end = tail.find('\x1b[0m') inside = tail[:end] if start >= 0 and end >= 0 else '' if 'greenme' in inside: print(" ✓ 输入的字也被包在绿色里") else: fail += 1 print(" ✗ 输入的字也被包在绿色里") print(" 绿色区间: %r" % inside) # 粘贴多行后提交: 必须作为"一条"消息发出去 (回显里能看到内嵌换行) if base: raw = run(binary, args, [PASTE + 'ECHO_TEST 第一行\n第二行\n第三行' + PASTE_END, '\\r'], keyset_settle=1.2, settle=0.8, idle=0.5) term = Term() term.feed(raw.decode('utf-8', 'replace')) joined = '\n'.join(term.nonempty()) if 'ECHO[ECHO_TEST 第一行\\n第二行\\n第三行]' in joined: print(" ✓ 粘贴的多行作为一条消息提交") else: fail += 1 print(" ✗ 粘贴的多行作为一条消息提交") print(" 屏幕尾部: %r" % term.nonempty()[-6:]) # 上下方向键取历史: 需要真的提交一行, 所以只有在给了服务端时才测 if base: raw = run(binary, args, ['recalldemo', '\\r'], keyset_settle=0.8, settle=0.6, idle=0.5) term = Term() term.feed(raw.decode('utf-8', 'replace')) raw2 = run(binary, args, ['firstline', '\\r', '\\x1b[A'], keyset_settle=0.8, settle=0.6, idle=0.5) term2 = Term() term2.feed(raw2.decode('utf-8', 'replace')) lines = [l for l in term2.nonempty() if l.startswith('ligent> ')] if lines and 'firstline' in lines[-1]: print(" ✓ 上方向键取回上一条输入") else: fail += 1 print(" ✗ 上方向键取回上一条输入") print(" 提示符行: %r" % lines[-3:]) cases.append(("history", [], "ligent> firstline")) # 历史要跨进程: 上一个实例敲过的, 新实例按上方向键就能翻到. # 用独立的 LIGENT_HISTORY, 以免受其他用例和真实历史文件影响. hist = os.path.join(tempfile.gettempdir(), "ligent_hist_check_%d" % os.getpid()) for p in (hist, hist + ".tmp"): if os.path.exists(p): os.unlink(p) os.environ["LIGENT_HISTORY"] = hist try: # 实时性: 进程还活着、还没退出, 文件里就该已经有这条了 pid, fd = pty.fork() if pid == 0: os.environ['TERM'] = 'xterm-256color' os.execv(binary, [binary] + args) os._exit(127) wait_ready(fd) # 等横幅画完、回到提示符 os.write(fd, b'live_append\r') wait_ready(fd) # 等这条提交并落盘 content = "" if os.path.exists(hist): with open(hist, encoding='utf-8') as fh: content = fh.read() if 'live_append' in content: print(" ✓ 提交后立刻落盘 (进程未退出)") else: fail += 1 print(" ✗ 提交后立刻落盘 (进程未退出)") print(" 文件内容: %r" % content) # 继续敲两条再正常退出, 供下一个实例翻 for k in ('persist_one\r', '/think high\r'): os.write(fd, k.encode()) wait_ready(fd) os.write(fd, b'/exit\r') wait_ready(fd) # 退出后这里会读到 EOF try: os.close(fd) except OSError: pass try: os.waitpid(pid, 0) except OSError: pass # 文件里依次是 live_append / persist_one / /think high / /exit, # 所以按三次上方向键才轮到 persist_one raw3 = run(binary, args, [UP, UP, UP], keyset_settle=0.3, settle=0.6, idle=0.03) term3 = Term() term3.feed(raw3.decode('utf-8', 'replace')) lines3 = [l for l in term3.nonempty() if l.startswith('ligent> ')] if lines3 and 'persist_one' in lines3[-1]: print(" ✓ 新实例能翻到上一个实例的历史") else: fail += 1 print(" ✗ 新实例能翻到上一个实例的历史") print(" 提示符行: %r" % lines3[-3:]) cases.append(("history-persist", [], "ligent> persist_one")) finally: for p in (hist, hist + ".tmp"): if os.path.exists(p): os.unlink(p) os.environ["LIGENT_HISTORY"] = scratch # 还给整轮共用的那个 # Ctrl-C 应当丢弃当前行并回到新提示符 raw = run(binary, args, ['garbage', CTRLC, 'typed', CTRLC], idle=0.5) term = Term() term.feed(raw.decode('utf-8', 'replace')) prompts = [l for l in term.nonempty() if l.startswith('ligent> ')] # 期望: Ctrl-C 丢掉了 "garbage", 之后那一行只含 "typed" last = prompts[-1] if prompts else '' ok = len(prompts) >= 2 and 'garbage' not in last and 'typed' in last if not ok: fail += 1 print(" ✗ Ctrl-C 丢弃当前行并开新行") print(" 提示符行: %r" % prompts[-3:]) else: print(" ✓ Ctrl-C 丢弃当前行并开新行") total = len(cases) + len(color_cases) + 2 print() print("通过 %d 项, 失败 %d 项" % (total - fail, fail)) for p in (scratch, scratch + ".tmp"): if os.path.exists(p): os.unlink(p) os.environ.pop("LIGENT_HISTORY", None) return 1 if fail else 0 if __name__ == '__main__': sys.exit(main())