254 lines
10 KiB
Python
254 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
P1-02 跨平台 shell 契约单测（纯函数级 + 真实进程级，无 UI 依赖）
|
||
|
||
覆盖：
|
||
A. 平台参数：Windows `cmd.exe /d /c` argv / Linux `/bin/bash -lc` argv；
|
||
Linux Popen 独立进程组（start_new_session），Windows 无该参数
|
||
B. 提示词平台段：通用正文完全相同、只插入对应平台段、互不串段；
|
||
load_system_prompt() 与文件内容一致（当前平台）
|
||
C. 进程树（真实进程，父+孙）：
|
||
- 当前平台：超时 → 父与孙都不存在
|
||
- 当前平台：主动中止 → 父与孙都不存在
|
||
- Linux 专属：独立进程组整组终止（Windows 上跳过）
|
||
D. kill_process_tree 安全边界：已退出的进程 / 无独立组 → 不抛异常、不误杀
|
||
|
||
运行: PYTHONIOENCODING=utf-8 python tests/test_cross_platform_shell.py
|
||
"""
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from core import platform_shell # noqa: E402
|
||
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)
|
||
|
||
|
||
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))
|
||
|
||
|
||
_TMP = tempfile.mkdtemp(prefix="haocode_crossshell_")
|
||
PY = sys.executable
|
||
IS_WIN = os.name == "nt"
|
||
|
||
# ======================================================================
|
||
# A. 平台参数（mock 平台，纯逻辑）
|
||
# ======================================================================
|
||
_orig_is_win = platform_shell.is_windows
|
||
|
||
|
||
def _mock_win(v):
|
||
platform_shell.is_windows = lambda: v
|
||
|
||
|
||
try:
|
||
_mock_win(True)
|
||
check("A1 Windows 命令 = cmd.exe /d /s /c \"<cmd>\"",
|
||
platform_shell.shell_command("echo hi")
|
||
== 'cmd.exe /d /s /c "echo hi"',
|
||
repr(platform_shell.shell_command("echo hi")))
|
||
check("A2 Windows popen_flags 无进程组参数",
|
||
platform_shell.popen_flags() == {}, repr(platform_shell.popen_flags()))
|
||
|
||
_mock_win(False)
|
||
check("A3 Linux argv = /bin/bash -lc <cmd>",
|
||
platform_shell.shell_command("echo hi") == ["/bin/bash", "-lc", "echo hi"],
|
||
repr(platform_shell.shell_command("echo hi")))
|
||
check("A4 Linux popen_flags 独立进程组",
|
||
platform_shell.popen_flags() == {"start_new_session": True},
|
||
repr(platform_shell.popen_flags()))
|
||
finally:
|
||
platform_shell.is_windows = _orig_is_win
|
||
|
||
# ======================================================================
|
||
# B. 提示词平台段
|
||
# ======================================================================
|
||
_PROMPT_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"SYSTEM_PROMPT.md")
|
||
with open(_PROMPT_FILE, "r", encoding="utf-8") as f:
|
||
_PROMPT_TEXT = f.read().strip()
|
||
|
||
check("B1 通用正文含占位符",
|
||
platform_shell.SHELL_SECTION_PLACEHOLDER in _PROMPT_TEXT)
|
||
|
||
_GENERIC_NO_LEAK = ("cmd.exe" not in _PROMPT_TEXT
|
||
and "/bin/bash" not in _PROMPT_TEXT)
|
||
check("B2 通用正文无平台泄漏（无 cmd.exe / 无 /bin/bash）", _GENERIC_NO_LEAK)
|
||
|
||
try:
|
||
_mock_win(True)
|
||
full_win = platform_shell.apply_platform_section(_PROMPT_TEXT)
|
||
sec_win = platform_shell.shell_prompt_section()
|
||
_mock_win(False)
|
||
full_lin = platform_shell.apply_platform_section(_PROMPT_TEXT)
|
||
sec_lin = platform_shell.shell_prompt_section()
|
||
finally:
|
||
platform_shell.is_windows = _orig_is_win
|
||
|
||
check("B3 Windows 段含 cmd.exe、无 /bin/bash",
|
||
"cmd.exe" in sec_win and "/bin/bash" not in sec_win)
|
||
check("B4 Linux 段含 /bin/bash -lc、无 cmd.exe",
|
||
"/bin/bash -lc" in sec_lin and "cmd.exe" not in sec_lin)
|
||
check("B5 只有对应平台段被插入",
|
||
full_win == _PROMPT_TEXT.replace(platform_shell.SHELL_SECTION_PLACEHOLDER, sec_win)
|
||
and full_lin == _PROMPT_TEXT.replace(platform_shell.SHELL_SECTION_PLACEHOLDER, sec_lin))
|
||
check("B6 通用正文两平台完全相同",
|
||
full_win.replace(sec_win, "§SEC§") == full_lin.replace(sec_lin, "§SEC§"))
|
||
check("B7 无占位符文本原样返回",
|
||
platform_shell.apply_platform_section("无占位符的兜底文本") == "无占位符的兜底文本")
|
||
|
||
# load_system_prompt()（当前平台，真实文件）
|
||
from core import llm_engine # noqa: E402 (PyQt6/openai 已装，无需 QApplication)
|
||
check("B8 load_system_prompt = 文件内容 + 当前平台段",
|
||
llm_engine.load_system_prompt()
|
||
== platform_shell.apply_platform_section(_PROMPT_TEXT))
|
||
|
||
# ======================================================================
|
||
# C. 进程树（真实进程：父 + 孙，心跳文件证明生死）
|
||
# ======================================================================
|
||
_CHILD = os.path.join(_TMP, "child.py")
|
||
with open(_CHILD, "w", encoding="utf-8") as f:
|
||
f.write(
|
||
"import sys, time\n"
|
||
"hb = open(sys.argv[1], 'a', encoding='utf-8')\n"
|
||
"t0 = time.time()\n"
|
||
"while time.time() - t0 < 30:\n"
|
||
" hb.write(f'{time.time():.3f}\\n')\n"
|
||
" hb.flush()\n"
|
||
" time.sleep(0.2)\n"
|
||
"open(sys.argv[2], 'w').write('done')\n"
|
||
)
|
||
|
||
|
||
def _spawn_parent(hb_path, done_path, out_path):
|
||
"""写一个父进程脚本（启动孙进程后挂 30s），返回 bash 命令字符串。"""
|
||
parent = os.path.join(_TMP, f"parent_{os.path.basename(hb_path)}.py")
|
||
with open(parent, "w", encoding="utf-8") as f:
|
||
f.write(
|
||
"import subprocess, sys, time\n"
|
||
"p = subprocess.Popen([sys.executable, "
|
||
+ repr(_CHILD) + ", sys.argv[1], sys.argv[2]], "
|
||
"stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)\n"
|
||
"time.sleep(30)\n"
|
||
)
|
||
cmd = f'"{PY}" -u "{parent}" "{hb_path}" "{done_path}"'
|
||
return cmd
|
||
|
||
|
||
def _hb_last(hb_path):
|
||
try:
|
||
with open(hb_path, "r", encoding="utf-8") as f:
|
||
lines = [l.strip() for l in f if l.strip()]
|
||
return float(lines[-1]) if lines else None
|
||
except FileNotFoundError:
|
||
return None
|
||
|
||
|
||
def _tree_dead(hb_path, done_path, quiesce=1.2):
|
||
"""等待静默后判断：孙进程不再有心跳、且未跑完 → 进程树已死。"""
|
||
time.sleep(quiesce)
|
||
last = _hb_last(hb_path)
|
||
return last is not None and (time.time() - last) > 0.6, os.path.exists(done_path)
|
||
|
||
|
||
def run_tree_case(tag, timeout=None, abort_after=None):
|
||
hb = os.path.join(_TMP, f"hb_{tag}.txt")
|
||
done = os.path.join(_TMP, f"done_{tag}.txt")
|
||
out = os.path.join(_TMP, f"out_{tag}.log")
|
||
cmd = _spawn_parent(hb, done, out)
|
||
|
||
updates, timers = [], []
|
||
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()
|
||
|
||
t0 = time.time()
|
||
res = tool_bash(f"call_{tag}", {"command": cmd, "timeout": timeout or 60}, sig,
|
||
lambda t: updates.append(t),
|
||
{"cwd": _TMP, "on_timer": lambda e, tt: timers.append((e, tt))})
|
||
dur = time.time() - t0
|
||
return res, dur, hb, done, timers
|
||
|
||
|
||
# ---- C1: 超时 → 父与孙都不存在 ----
|
||
res, dur, hb, done, timers = run_tree_case("timeout", timeout=3)
|
||
txt = text_of(res)
|
||
dead, finished = _tree_dead(hb, done)
|
||
check("C1.1 超时返回错误结果", res.is_error and "超时" in txt, f"{txt!r}")
|
||
check("C1.2 超时快速返回（<15s，含 taskkill 缓冲）", dur < 15, f"dur={dur:.1f}s")
|
||
check("C1.3 超时后父+孙都不存在（心跳停止且未跑完）", dead and not finished,
|
||
f"hb_last={_hb_last(hb)} finished={finished}")
|
||
check("C1.4 读秒滴答仍工作", len(timers) >= 2, f"timers={timers}")
|
||
|
||
# ---- C2: 主动中止 → 父与孙都不存在 ----
|
||
res, dur, hb, done, _timers2 = run_tree_case("abort", timeout=60, abort_after=1.5)
|
||
txt = text_of(res)
|
||
dead, finished = _tree_dead(hb, done)
|
||
check("C2.1 中止返回错误结果", res.is_error and "中止" in txt, f"{txt!r}")
|
||
check("C2.2 中止后父+孙都不存在（心跳停止且未跑完）", dead and not finished,
|
||
f"hb_last={_hb_last(hb)} finished={finished}")
|
||
|
||
# ---- C3: Linux 独立进程组整组终止（Windows 跳过） ----
|
||
if not IS_WIN:
|
||
proc = subprocess.Popen(
|
||
platform_shell.shell_command("sleep 30"),
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||
**platform_shell.popen_flags())
|
||
time.sleep(0.3)
|
||
try:
|
||
pgid = os.getpgid(proc.pid)
|
||
check("C3.1 start_new_session 生效（pgid == 子 pid）", pgid == proc.pid,
|
||
f"pgid={pgid} pid={proc.pid}")
|
||
platform_shell.kill_process_tree(proc, grace_s=1.0)
|
||
time.sleep(0.5)
|
||
check("C3.2 整组终止后子进程不存在",
|
||
proc.poll() is not None, f"poll={proc.poll()}")
|
||
except Exception as e:
|
||
try:
|
||
platform_shell.kill_process_tree(proc, grace_s=1.0)
|
||
except Exception:
|
||
pass
|
||
check("C3 Linux 进程组用例", False, repr(e))
|
||
else:
|
||
print("SKIP C3 Linux 进程组用例（当前为 Windows，Windows 用例见 C1/C2）", flush=True)
|
||
|
||
# ======================================================================
|
||
# D. kill_process_tree 安全边界
|
||
# ======================================================================
|
||
_p = subprocess.Popen([PY, "-c", "pass"])
|
||
_p.wait(timeout=5)
|
||
try:
|
||
platform_shell.kill_process_tree(_p, grace_s=0.5)
|
||
check("D1 已退出进程：不抛异常", True)
|
||
except Exception as e:
|
||
check("D1 已退出进程：不抛异常", False, repr(e))
|
||
|
||
try:
|
||
platform_shell.kill_process_tree(None)
|
||
check("D2 None 输入：不抛异常", True)
|
||
except Exception as e:
|
||
check("D2 None 输入：不抛异常", False, repr(e))
|
||
|
||
# ======================================================================
|
||
print(f"\n===== {sum(1 for _, ok in RESULTS if ok)}/{len(RESULTS)} passed =====", flush=True)
|
||
sys.exit(0 if all(ok for _, ok in RESULTS) else 1)
|