58 lines
3.5 KiB
Python
58 lines
3.5 KiB
Python
"""cheap_gates.run_cheap_gates 单测:同步跑九门收口的端口透传 + 逐段桩(monkeypatch cheap_run,零真 chrome/esbuild)。
|
|
跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_gates.py -v
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
|
|
import _bootstrap # noqa: E402,F401 仅加 sys.path(import 时不取 key、不触网)
|
|
import cheap_gates # noqa: E402
|
|
import cheap_run # noqa: E402
|
|
|
|
|
|
def test_run_cheap_gates_threads_ports_and_returns_verdict(monkeypatch):
|
|
seen = {}
|
|
monkeypatch.setattr(cheap_run, "stage", lambda gid: {"ok": True, "output": ""})
|
|
monkeypatch.setattr(cheap_run, "smoke",
|
|
lambda gid, port, cdp_port: seen.update(smoke=(port, cdp_port)) or {"ok": True, "state": {"targets": []}})
|
|
monkeypatch.setattr(cheap_run, "ensure_play_spec", lambda gid, state: {"ok": True, "wrote": True, "driverType": "tap-targets"})
|
|
monkeypatch.setattr(cheap_run, "play",
|
|
lambda gid, port, cdp_port: seen.update(play=(port, cdp_port)) or {"ok": True, "verdict": {"pass": True, "guards": {}}})
|
|
out = cheap_gates.run_cheap_gates("g1", 4407, 9407)
|
|
assert out == {"pass": True, "guards": {}}
|
|
assert seen["smoke"] == (4407, 9407), "smoke 应收到派生端口"
|
|
assert seen["play"] == (4407, 9407), "play 应收到派生端口"
|
|
|
|
|
|
def test_run_cheap_gates_stage_fail_returns_empty(monkeypatch):
|
|
monkeypatch.setattr(cheap_run, "stage", lambda gid: {"ok": False, "output": "esbuild bundle 缺失"})
|
|
called = {"smoke": 0}
|
|
monkeypatch.setattr(cheap_run, "smoke", lambda *a, **k: called.update(smoke=called["smoke"] + 1) or {"ok": True})
|
|
out = cheap_gates.run_cheap_gates("g1", 4407, 9407)
|
|
assert out == {}, "stage 失败应短路返 {}(judge 判未过)"
|
|
assert called["smoke"] == 0, "stage 失败不应再跑 smoke"
|
|
|
|
|
|
def test_run_cheap_gates_unlinks_stale_verdict_before_play(tmp_path, monkeypatch):
|
|
# 红线③(封陈旧 verdict 假绿,镜像 tier2 run.py:866):play 前必清残留 verdict —— 预铺上一局绿 verdict,
|
|
# 断言 play 被调用当刻它已不在盘上(证明 unlink 在 play 之前跑)。不清则 serve-and-play.sh 早退/超时不写新
|
|
# verdict 时,play 会读回上一局绿 verdict → judge 只看 verdict 不看 rc → 未过门新代码被假绿放行。
|
|
import json
|
|
monkeypatch.setattr(cheap_run, "wg1_game_dir", lambda gid: tmp_path / "_wg1-gen" / gid)
|
|
ev = tmp_path / "_wg1-gen" / "g1" / "evidence"
|
|
ev.mkdir(parents=True)
|
|
(ev / "verdict.json").write_text(json.dumps({"pass": True, "guards": {}}), encoding="utf-8") # 陈旧绿残留
|
|
monkeypatch.setattr(cheap_run, "stage", lambda gid: {"ok": True, "output": ""})
|
|
monkeypatch.setattr(cheap_run, "smoke", lambda gid, port, cdp_port: {"ok": True, "state": {"targets": []}})
|
|
monkeypatch.setattr(cheap_run, "ensure_play_spec", lambda gid, state: {"ok": True})
|
|
seen = {}
|
|
|
|
def _stub_play(gid, port, cdp_port):
|
|
# 记录 play 被调用当刻,陈旧 verdict 是否还在盘上(封红线③:必须已被 run_cheap_gates 在 play 前清掉)。
|
|
seen["verdict_exists_at_play"] = (ev / "verdict.json").exists()
|
|
return {"ok": True, "verdict": {"pass": False, "guards": {}}}
|
|
|
|
monkeypatch.setattr(cheap_run, "play", _stub_play)
|
|
cheap_gates.run_cheap_gates("g1", 4407, 9407)
|
|
assert seen["verdict_exists_at_play"] is False, "play 前必须已清残留 verdict(否则读回上一局绿=假绿放行)"
|