games-development-ai/cheap-worker/tests/test_trace_source.py

"""test_trace_source.py — M3b U1：build_trace_source 9d-trace 源维度富化单测。

守的不变量：
  · 纯函数 build_trace_source(verdict, driver_type, attempts, stage, model) 产 additive 4 键
    verdictFull/driverType/models/stage，供 result_out 组 payload["trace"]、后端 D11 算分；
  · verdictFull 保留完整 verdict（guards.H_progress 仍是嵌套对象，含 pass）；play 没跑 → None；
  · driverType 透传（没产 play-spec → None）；models = {code: model} map；
  · 不回写既有键 attempts（守「不动既有键」，repairs 由 result_out 据 summary.attempts 派生）；
  · _furthest_stage 据收口实际到达步判最远阶段。

不跑真 gen（纯函数，合成 verdict dict）。
跑：cheap-worker/.venv/bin/python cheap-worker/tests/test_trace_source.py
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # → cheap-worker/
import cheap_studio as S  # noqa: E402

# 合成完整九门 verdict（含 guards.H_progress 嵌套对象，对齐真 verdict.json 形态；带 url/ts 噪声供 U2 验去噪）。
_FULL_VERDICT = {
    "pass": True,
    "url": "http://localhost:4320/?cheap",
    "ts": 1700000000,
    "guards": {
        "A_boot": {"pass": True},
        "H_progress": {"pass": True, "checks": ["score+"], "latch": True, "state0": 0, "state1": 3},
    },
}


def test_build_trace_source_success_shape():
    """成功路：verdictFull 含完整 guards（H_progress 嵌套保 pass）、driverType 透传、models map、stage。"""
    out = S.build_trace_source(_FULL_VERDICT, "tap-targets", 1, "play", "MiniMax-M3")
    assert out["verdictFull"]["pass"] is True
    # H_progress 保持嵌套对象（不是裸 bool），后端 ReadinessScorer.guardPass 读其 .pass
    assert isinstance(out["verdictFull"]["guards"]["H_progress"], dict)
    assert out["verdictFull"]["guards"]["H_progress"]["pass"] is True
    assert out["driverType"] == "tap-targets"
    assert out["models"] == {"code": "MiniMax-M3"}
    assert out["stage"] == "play"


def test_build_trace_source_none_verdict():
    """play 没跑（verdict=None）→ verdictFull=None；driverType=None（不崩）。"""
    out = S.build_trace_source(None, None, 1, "smoke", "MiniMax-M3")
    assert out["verdictFull"] is None
    assert out["driverType"] is None
    assert out["models"] == {"code": "MiniMax-M3"}
    assert out["stage"] == "smoke"


def test_build_trace_source_does_not_emit_attempts():
    """不回写既有键 attempts（守「不动既有键」；repairs 由 result_out 据 summary.attempts 派生）。"""
    out = S.build_trace_source(_FULL_VERDICT, "tap-targets", 3, "play", "MiniMax-M3")
    assert "attempts" not in out


def test_build_trace_source_non_dict_verdict_to_none():
    """verdict 非 dict（异常形态）→ verdictFull=None（防御）。"""
    out = S.build_trace_source("not-a-dict", None, 1, "code", "MiniMax-M3")
    assert out["verdictFull"] is None


def test_furthest_stage_ordering():
    """_furthest_stage 据收口实际到达步取最远：play>smoke>stage>code>scaffold。"""
    assert S._furthest_stage(finished=True, staged=True, smoke_ran=True, played=True) == "play"
    assert S._furthest_stage(finished=True, staged=True, smoke_ran=True, played=False) == "smoke"
    assert S._furthest_stage(finished=True, staged=True, smoke_ran=False, played=False) == "stage"
    assert S._furthest_stage(finished=True, staged=False, smoke_ran=False, played=False) == "code"
    assert S._furthest_stage(finished=False, staged=False, smoke_ran=False, played=False) == "scaffold"


def test_closeout_gates_passes_optional_interaction_profile(monkeypatch, tmp_path):
    """完整 closeout 的 smoke 透传 profile；play 顺序和默认结果形状保持不变。"""
    received = []
    monkeypatch.setattr(S.cheap_run, "stage", lambda _gid: {"ok": True, "output": ""})
    monkeypatch.setattr(
        S.cheap_run, "smoke",
        lambda _gid, **kwargs: received.append(kwargs.get("interaction_profile_id"))
        or {"ok": True, "state": None, "raw": ""},
    )
    monkeypatch.setattr(S.cheap_run, "play", lambda _gid, **_kwargs: {"verdict": {"pass": True}})
    monkeypatch.setattr(S.cheap_run, "wg1_game_dir", lambda _gid: tmp_path)

    out = S._closeout_gates(
        "match3-closeout", port=4320, cdp_port=9222,
        interaction_profile_id="match3.orthogonal-swap-v1",
    )
    assert received == ["match3.orthogonal-swap-v1"]
    assert out["staged"] is True and out["smoke_ok"] is True


def test_closeout_gates_default_profile_is_none(monkeypatch, tmp_path):
    """未传 profile 时显式透传 None，兼容既有普通游戏 smoke。"""
    received = []
    monkeypatch.setattr(S.cheap_run, "stage", lambda _gid: {"ok": True, "output": ""})
    monkeypatch.setattr(
        S.cheap_run, "smoke",
        lambda _gid, **kwargs: received.append(kwargs.get("interaction_profile_id"))
        or {"ok": True, "state": None, "raw": ""},
    )
    monkeypatch.setattr(S.cheap_run, "play", lambda _gid, **_kwargs: {"verdict": {"pass": True}})
    monkeypatch.setattr(S.cheap_run, "wg1_game_dir", lambda _gid: tmp_path)

    S._closeout_gates("default-closeout", port=4320, cdp_port=9222)
    assert received == [None]


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)