# -*- coding: utf-8 -*- """ 四大工具参数层单测（纯函数级，无 DB/无 UI 依赖） 覆盖本轮修复： T1-T9 参数归一化 + JSON Schema 校验（全错误上报 / null 归一化 / 轻量强制转换 / bool 漏洞） T10-T13 edit 参数预处理（legacy 单条 / edits 为 JSON 字符串） T14-T19 edit 区间规划（唯一性 / 重叠检测 / 空 oldText） T20-T25 read 参数钳制（limit 负数漏洞 / 越界 / 空文件） T26-T29 write + edit 原子写（无临时文件残留） T30-T31 before 钩子改参后重新校验 运行: PYTHONIOENCODING=utf-8 python tests/test_tool_params.py """ import os import sys import json import glob import tempfile import shutil sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from core.agent.tools import (validate_json_schema, normalize_and_coerce, # noqa: E402 _prepare_edit_args, _plan_edits, tool_read, tool_write, tool_edit, execute_tool_call, PreparedToolCall) from core.agent.types import (AgentTool, AgentToolResult, AbortSignal, # noqa: E402 ToolCall, AgentMessage) 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) # ====================================================================== # 测试用 schema（与 default_tools() 一致） # ====================================================================== S_READ = {"type": "object", "properties": {"path": {"type": "string"}, "offset": {"type": "integer"}, "limit": {"type": "integer"}}, "required": ["path"]} S_BASH = {"type": "object", "properties": {"command": {"type": "string"}, "timeout": {"type": "number"}}, "required": ["command"]} S_EDIT = {"type": "object", "properties": { "path": {"type": "string"}, "edits": {"type": "array", "items": {"type": "object", "properties": {"oldText": {"type": "string"}, "newText": {"type": "string"}}, "required": ["oldText"]}}}, "required": ["path", "edits"]} SG = AbortSignal() # ====================================================================== # T1-T9 校验层 # ====================================================================== def err_of(args, schema): return validate_json_schema(normalize_and_coerce(args, schema), schema) check("T1.缺必填参数", err_of({"offset": 1}, S_READ) == "缺少必填参数: path", err_of({"offset": 1}, S_READ)) check("T2.string 类型错误", "应为 string" in (err_of({"path": 123}, S_READ) or ""), err_of({"path": 123}, S_READ)) check("T3.bool 不得冒充 integer（旧版漏洞）", "应为 integer" in (err_of({"path": "a", "offset": True}, S_READ) or ""), err_of({"path": "a", "offset": True}, S_READ)) check("T4.bool 不得冒充 number（旧版漏洞）", "应为 number" in (err_of({"command": "x", "timeout": True}, S_BASH) or ""), err_of({"command": "x", "timeout": True}, S_BASH)) check("T5.整数 1.5 拒绝", "期望 integer" in (err_of({"path": "a", "offset": 1.5}, S_READ) or ""), err_of({"path": "a", "offset": 1.5}, S_READ)) check("T6.多余参数宽容（与 pi 一致）", err_of({"path": "a", "lines": 10}, S_READ) is None, err_of({"path": "a", "lines": 10}, S_READ)) _err2 = err_of({"path": 123, "offset": "bad"}, S_READ) or "" check("T7.多错全量上报（不再只报第 1 个）", _err2.count(";") >= 1 and "path" in _err2 and "offset" in _err2, _err2) check("T8.嵌套必填字段", err_of({"path": "a", "edits": [{"newText": "y"}]}, S_EDIT) == "参数 edits[0] 缺少必填字段 oldText", err_of({"path": "a", "edits": [{"newText": "y"}]}, S_EDIT)) check("T9.数组元素类型错误定位到下标", "edits[1]" in (err_of({"path": "a", "edits": [{"oldText": "x"}, {"oldText": 5}]}, S_EDIT) or ""), err_of({"path": "a", "edits": [{"oldText": "x"}, {"oldText": 5}]}, S_EDIT)) # ---- 归一化 / 强制转换 ---- check("T10.可选字段 null 被删除（不再误报类型错）", normalize_and_coerce({"path": "a", "offset": None}, S_READ) == {"path": "a"}) check("T11.数字字符串被转换 \"30\"→30", normalize_and_coerce({"command": "x", "timeout": "30"}, S_BASH)["timeout"] == 30) check("T12.integer 字段收 \"1.5\" 保持原值（交由校验报错）", normalize_and_coerce({"path": "a", "offset": "1.5"}, S_READ)["offset"] == "1.5") _orig = {"path": "a", "offset": None} normalize_and_coerce(_orig, S_READ) check("T13.归一化不修改入参", _orig == {"path": "a", "offset": None}, _orig) # ====================================================================== # T14-T15 edit 参数预处理 # ====================================================================== check("T14.legacy 单条形式 {oldText,newText} → edits[]", _prepare_edit_args({"path": "a", "oldText": "x", "newText": "y"}) == {"path": "a", "edits": [{"oldText": "x", "newText": "y"}]}) check("T15.edits 为 JSON 字符串 → 解析为数组", _prepare_edit_args({"path": "a", "edits": '[{"oldText":"x"}]'}) == {"path": "a", "edits": [{"oldText": "x"}]}) # ====================================================================== # T16-T19 edit 区间规划 # ====================================================================== _e, _sp = _plan_edits("hello world", [{"oldText": "world", "newText": "there"}]) check("T16.唯一命中 → 规划成功", _e is None and _sp[0][0] == 6, (_e, _sp)) check("T17.未命中 → 报错", "未找到匹配文本" in (_plan_edits("abc", [{"oldText": "zz"}])[0] or ""), _plan_edits("abc", [{"oldText": "zz"}])[0]) check("T18.重复命中 → 报错", "匹配到 2 处" in (_plan_edits("foo foo", [{"oldText": "foo"}])[0] or ""), _plan_edits("foo foo", [{"oldText": "foo"}])[0]) # 旧版会：两条对原始内容各自唯一 → 校验通过 → 应用时第 2 条已找不到（静默 no-op，仍报“已应用 2 处”） # 现在必须判定为重叠并整批拒绝 _e2 = _plan_edits("foo bar", [{"oldText": "foo bar", "newText": "foo BAR"}, {"oldText": "bar", "newText": "baz"}])[0] check("T19.区间重叠被检出（旧版静默 no-op 第 2 条）", "重叠" in (_e2 or ""), _e2) check("T19b.嵌套重叠被检出", "重叠" in (_plan_edits("abc", [{"oldText": "abc", "newText": "Z"}, {"oldText": "ab", "newText": "Q"}])[0] or "")) check("T19d.原文件不存在的 oldText 被拒（对齐 pi：对原始快照匹配）", "未找到匹配文本" in (_plan_edits("abc", [{"oldText": "ab", "newText": "aX"}, {"oldText": "aXb", "newText": "ZZ"}])[0] or "")) check("T19c.空 oldText 被拒（旧版空文件会静默插入）", "不能为空" in (_plan_edits("", [{"oldText": "", "newText": "INJ"}])[0] or ""), _plan_edits("", [{"oldText": "", "newText": "INJ"}])[0]) # ====================================================================== # T20-T25 read 参数钳制（临时文件，不碰项目数据） # ====================================================================== _TMP = tempfile.mkdtemp(prefix="haocode_tp_") CTX = {"cwd": _TMP} def _mk(name, text): p = os.path.join(_TMP, name) with open(p, "w", encoding="utf-8") as f: f.write(text) return p try: _mk("lines.txt", "".join(f"L{i}\n" for i in range(1, 11))) # 10 行 _mk("empty.txt", "") r = tool_read("t", {"path": "lines.txt", "limit": -5}, SG, None, CTX) _t20 = r.as_text() check("T20.limit 负数不再读全文件（旧版返回 10 行减 5）", "L1" in _t20 and "L2" not in _t20, _t20.replace("\n", " | ")[:120]) r = tool_read("t", {"path": "lines.txt", "limit": 0}, SG, None, CTX) check("T21.limit=0 钳到 1（旧版返回空块 + 无意义脚注）", "L1" in r.as_text() and "L2" not in r.as_text(), r.as_text().replace("\n", " | ")[:120]) r = tool_read("t", {"path": "lines.txt", "offset": 999}, SG, None, CTX) check("T22.offset 越界 → 明确提示（旧版'已显示 999–998 行'）", "超出文件范围" in r.as_text() and "共 10 行" in r.as_text(), r.as_text()) r = tool_read("t", {"path": "empty.txt"}, SG, None, CTX) check("T23.空文件 → 明确提示", "文件为空" in r.as_text(), r.as_text()) r = tool_read("t", {"path": ""}, SG, None, CTX) check("T24.path 为空 → 'path 不能为空'（旧版报'文件不存在: '）", r.is_error and "不能为空" in r.as_text(), r.as_text()) r = tool_read("t", {"path": "lines.txt", "offset": 3, "limit": 2}, SG, None, CTX) check("T25.正常行窗口不变（offset=3,limit=2 → L3,L4）", "L3" in r.as_text() and "L4" in r.as_text() and "L5" not in r.as_text() and "offset=5" in r.as_text(), r.as_text().replace("\n", " | ")[:140]) # ================================================================== # T26-T29 原子写 / edit 端到端 # ================================================================== r = tool_write("t", {"path": "w.txt", "content": "a\nb\n"}, SG, None, CTX) left = glob.glob(os.path.join(_TMP, ".hocode_w_*")) check("T26.write 成功且无临时文件残留", not r.is_error and not left, (r.as_text(), left)) tool_write("t", {"path": "w2.txt", "content": "foo bar"}, SG, None, CTX) r = tool_edit("t", {"path": "w2.txt", "edits": [{"oldText": "foo bar", "newText": "foo BAR"}, {"oldText": "bar", "newText": "baz"}]}, SG, None, CTX) check("T27.重叠 edit 整批拒绝且文件未被改动", r.is_error and "重叠" in r.as_text(), r.as_text()) r = tool_edit("t", {"path": "w.txt", "edits": [{"oldText": "a", "newText": "A"}, {"oldText": "b", "newText": "B"}]}, SG, None, CTX) with open(os.path.join(_TMP, "w.txt"), encoding="utf-8") as f: after = f.read() check("T28.多条不重叠 edit 一次性应用（对原始快照）", not r.is_error and after == "A\nB\n", (r.as_text(), repr(after))) r = tool_edit("t", {"path": "w.txt", "edits": [{"oldText": "", "newText": "X"}]}, SG, None, CTX) with open(os.path.join(_TMP, "w.txt"), encoding="utf-8") as f: after = f.read() check("T29.空 oldText 被拒且文件未变", r.is_error and after == "A\nB\n", (r.as_text(), repr(after))) # ================================================================== # T30-T31 before 钩子改参后重新校验 # ================================================================== def _echo(args): return AgentToolResult.text(f"echo:{json.dumps(args, sort_keys=True)}") def _echo5(tool_call_id, args, signal, on_update, ctx): return _echo(args) class _Cfg: tool_context = {"cwd": _TMP} before_tool_call = None after_tool_call = None etool = AgentTool(name="echo", description="d", parameters=S_READ, execute=_echo5) tc = ToolCall(id="c1", name="echo", arguments={"path": "a"}) cfg_bad = _Cfg() cfg_bad.before_tool_call = lambda payload, sig: {"args": {"path": 123}} res = execute_tool_call(PreparedToolCall(tool_call=tc, tool=etool, args={"path": "a"}), AgentMessage(role="assistant", content=""), cfg_bad, SG, None) check("T30.before 钩子返回非法参数 → 拒绝执行（pi 同款重校验）", res.is_error and "校验失败" in res.as_text(), res.as_text()) cfg_ok = _Cfg() cfg_ok.before_tool_call = lambda payload, sig: {"args": {"path": "b", "offset": "7"}} res = execute_tool_call(PreparedToolCall(tool_call=tc, tool=etool, args={"path": "a"}), AgentMessage(role="assistant", content=""), cfg_ok, SG, None) check("T31.before 钩子改参后归一化生效（\"7\"→7）", not res.is_error and '"offset": 7' in res.as_text(), res.as_text()) cfg_blk = _Cfg() cfg_blk.before_tool_call = lambda payload, sig: {"block": True, "reason": "nope"} res = execute_tool_call(PreparedToolCall(tool_call=tc, tool=etool, args={"path": "a"}), AgentMessage(role="assistant", content=""), cfg_blk, SG, None) check("T32.before 钩子 block 语义不变", res.is_error and "nope" in res.as_text(), res.as_text()) 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)