11436f07cc
test_a11_m3_deterministic 与 test_a11_m4_regenerate 在 import 时无恢复地替换
cheap_run.game_dir(两文件同构),污染同会话后续测试模块:受害者按真 games/ 根校验
write 边界,game_dir 被改指临时目录后一写即判越界。改为 module 级 autouse fixture
(yield 后恢复原值),__main__ 独立运行路径同装(单进程无需恢复),dedup 登记表一并收拢。
证据:全套 12→7 failed、三种文件顺序结果恒定(7 failed / 322 passed);二分钉死纯
game_dir 污染 = 5 个,均已恢复 = test_run::test_scaffold + test_toolkit::{write_l1_rejected,
write_l3_ok,check_shape_drawbutton_ok,check_shape_define_fail}。
余 7 个非本污染、隔离下同红,系环境前置:worktree 无 esbuild →
test_run::{test_stage,test_play_real}、test_toolkit::test_check_and_build_clean_pass;
缺真实 e2e 产物 → test_studio_smoke×4(须先跑 cheap_studio 单局)。须集成段 mini-desktop
才具备,未纳入本单。同族但未致失败的 test_worker_service 模块级 dedup 泄漏按最小改动未动。
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
293 lines
16 KiB
Python
293 lines
16 KiB
Python
"""test_a11_m4_regenerate.py — A11 切片三 M4「模块重生成执行」(改玩法)单测(全注入桩、零真 LLM)。
|
|
|
|
守的不变量:
|
|
· execute_regenerate_modify:有界单文件重写 game-logic.js,复用 cheap_studio resume + 三层校验,
|
|
但写边界收窄到只许写 game-logic.js;断言②对非目标文件 hash 稳定性把关。
|
|
- 注入假 rewrite(改 game-logic.js)→ manifest.changed=true、非目标稳(untouchedStable=true)、走到九门 verdict;
|
|
- intent 空 → failed 不重写(不调重写器);
|
|
- 假 rewrite 篡改非目标 → untouchedStable=false(断言②真能抓误伤)+ status failed;
|
|
- 无 base 源 / 重写异常 → best-effort failed,不抛。
|
|
· worker 接线:modifyMode=regenerate-module → 走 execute_regenerate_modify(不再占位 failed);
|
|
manifest(单 dict)附进 trace、成功带新版 sourceProject + 富 run-summary(cost/attempts)。
|
|
|
|
跑:cheap-worker/.venv/bin/python cheap-worker/tests/test_a11_m4_regenerate.py
|
|
"""
|
|
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
|
|
import cheap_run # noqa: E402
|
|
import cheap_modify as M # noqa: E402
|
|
import worker_service as W # noqa: E402
|
|
import dedup # noqa: E402
|
|
import pytest # noqa: E402
|
|
|
|
# game_dir 与 dedup 登记表在本模块测试期间重定向到临时目录:materialize、worker 接线的落盘都进临时目录,
|
|
# 不碰真 games/(cheap_modify、worker_service 与本模块同引用一个 cheap_run 模块对象,改模块属性即对三者同时
|
|
# 生效);本测虽不走 create 路,dedup 登记表仍一并重定向以防万一。重定向收进 autouse fixture:只在本模块测试
|
|
# 期间生效、跑完即恢复原值,不再在 import 时无恢复地替换 cheap_run.game_dir 而污染同会话后续测试模块。
|
|
def _install_tmp_redirect():
|
|
"""把 cheap_run.game_dir、dedup.DEDUP_REGISTRY 指向新建临时目录,返回二者原值供恢复。"""
|
|
orig = (cheap_run.game_dir, dedup.DEDUP_REGISTRY)
|
|
tmp_games = Path(tempfile.mkdtemp(prefix="a11-m4-games-"))
|
|
cheap_run.game_dir = lambda gid: tmp_games / f"amgen-{gid}" # noqa: E731
|
|
dedup.DEDUP_REGISTRY = Path(tempfile.mkdtemp(prefix="a11m4-dedup-")) / "_dedup_registry.jsonl"
|
|
return orig
|
|
|
|
|
|
@pytest.fixture(autouse=True, scope="module")
|
|
def _tmp_redirect():
|
|
orig_game_dir, orig_registry = _install_tmp_redirect()
|
|
try:
|
|
yield
|
|
finally:
|
|
cheap_run.game_dir = orig_game_dir
|
|
dedup.DEDUP_REGISTRY = orig_registry
|
|
|
|
_GOOD_BUNDLE = "var __GameBundle=(function(){return{bootGameHost(){}}})();"
|
|
_BASE_LOGIC = "// base game-logic\nexport function createGame({ plugins, bundle, viewport }) { return {}; }\n"
|
|
|
|
|
|
def _setup_base(game_id: str):
|
|
"""预备一个「已能玩」的便宜档产物:写好 L1 根件(entry-bundle/index,让 scaffold 跳过)+ 返回 base 源 dict。
|
|
|
|
base 源 files 含目标 game-logic.js + 两个非目标(core 数值 / host-config L1);execute_regenerate_modify 会
|
|
materialize 这些到 game_dir。返回 (game_dir, source_project_dict)。
|
|
"""
|
|
gd = cheap_run.game_dir(game_id)
|
|
(gd / "src").mkdir(parents=True, exist_ok=True)
|
|
# 预置 entry-bundle.js 让 execute 的 scaffold-if-needed 跳过(单测不 shell-out node scaffold)。
|
|
(gd / "entry-bundle.js").write_text("// L1 entry-bundle\n", encoding="utf-8")
|
|
(gd / "index.html").write_text("<html><body>g</body></html>", encoding="utf-8")
|
|
sp = {"files": {
|
|
"src/game-logic.js": _BASE_LOGIC,
|
|
"src/core.js": "export const SPEED = 5;\n", # 非目标(数值)
|
|
"src/host-config.js": "// L1 host-config plumbing\n", # 非目标(L1)
|
|
}}
|
|
return gd, sp
|
|
|
|
|
|
# ───────────────────────── ① execute_regenerate_modify ─────────────────────────
|
|
|
|
def _fake_rewrite_ok(game_id, intent, **kw):
|
|
"""假重写器:只改 game-logic.js(模拟 agent 有界重写),返回九门桩 run-summary(verdict pass)。"""
|
|
gd = cheap_run.game_dir(game_id)
|
|
(gd / "src" / "game-logic.js").write_text(
|
|
f"// rewritten for: {intent}\nexport function createGame({{ plugins, bundle, viewport }}) {{ return {{}}; }}\n",
|
|
encoding="utf-8")
|
|
return {"verdict": {"pass": True, "failedGates": []},
|
|
"verdictFull": {"pass": True, "guards": {}},
|
|
"attempts": 3, "stage": "play", "costRmb": 0.5,
|
|
"models": {"code": "MiniMax-M3"}, "gameId": game_id}
|
|
|
|
|
|
def _fake_rewrite_tamper(game_id, intent, **kw):
|
|
"""假重写器(误伤):改了 game-logic.js,**还动了非目标 core.js**——断言②应抓到。"""
|
|
gd = cheap_run.game_dir(game_id)
|
|
(gd / "src" / "game-logic.js").write_text("// rewritten\nexport function createGame(o){ return {}; }\n", encoding="utf-8")
|
|
(gd / "src" / "core.js").write_text("export const SPEED = 999;\n", encoding="utf-8") # 误伤非目标
|
|
return {"verdict": {"pass": True, "failedGates": []}, "verdictFull": {"pass": True, "guards": {}},
|
|
"attempts": 2, "stage": "play", "gameId": game_id}
|
|
|
|
|
|
def test_regenerate_changed_and_nontarget_stable():
|
|
"""假 rewrite 改 game-logic → changed=true、非目标稳(untouchedStable=true)、走到九门 verdict、status succeeded。"""
|
|
gid = "m4ok"
|
|
_, sp = _setup_base(gid)
|
|
patch = {"target": {"kind": "behavior"}, "payload": {"intent": "把命中加分从固定改成连击递增"}}
|
|
out = M.execute_regenerate_modify(gid, sp, patch, rewrite_fn=_fake_rewrite_ok)
|
|
assert out["status"] == "succeeded", out # 九门桩 pass ∧ 非目标稳 → succeeded
|
|
man = out["manifest"]
|
|
assert man["file"] == "src/game-logic.js" and man["kind"] == "behavior"
|
|
assert man["intent"] == "把命中加分从固定改成连击递增"
|
|
assert man["changed"] is True, "game-logic 被重写 → changed"
|
|
assert man["untouchedStable"] is True, "非目标(core/host-config/entry/index)未动 → stable"
|
|
assert out["verdict"] == {"pass": True, "guards": {}}, "九门 verdict 透传(走到 verdict)"
|
|
assert out["stage"] == "play"
|
|
assert out["summary"]["attempts"] == 3, "富 run-summary 透传(cost/attempts 在)"
|
|
# 非目标文件内容确实没变。
|
|
assert (cheap_run.game_dir(gid) / "src" / "core.js").read_text(encoding="utf-8") == "export const SPEED = 5;\n"
|
|
|
|
|
|
def test_regenerate_empty_intent_failed_no_rewrite():
|
|
"""intent 空(空串/缺 intent/缺 payload)→ failed 不重写(不调用重写器)。"""
|
|
gid = "m4empty"
|
|
_, sp = _setup_base(gid)
|
|
called = {}
|
|
|
|
def _spy(game_id, intent, **kw):
|
|
called["x"] = True
|
|
return {"verdict": {"pass": True}}
|
|
|
|
for patch in ({"target": {"kind": "behavior"}, "payload": {"intent": " "}},
|
|
{"target": {"kind": "behavior"}, "payload": {"intent": ""}},
|
|
{"target": {"kind": "behavior"}, "payload": {}},
|
|
{"target": {"kind": "behavior"}}):
|
|
out = M.execute_regenerate_modify(gid, sp, patch, rewrite_fn=_spy)
|
|
assert out["status"] == "failed", out
|
|
assert out["stage"] == "intent"
|
|
assert "intent" in (out["error"] or "")
|
|
assert out["manifest"]["untouchedStable"] is None, "没重写 → 稳定性未知=None"
|
|
assert "x" not in called, "intent 空绝不应调用重写器"
|
|
|
|
|
|
def test_regenerate_tamper_nontarget_caught_by_assert2():
|
|
"""假 rewrite 篡改非目标(core.js)→ untouchedStable=false(断言②真能抓误伤)+ 即便九门过也判 failed。"""
|
|
gid = "m4tamper"
|
|
_, sp = _setup_base(gid)
|
|
patch = {"target": {"kind": "behavior"}, "payload": {"intent": "改玩法但误伤了 core"}}
|
|
out = M.execute_regenerate_modify(gid, sp, patch, rewrite_fn=_fake_rewrite_tamper)
|
|
assert out["manifest"]["changed"] is True, "game-logic 仍被改"
|
|
assert out["manifest"]["untouchedStable"] is False, "断言②应抓到非目标 core.js 被改"
|
|
assert out["status"] == "failed", "误伤非目标=边界泄漏,即便九门桩 pass 也判失败"
|
|
assert "断言②" in (out["error"] or "") or "非目标" in (out["error"] or "")
|
|
|
|
|
|
def test_regenerate_no_source_failed():
|
|
"""无 base 源 files → failed(无可改源),不重写。"""
|
|
patch = {"target": {"kind": "behavior"}, "payload": {"intent": "随便改"}}
|
|
out = M.execute_regenerate_modify("m4nosrc", {"files": {}}, patch, rewrite_fn=_fake_rewrite_ok)
|
|
assert out["status"] == "failed" and "源" in (out["error"] or "")
|
|
out2 = M.execute_regenerate_modify("m4nosrc2", "{bad json", patch, rewrite_fn=_fake_rewrite_ok)
|
|
assert out2["status"] == "failed", "坏 source_project 容错 failed,不抛"
|
|
|
|
|
|
def test_regenerate_rewrite_exception_best_effort():
|
|
"""重写器抛异常 → best-effort failed + 诊断,绝不抛崩。"""
|
|
gid = "m4boom"
|
|
_, sp = _setup_base(gid)
|
|
|
|
def _boom(game_id, intent, **kw):
|
|
raise RuntimeError("LLM exploded")
|
|
|
|
out = M.execute_regenerate_modify(gid, sp, {"target": {"kind": "behavior"}, "payload": {"intent": "x"}},
|
|
rewrite_fn=_boom)
|
|
assert out["status"] == "failed" and "异常" in (out["error"] or "")
|
|
assert out["stage"] == "code"
|
|
|
|
|
|
def test_regenerate_gates_fail_is_plumbing_pass():
|
|
"""九门桩不过(verdict pass=false)但非目标稳 → status failed,但 changed/stable 正确(M4 验 plumbing,质量归 agent)。"""
|
|
gid = "m4gatefail"
|
|
_, sp = _setup_base(gid)
|
|
|
|
def _rewrite_gatefail(game_id, intent, **kw):
|
|
gd = cheap_run.game_dir(game_id)
|
|
(gd / "src" / "game-logic.js").write_text("// rewritten low quality\n", encoding="utf-8")
|
|
return {"verdict": {"pass": False, "failedGates": ["H_progress"]},
|
|
"verdictFull": {"pass": False, "guards": {"H_progress": {"pass": False}}},
|
|
"attempts": 6, "stage": "play", "gameId": game_id}
|
|
|
|
out = M.execute_regenerate_modify(gid, sp, {"target": {"kind": "behavior"}, "payload": {"intent": "改玩法"}},
|
|
rewrite_fn=_rewrite_gatefail)
|
|
assert out["status"] == "failed" and out["error"] == "九门未通过"
|
|
assert out["manifest"]["changed"] is True and out["manifest"]["untouchedStable"] is True, "plumbing 仍通(改了+非目标稳)"
|
|
|
|
|
|
# ───────────────────────── ② worker 接线 ─────────────────────────
|
|
|
|
def test_worker_regenerate_route_success_carries_manifest_bundle_source():
|
|
"""regenerate-module 成功 → result-out 带 engineBundle + manifest(单 dict)进 trace + 新版 sourceProject + 富 trace。"""
|
|
gid = "m4wok"
|
|
gd = cheap_run.game_dir(gid)
|
|
(gd / "src").mkdir(parents=True, exist_ok=True)
|
|
(gd / "bundle.iife.js").write_text(_GOOD_BUNDLE, encoding="utf-8") # 成功路需 engineBundle
|
|
(gd / "src" / "game-logic.js").write_text("// rewritten\n", encoding="utf-8")
|
|
captured = {}
|
|
fake = {"status": "succeeded", "stage": "play",
|
|
"manifest": {"file": "src/game-logic.js", "kind": "behavior", "intent": "连击递增",
|
|
"changed": True, "untouchedStable": True},
|
|
"verdict": {"pass": True, "guards": {}},
|
|
"summary": {"verdict": {"pass": True, "failedGates": []}, "verdictFull": {"pass": True, "guards": {}},
|
|
"attempts": 4, "stage": "play", "costRmb": 0.7, "models": {"code": "MiniMax-M3"}, "gameId": gid},
|
|
"error": None}
|
|
profile = {"tickModel": "realtime", "inputModel": "discrete-choice", "progressModel": "metric"}
|
|
state = W.WorkerState(regen_fn=lambda *a, **kw: fake,
|
|
send_fn=lambda u, p, s: captured.update(payload=p) or (200, ""),
|
|
profile_fn=lambda j, g: profile)
|
|
job = {"job_id": "rmok", "traceId": "rmok", "modifyMode": "regenerate-module", "gameId": gid,
|
|
"sourceProject": "{}", "modifyPatch": "{}", "brief": "连击小游戏", "callback": {"target": "http://cb"}}
|
|
W.process_job(state, job)
|
|
p = captured["payload"]
|
|
assert p["status"] == "succeeded"
|
|
assert "__GameBundle" in p["engineBundle"], "成功路携重建后的 bundle"
|
|
man = p["trace"]["modifyManifest"]
|
|
assert man["untouchedStable"] is True and man["changed"] is True, "manifest(单 dict)进 trace"
|
|
assert man["kind"] == "behavior"
|
|
assert "sourceProject" in p, "成功路 best-effort 带新版源(供链式改)"
|
|
assert p["trace"]["attempts"] == 4, "富 run-summary → trace.attempts"
|
|
assert p["trace"]["cost"]["totalRmb"] == 0.7, "富 run-summary → trace.cost"
|
|
|
|
|
|
def test_worker_regenerate_route_failed_attaches_error_and_manifest():
|
|
"""regenerate-module 失败(误伤非目标)→ failed 回调 + manifest/error 进 trace(不静默吞)。"""
|
|
gid = "m4wfail"
|
|
captured = {}
|
|
fake = {"status": "failed", "stage": "play",
|
|
"manifest": {"file": "src/game-logic.js", "kind": "behavior", "intent": "x",
|
|
"changed": True, "untouchedStable": False},
|
|
"verdict": {"pass": True, "guards": {}}, "summary": None,
|
|
"error": "断言②失败:模块重生成误伤了非目标文件"}
|
|
state = W.WorkerState(regen_fn=lambda *a, **kw: fake,
|
|
send_fn=lambda u, p, s: captured.update(payload=p) or (200, ""),
|
|
profile_fn=lambda j, g: None)
|
|
job = {"job_id": "rmf", "traceId": "rmf", "modifyMode": "regenerate-module", "gameId": gid,
|
|
"sourceProject": "{}", "modifyPatch": "{}", "brief": "x", "callback": {"target": "http://cb"}}
|
|
out = W.process_job(state, job)
|
|
assert out["status"] == "failed", "无 bundle + 执行器 failed → failed"
|
|
assert captured["payload"]["trace"]["modifyManifest"]["untouchedStable"] is False
|
|
assert captured["payload"]["trace"]["modifyError"], "失败原因附进 trace"
|
|
assert "sourceProject" not in captured["payload"], "失败路不带新版源"
|
|
|
|
|
|
def _boom_regen(*a, **kw):
|
|
raise AssertionError("deterministic/create 路不应调用 regen_fn")
|
|
|
|
|
|
def test_worker_deterministic_unaffected_by_regen_wiring():
|
|
"""modifyMode=deterministic 仍走 modify_fn、不触 regen_fn(regen 接线对 deterministic 零副作用)。"""
|
|
gid = "m4det"
|
|
gd = cheap_run.game_dir(gid)
|
|
(gd / "src").mkdir(parents=True, exist_ok=True)
|
|
(gd / "bundle.iife.js").write_text(_GOOD_BUNDLE, encoding="utf-8")
|
|
captured = {}
|
|
det = {"status": "succeeded",
|
|
"manifest": [{"file": "src/core.js", "kind": "config", "key": "SPEED", "old": "5", "new": "8", "found": True}],
|
|
"verdict": {"pass": True, "guards": {}}, "error": None, "stage": "play"}
|
|
state = W.WorkerState(modify_fn=lambda *a, **kw: det, regen_fn=_boom_regen,
|
|
send_fn=lambda u, p, s: captured.update(payload=p) or (200, ""),
|
|
profile_fn=lambda j, g: None)
|
|
job = {"job_id": "det1", "traceId": "det1", "modifyMode": "deterministic", "gameId": gid,
|
|
"sourceProject": "{}", "modifyPatch": "{}", "brief": "x", "callback": {"target": "http://cb"}}
|
|
W.process_job(state, job)
|
|
assert captured["payload"]["status"] == "succeeded"
|
|
assert isinstance(captured["payload"]["trace"]["modifyManifest"], list), "deterministic manifest 仍是 list"
|
|
|
|
|
|
def test_worker_unknown_mode_still_failed():
|
|
"""未知 modifyMode → 显式 failed 兜底回调(不静默吞;regen 接线后未知 mode 行为不变)。"""
|
|
captured = {}
|
|
state = W.WorkerState(send_fn=lambda u, p, s: captured.update(payload=p) or (200, ""))
|
|
job = {"job_id": "unk1", "traceId": "unk1", "modifyMode": "teleport", "gameId": "g",
|
|
"brief": "x", "callback": {"target": "http://cb"}}
|
|
out = W.process_job(state, job)
|
|
assert out["status"] == "failed" and out["mode"] == "teleport"
|
|
assert captured["payload"]["status"] == "failed"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_install_tmp_redirect() # 独立运行(非 pytest,autouse fixture 不触发):单进程一次性装重定向,退出即释放、无需恢复
|
|
_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[test_a11_m4_regenerate] {len(_fns) - _failed}/{len(_fns)} passed")
|
|
sys.exit(1 if _failed else 0)
|