198 lines
8.2 KiB
Python
198 lines
8.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
tool_bash 实时输出（增量流）单测 —— 纯函数级，无 UI 依赖
|
||
|
||
本轮改动核心：communicate(timeout=1) 阻塞式收集
|
||
→ stdout/stderr reader 线程 + 队列 + 主循环抽干回调 on_update
|
||
本测试锁死「改动没有破坏任何既有语义」+「输出真的是实时的」。
|
||
|
||
覆盖：
|
||
T1 增量流：on_update 被多次调用（不是结束时一次性给）
|
||
T2 实时性：第一块输出到达时刻 << 总耗时（证明是流式而非收尾补发）
|
||
T3 上下文文本格式完全不变（$ cmd / stdout / [stderr] / [exit N] (Ns)）
|
||
T4 stderr 实时流带 [stderr] 分隔标记，且最终文本仍有 [stderr] 段
|
||
T5 每 1 秒读秒回调 on_timer 仍然工作
|
||
T6 超时：杀进程树 + 「命令超时（>Ns）已终止」+ 实际耗时远小于命令时长
|
||
T7 中止：AbortSignal.aborted → 「操作已中止」
|
||
T8 50KB 截断仍然生效
|
||
T9 退出非 0 → is_error，且 exit code 写入 details
|
||
T10 空命令 → 参数校验错误
|
||
T11 无输出命令：不产生 on_update，但读秒正常
|
||
T12 最终文本 = 全部 stdout 拼接（不丢不重）
|
||
|
||
运行: PYTHONIOENCODING=utf-8 python tests/test_bash_stream.py
|
||
"""
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
import tempfile
|
||
import shutil
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from core.agent.tools import tool_bash # noqa: E402
|
||
from core.agent.types import AbortSignal # noqa: E402
|
||
|
||
RESULTS = []
|
||
|
||
|
||
def check(name, cond, extra=""):
|
||
RESULTS.append((name, bool(cond)))
|
||
print(f"{'PASS' if cond else 'FAIL'} {name} {extra if not cond else ''}", flush=True)
|
||
|
||
|
||
_TMP = tempfile.mkdtemp(prefix="hocode_bashstream_")
|
||
PY = sys.executable
|
||
|
||
|
||
def run(script=None, timeout=None, abort_after=None, shell_script=None):
|
||
"""把脚本写成临时文件再执行（避开 cmd 引号地狱）。
|
||
返回 (result, updates, first_t, dur, timer_calls)"""
|
||
path = None
|
||
if script is not None:
|
||
path = os.path.join(_TMP, f"s_{int(time.time() * 1000000) % 10**9}.py")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write(script)
|
||
cmd = f'"{PY}" -u "{path}"'
|
||
else:
|
||
cmd = shell_script
|
||
|
||
updates = []
|
||
first_t = [None]
|
||
t0 = time.time()
|
||
|
||
def on_update(text):
|
||
if first_t[0] is None:
|
||
first_t[0] = time.time() - t0
|
||
updates.append(text)
|
||
|
||
timers = []
|
||
|
||
def on_timer(elapsed, total):
|
||
timers.append((elapsed, total))
|
||
|
||
sig = AbortSignal()
|
||
if abort_after is not None:
|
||
import threading
|
||
|
||
def _aborter():
|
||
time.sleep(abort_after)
|
||
sig.abort("test")
|
||
threading.Thread(target=_aborter, daemon=True).start()
|
||
|
||
args = {"command": cmd}
|
||
if timeout is not None:
|
||
args["timeout"] = timeout
|
||
res = tool_bash("call_test", args, sig, on_update,
|
||
{"cwd": _TMP, "shell": True, "on_timer": on_timer})
|
||
dur = time.time() - t0
|
||
return res, updates, first_t[0], dur, timers
|
||
|
||
|
||
def text_of(res):
|
||
c = res.content
|
||
if isinstance(c, str):
|
||
return c
|
||
return "".join(x.get("text", "") for x in c if isinstance(x, dict))
|
||
|
||
|
||
# ======================================================================
|
||
try:
|
||
# ---------- T1/T2/T3/T12: 流式 + 上下文格式 ----------
|
||
res, ups, first_t, dur, timers = run(
|
||
"import sys, time\n"
|
||
"for i in range(3):\n"
|
||
" print('LINE%d' % i, flush=True)\n"
|
||
" time.sleep(0.5)\n", timeout=30)
|
||
txt = text_of(res)
|
||
check("T1.1 增量流被多次回调（>=3 块）", len(ups) >= 3, f"updates={len(ups)} {ups!r}")
|
||
check("T1.2 每块都是本次输出的一部分", all(u.strip() for u in ups), repr(ups[:3]))
|
||
check("T2.1 第一块在 1s 内到达（实时，不是收尾补发）",
|
||
first_t is not None and first_t < 1.0, f"first_t={first_t}")
|
||
check("T2.2 第一块明显早于总耗时",
|
||
first_t is not None and (dur - first_t) > 0.6, f"first={first_t} dur={dur:.2f}")
|
||
check("T3.1 上下文文本以 $ 命令开头", txt.startswith("$ "), repr(txt[:60]))
|
||
check("T3.2 含 [exit 0] 与耗时",
|
||
re.search(r"\[exit 0\] \(\d+\.\d+s\)", txt) is not None, repr(txt[-60:]))
|
||
check("T12.1 三行输出全部进入最终文本，不丢不重",
|
||
all(f"LINE{i}" in txt for i in range(3)) and txt.count("LINE0") == 1, repr(txt))
|
||
check("T3.3 is_error=False", res.is_error is False)
|
||
check("T3.4 details 带 exit_code", res.details.get("exit_code") == 0, str(res.details))
|
||
|
||
# ---------- T4: stderr 实时 + 最终段 ----------
|
||
res, ups, first_t, dur, timers = run(
|
||
"import sys\nprint('OUT1', flush=True)\nprint('ERR1', file=sys.stderr, flush=True)\n",
|
||
timeout=30)
|
||
txt = text_of(res)
|
||
joined = "".join(ups)
|
||
check("T4.1 实时流里有 [stderr] 分隔标记", "[stderr]" in joined, repr(joined))
|
||
check("T4.2 最终文本含 [stderr] 段与内容",
|
||
"[stderr]" in txt and "ERR1" in txt, repr(txt))
|
||
check("T4.3 stdout 内容也在最终文本里", "OUT1" in txt, repr(txt))
|
||
|
||
# ---------- T5/T11: 读秒 ----------
|
||
res, ups, first_t, dur, timers = run(
|
||
"import time\ntime.sleep(2.6)\nprint('done', flush=True)\n", timeout=30)
|
||
check("T5.1 长命令期间读秒回调（>=2 次）", len(timers) >= 2, f"timers={timers}")
|
||
check("T5.2 读秒总量正确", all(t[1] == 30 for t in timers), str(timers))
|
||
|
||
res, ups, first_t, dur, timers = run(
|
||
"import time\ntime.sleep(1.5)\n", timeout=30)
|
||
check("T11.1 无输出命令不产生 on_update", len(ups) == 0, repr(ups))
|
||
check("T11.2 无输出命令仍有读秒", len(timers) >= 1, str(timers))
|
||
|
||
# ---------- T6: 超时杀树 ----------
|
||
res, ups, first_t, dur, timers = run(
|
||
"import time\nprint('start', flush=True)\ntime.sleep(30)\n", timeout=2)
|
||
txt = text_of(res)
|
||
check("T6.1 超时文案", "命令超时（>2s）已终止" in txt, repr(txt))
|
||
check("T6.2 真的提前返回（耗时应 <10s）", dur < 10, f"dur={dur:.2f}")
|
||
check("T6.3 超时结果是 is_error", res.is_error is True)
|
||
|
||
# ---------- T7: 中止 ----------
|
||
res, ups, first_t, dur, timers = run(
|
||
"import time\ntime.sleep(30)\n", timeout=60, abort_after=1.5)
|
||
txt = text_of(res)
|
||
check("T7.1 中止文案", "操作已中止" in txt, repr(txt))
|
||
check("T7.2 中止后迅速返回（<8s）", dur < 8, f"dur={dur:.2f}")
|
||
|
||
# ---------- T8: 50KB 截断 ----------
|
||
res, ups, first_t, dur, timers = run(
|
||
"import sys\n"
|
||
"line = 'X' * 100 + '\\n'\n"
|
||
"for _ in range(1200):\n"
|
||
" sys.stdout.write(line)\n"
|
||
"sys.stdout.flush()\n", timeout=30)
|
||
txt = text_of(res)
|
||
check("T8.1 超过 50KB 触发截断标记", "[输出超过 50KB 已截断]" in txt, repr(txt[-80:]))
|
||
check("T8.2 截断后字节数 <= 50KB+余量",
|
||
len(txt.encode("utf-8")) <= 50 * 1024 + 200, str(len(txt.encode("utf-8"))))
|
||
|
||
# ---------- T9: 非 0 退出 ----------
|
||
res, ups, first_t, dur, timers = run("import sys\nsys.exit(3)\n", timeout=30)
|
||
txt = text_of(res)
|
||
check("T9.1 exit 3 写入文本", "[exit 3]" in txt, repr(txt))
|
||
check("T9.2 is_error=True", res.is_error is True)
|
||
check("T9.3 details.exit_code=3", res.details.get("exit_code") == 3, str(res.details))
|
||
|
||
# ---------- T10: 空命令 ----------
|
||
res = tool_bash("c", {"command": " "}, AbortSignal(), None,
|
||
{"cwd": _TMP, "shell": True})
|
||
check("T10.1 空命令被拒", "command 不能为空" in text_of(res), repr(text_of(res)))
|
||
|
||
# ---------- T13: 秒级短命令仍正常（回归：不因线程化而变慢/丢输出） ----------
|
||
res, ups, first_t, dur, timers = run(shell_script="echo hello_from_cmd")
|
||
txt = text_of(res)
|
||
check("T13.1 短命令输出正确", "hello_from_cmd" in txt, repr(txt))
|
||
check("T13.2 短命令耗时 <3s", dur < 3, f"dur={dur:.2f}")
|
||
check("T13.3 短命令实时流也拿到了输出", "hello_from_cmd" in "".join(ups), repr(ups))
|
||
|
||
finally:
|
||
shutil.rmtree(_TMP, ignore_errors=True)
|
||
|
||
failed = [n for n, ok in RESULTS if not ok]
|
||
print(f"\n===== {len(RESULTS) - len(failed)}/{len(RESULTS)} PASS =====", flush=True)
|
||
print("ALL PASS" if not failed else f"FAILED: {failed}", flush=True)
|
||
sys.exit(0 if not failed else 1)
|