503 lines
20 KiB
Python
Executable File
503 lines
20 KiB
Python
Executable File
#!/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())
|