686aaaa421
冻结地图1二十分钟纵切版的平衡、证据与金标登记。 新增 ReferenceAsset/2 清单、策略、release、受信快照及 CLI/Service/acceptance provenance /4 消费链；保持 survivor live、R1 签名与部署关闭。
435 lines
16 KiB
Python
435 lines
16 KiB
Python
"""
|
||
test_toolkit.py — U2 工具底座（read/list/write）+ 形状门三态。
|
||
|
||
形状门：define 单对象 FAIL / 干净源 PASS / drawButton 赋值改签名后 PASS（#5 回吐命中矩形）经 shell-out 现有 node tools.mjs check —
|
||
便宜档与 Node 侧跑同一份门代码，口径必然一致。真 shell-out node（确定性命令，非 LLM gen 循环，前台串行）。
|
||
|
||
跑：cheap-worker/.venv/bin/python cheap-worker/tests/test_toolkit.py
|
||
"""
|
||
|
||
import asyncio
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
|
||
import cheap_run
|
||
import cheap_toolkit
|
||
from cheap_toolkit import CheapSession, build_toolkit
|
||
|
||
_GAME_RUNTIME = cheap_run._GAME_RUNTIME
|
||
_AMODEL_GEN = cheap_run._AMODEL_GEN
|
||
GID = "cheap-tk-test" # gitignore: amgen-cheap-* / _wg1-gen/cheap-*
|
||
|
||
|
||
def _scaffold(gid: str = GID) -> None:
|
||
"""shell-out node tools.mjs scaffold-saa 起一个干净工程（SAA 信封）。"""
|
||
subprocess.run(
|
||
["node", str(_AMODEL_GEN / "tools.mjs"), "scaffold-saa", gid],
|
||
cwd=str(_GAME_RUNTIME), check=True, capture_output=True, text=True,
|
||
)
|
||
|
||
|
||
def _write_logic(src: str, gid: str = GID) -> None:
|
||
r = cheap_run.write_file(gid, f"game-runtime/games/amgen-{gid}/src/game-logic.js", src)
|
||
assert r["ok"], r
|
||
|
||
|
||
# ── 形状门合成输入：define 触发目标 footgun；drawButton 改签名后赋值消费已合法（#5 回吐命中矩形），其余 check 门尽量满足以免掩盖 ──
|
||
_DRAWBUTTON_RETURN_OK = """export function createGame({ plugins, bundle, viewport }) {
|
||
const { hudUi } = plugins;
|
||
return {
|
||
init(boot) { this.ctx = boot.ctx; },
|
||
update(dt) { bundle.tick(dt); },
|
||
render(g) { const btn = hudUi.drawButton(g, { x: 0, y: 0, w: 10, h: 10 }, { label: 'x' }); if (btn) { this.x = 1; } },
|
||
destroy() {},
|
||
handleTap(x, y) { this.tapped = true; },
|
||
_forensicsView() { return { state() { return { phase: 'menu', score: 0 }; }, measures() { return {}; } }; },
|
||
};
|
||
}
|
||
"""
|
||
|
||
_DEFINE_BAD = """export function createGame({ plugins, bundle, viewport }) {
|
||
const { sceneFsm } = plugins;
|
||
return {
|
||
init(boot) { this.ctx = boot.ctx; sceneFsm.define({ menu: {}, play: {} }); },
|
||
update(dt) { bundle.tick(dt); },
|
||
render(g) {},
|
||
destroy() {},
|
||
handleTap(x, y) { this.tapped = true; },
|
||
_forensicsView() { return { state() { return { phase: 'menu', score: 0 }; }, measures() { return {}; } }; },
|
||
};
|
||
}
|
||
"""
|
||
|
||
|
||
def test_read_file_ok():
|
||
r = cheap_run.read_file("game-runtime/games/_template/src/game-logic.js")
|
||
assert r["ok"] and "createGame" in r["content"]
|
||
|
||
|
||
def test_read_out_of_repo():
|
||
r = cheap_run.read_file("../../../etc/hosts")
|
||
assert not r["ok"] and "越界" in r["error"]
|
||
|
||
|
||
def test_list_dir():
|
||
r = cheap_run.list_dir("game-runtime/games/_template/src")
|
||
assert r["ok"] and "game-logic.js" in r["entries"]
|
||
|
||
|
||
def test_write_l1_rejected():
|
||
_scaffold()
|
||
r = cheap_run.write_file(GID, f"game-runtime/games/amgen-{GID}/src/host-config.js", "x")
|
||
assert not r["ok"] and "L1" in r["error"]
|
||
|
||
|
||
def test_write_l3_ok():
|
||
_scaffold()
|
||
r = cheap_run.write_file(GID, f"game-runtime/games/amgen-{GID}/src/game-logic.js", "// hi\n")
|
||
assert r["ok"] and r["bytes"] > 0
|
||
|
||
|
||
def test_check_shape_drawbutton_ok():
|
||
# 阶段一B #5:drawButton 改返命中矩形后,const btn = drawButton(...); if (btn) 是合法直觉写法、形状门不再拦。
|
||
_scaffold()
|
||
_write_logic(_DRAWBUTTON_RETURN_OK)
|
||
r = cheap_run.check(GID)
|
||
assert r["ok"], f"drawButton 赋值消费应通过(改签名后返命中矩形):{r.get('output', '')}"
|
||
|
||
|
||
def test_check_shape_define_fail():
|
||
_scaffold()
|
||
_write_logic(_DEFINE_BAD)
|
||
r = cheap_run.check(GID)
|
||
assert not r["ok"], "define 单对象批量应触发形状门②"
|
||
assert "define" in r["output"]
|
||
|
||
|
||
def test_check_and_build_clean_pass():
|
||
"""干净源（_template 合法游戏改一行使 ≠模板）应 check+build 全 PASS——证形状门/各门对合法游戏不误伤。"""
|
||
_scaffold()
|
||
tmpl = (cheap_run._TEMPLATE_DIR / "src" / "game-logic.js").read_text(encoding="utf-8")
|
||
_write_logic("// cheap-worker U2 clean-pass 测试：在 _template 合法游戏基础上改一行使 ≠模板\n" + tmpl)
|
||
rc = cheap_run.check(GID)
|
||
assert rc["ok"], "干净源应 check PASS，实际 FAIL：\n" + rc["output"]
|
||
rb = cheap_run.build(GID)
|
||
assert rb["ok"], "干净源应 build PASS，实际 FAIL：\n" + rb["output"]
|
||
|
||
|
||
def _tool_function(session: CheapSession, name: str):
|
||
"""从真实 Toolkit 取出指定函数，测试仍通过 AgentScope 的生产组装路径。"""
|
||
toolkit = build_toolkit(session)
|
||
for group in toolkit.tool_groups:
|
||
for tool in group.tools:
|
||
if tool.name == name:
|
||
return tool._func
|
||
raise AssertionError(f"Toolkit 缺少工具：{name}")
|
||
|
||
|
||
def _call_tool(session: CheapSession, name: str, **kwargs):
|
||
"""同步测试辅助：执行生产 Toolkit 中的 async 工具函数。"""
|
||
return asyncio.run(_tool_function(session, name)(**kwargs))
|
||
|
||
|
||
def test_reference_snapshot_read_ignores_rewritten_disk(monkeypatch, tmp_path):
|
||
"""快照捕获后磁盘同路径改写，read_file 仍只能返回旧 bytes。"""
|
||
disk_file = tmp_path / "assets" / "gold" / "guide.md"
|
||
disk_file.parent.mkdir(parents=True)
|
||
disk_file.write_text("captured", encoding="utf-8")
|
||
session = CheapSession(
|
||
game_id="snapshot-read",
|
||
reference_files={"assets/gold/guide.md": b"captured"},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
disk_file.write_text("rewritten", encoding="utf-8")
|
||
calls = []
|
||
|
||
def live_read(path):
|
||
calls.append(path)
|
||
return {"ok": True, "content": disk_file.read_text(encoding="utf-8"), "truncated": False}
|
||
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "read_file", live_read)
|
||
|
||
assert _call_tool(session, "read_file", path="assets/gold/guide.md") == "captured"
|
||
assert calls == []
|
||
|
||
|
||
def test_reference_snapshot_rejects_unsigned_read_without_io(monkeypatch):
|
||
"""受保护根内未列入快照的文件必须拒绝，且不得触发活目录读取。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-read-deny",
|
||
reference_files={"assets/gold/allowed.txt": b"allowed"},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
calls = []
|
||
|
||
def live_read(path):
|
||
calls.append(path)
|
||
return {"ok": True, "content": "unsigned", "truncated": False}
|
||
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "read_file", live_read)
|
||
|
||
result = _call_tool(session, "read_file", path="assets/gold/unsigned.txt")
|
||
assert result.startswith("ERROR:")
|
||
assert "unsigned" not in result
|
||
assert calls == []
|
||
|
||
|
||
def test_reference_snapshot_lists_only_virtual_children_without_io(monkeypatch):
|
||
"""受保护目录只投影快照中的直接子项和虚拟子目录，不读取磁盘新增项。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-list",
|
||
reference_files={
|
||
"assets/gold/allowed.txt": b"allowed",
|
||
"assets/gold/sub/child.txt": b"child",
|
||
},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
calls = []
|
||
|
||
def live_list(path):
|
||
calls.append(path)
|
||
return {"ok": True, "entries": ["unsigned.txt"]}
|
||
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "list_dir", live_list)
|
||
|
||
assert _call_tool(session, "list_dir", path="assets/gold") == "allowed.txt\nsub/"
|
||
assert _call_tool(session, "list_dir", path="assets/gold/sub") == "child.txt"
|
||
assert calls == []
|
||
|
||
|
||
def test_reference_prefix_similar_path_keeps_live_read_semantics(monkeypatch):
|
||
"""路径段相似但不属于受保护根时，仍走既有 cheap_run 读取语义。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-prefix",
|
||
reference_files={"assets/gold/allowed.txt": b"snapshot"},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
calls = []
|
||
|
||
def live_read(path):
|
||
calls.append(path)
|
||
return {"ok": True, "content": "live-golden", "truncated": False}
|
||
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "read_file", live_read)
|
||
|
||
assert _call_tool(session, "read_file", path="assets/golden/allowed.txt") == "live-golden"
|
||
assert calls == ["assets/golden/allowed.txt"]
|
||
|
||
|
||
def test_reference_path_escape_cannot_bypass_protected_root(monkeypatch):
|
||
"""回到仓根的路径别名也必须拒绝，不能绕过受保护根读取活目录。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-escape",
|
||
reference_files={"assets/gold/allowed.txt": b"snapshot"},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
calls = []
|
||
|
||
def live_read(path):
|
||
calls.append(path)
|
||
return {"ok": True, "content": "disk", "truncated": False}
|
||
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "read_file", live_read)
|
||
|
||
result = _call_tool(session, "read_file", path="../games-development-ai/assets/gold/unsigned.txt")
|
||
assert result == "ERROR: 路径必须是仓内相对路径"
|
||
assert calls == []
|
||
|
||
|
||
def test_empty_reference_snapshot_preserves_existing_read_and_list(monkeypatch):
|
||
"""未提供快照时保持现有 read_file/list_dir 回落行为。"""
|
||
session = CheapSession(game_id="snapshot-empty")
|
||
read_calls = []
|
||
list_calls = []
|
||
|
||
def live_read(path):
|
||
read_calls.append(path)
|
||
return {"ok": True, "content": "live-content", "truncated": False}
|
||
|
||
def live_list(path):
|
||
list_calls.append(path)
|
||
return {"ok": True, "entries": ["live.txt"]}
|
||
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "read_file", live_read)
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "list_dir", live_list)
|
||
|
||
assert _call_tool(session, "read_file", path=".agents/skills/example.md") == "live-content"
|
||
assert _call_tool(session, "list_dir", path=".agents/skills") == "live.txt"
|
||
assert read_calls == [".agents/skills/example.md"]
|
||
assert list_calls == [".agents/skills"]
|
||
|
||
|
||
def test_reference_snapshot_read_keeps_200kb_truncation_marker(monkeypatch):
|
||
"""快照 bytes 仍按现有 200KB 规则解码并带截断标记。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-truncate",
|
||
reference_files={"assets/gold/large.txt": b"x" * (200 * 1024 + 1)},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
monkeypatch.setattr(
|
||
cheap_toolkit.cheap_run,
|
||
"read_file",
|
||
lambda path: (_ for _ in ()).throw(AssertionError("受保护快照不应调用 cheap_run.read_file")),
|
||
)
|
||
|
||
result = _call_tool(session, "read_file", path="assets/gold/large.txt")
|
||
assert result.startswith("[内容已截断]\n")
|
||
assert len(result) == 30000
|
||
|
||
|
||
def test_reference_session_snapshot_fields_are_read_only():
|
||
"""Session 保存的快照映射和受保护根索引不能被调用方改写。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-immutable",
|
||
reference_files={"assets/gold/a.txt": b"a"},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
|
||
with pytest.raises(TypeError):
|
||
session.reference_files["assets/gold/b.txt"] = b"b"
|
||
with pytest.raises(TypeError):
|
||
session.reference_roots["other"] = ("assets/other",)
|
||
with pytest.raises(AttributeError):
|
||
session.reference_files = {}
|
||
assert session.reference_roots["gold"] == ("assets/gold",)
|
||
|
||
|
||
def test_reference_snapshot_requires_roots_for_nonempty_files():
|
||
"""非空快照没有有效受保护根时，构造必须 fail-closed。"""
|
||
for roots in (None, {}, {"gold": ()}):
|
||
with pytest.raises((TypeError, ValueError)) as exc_info:
|
||
CheapSession(
|
||
game_id="snapshot-missing-roots",
|
||
reference_files={"assets/gold/guide.md": b"secret"},
|
||
reference_roots=roots,
|
||
)
|
||
assert "secret" not in str(exc_info.value)
|
||
assert "/" not in str(exc_info.value)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"root",
|
||
("", "../sensitive-root", "/absolute/sensitive-root"),
|
||
)
|
||
def test_reference_roots_reject_invalid_paths_without_leaking_input(root):
|
||
"""受保护根含空、越界或绝对路径时，构造不得静默过滤或泄露输入。"""
|
||
with pytest.raises((TypeError, ValueError)) as exc_info:
|
||
CheapSession(
|
||
game_id="snapshot-invalid-root",
|
||
reference_roots={"gold": (root,)},
|
||
)
|
||
|
||
message = str(exc_info.value)
|
||
assert "sensitive-root" not in message
|
||
assert "/absolute" not in message
|
||
|
||
|
||
def test_reference_snapshot_rejects_files_outside_all_roots():
|
||
"""任一快照文件未被受保护根覆盖时，构造必须拒绝整个索引。"""
|
||
with pytest.raises((TypeError, ValueError)) as exc_info:
|
||
CheapSession(
|
||
game_id="snapshot-uncovered-file",
|
||
reference_files={
|
||
"assets/gold/guide.md": b"allowed",
|
||
"assets/other/secret.md": b"secret",
|
||
},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
|
||
assert "secret" not in str(exc_info.value)
|
||
assert "assets/other" not in str(exc_info.value)
|
||
|
||
|
||
def test_empty_reference_files_with_roots_deny_read_and_list_without_io(monkeypatch):
|
||
"""有根无快照表示根内全部拒绝，read/list 均不得回落磁盘。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-empty-files",
|
||
reference_files={},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
read_calls = []
|
||
list_calls = []
|
||
|
||
def live_read(path):
|
||
read_calls.append(path)
|
||
return {"ok": True, "content": "disk-secret", "truncated": False}
|
||
|
||
def live_list(path):
|
||
list_calls.append(path)
|
||
return {"ok": True, "entries": ["disk-secret.txt"]}
|
||
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "read_file", live_read)
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "list_dir", live_list)
|
||
|
||
assert _call_tool(session, "read_file", path="assets/gold/unknown.txt").startswith("ERROR:")
|
||
assert _call_tool(session, "list_dir", path="assets/gold") == ""
|
||
assert read_calls == []
|
||
assert list_calls == []
|
||
|
||
|
||
def test_reference_list_path_alias_stays_virtual_without_io(monkeypatch):
|
||
"""受保护根的规范化别名仍使用虚拟列表，不能触发活目录 I/O。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-list-alias",
|
||
reference_files={"assets/gold/guide.md": b"snapshot"},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
calls = []
|
||
|
||
def live_list(path):
|
||
calls.append(path)
|
||
return {"ok": True, "entries": ["disk-secret.txt"]}
|
||
|
||
monkeypatch.setattr(cheap_toolkit.cheap_run, "list_dir", live_list)
|
||
|
||
assert _call_tool(session, "list_dir", path="assets/gold/./nested/../") == "guide.md"
|
||
assert calls == []
|
||
|
||
|
||
def test_reference_virtual_list_keeps_toolkit_result_cap():
|
||
"""虚拟目录投影过大时仍限制在 30KB 工具结果上限内。"""
|
||
session = CheapSession(
|
||
game_id="snapshot-list-cap",
|
||
reference_files={
|
||
f"assets/gold/file-{index:05d}.txt": b""
|
||
for index in range(5000)
|
||
},
|
||
reference_roots={"gold": ("assets/gold",)},
|
||
)
|
||
|
||
result = _call_tool(session, "list_dir", path="assets/gold")
|
||
|
||
assert len(result.encode("utf-8")) == 30_000
|
||
|
||
|
||
def test_cheap_session_keeps_legacy_positional_arguments():
|
||
"""旧的四个 CheapSession 位置参数仍按原字段接收。"""
|
||
check = {"ok": True}
|
||
build = {"ok": True}
|
||
finished = {"summary": "legacy"}
|
||
|
||
session = CheapSession("legacy", check, build, finished)
|
||
|
||
assert session.game_id == "legacy"
|
||
assert session.last_check is check
|
||
assert session.last_build is build
|
||
assert session.finished is finished
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
|
||
_failed = 0
|
||
for _fn in _fns:
|
||
try:
|
||
_fn()
|
||
print(f" PASS {_fn.__name__}")
|
||
except Exception as e: # noqa: BLE001
|
||
_failed += 1
|
||
print(f" FAIL {_fn.__name__}: {type(e).__name__}: {e}")
|
||
print(f"\n{len(_fns) - _failed}/{len(_fns)} passed")
|
||
sys.exit(1 if _failed else 0)
|