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