sorrow404null
a7412824e0
Import the pre-repair source tree as the history baseline. Runtime data (data/), virtualenvs, bytecode caches and logs are gitignored so local secrets and user state stay out of the repo.
256 lines
13 KiB
Python
256 lines
13 KiB
Python
# -*- 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 不能为空'（旧版报'文件不存在: <cwd>'）",
|
||
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)
|