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

"""test_result_out.py — §6.1 result-out 三字段组装单测(M3a U1)。

守的不变量(KTD2 BLOCKER 修正,镜像 SaaGraphDispatcher.buildCallbackReqVO + service.py build_callback_payload):
  · gameConfig = 最小占位 Map(templateId/title/theme/engineDriven),绝不塞源工程;
  · engineBundle = bundle.iife.js 全文(含 __GameBundle)= 运行时承重产物;
  · 成功门 = verdict.pass ∧ engineBundle 含 __GameBundle,否则 failed + 七值 failureReason;
  · sourceProject = best-effort 可选(profile 派生不出即省略,不伪造、不阻断成功);
  · traceId 取 job.traceId 或 job_id;纯函数无网络无 LLM、同输入恒同输出。

跑:cheap-worker/.venv/bin/python cheap-worker/tests/test_result_out.py
"""

import sys
import tempfile
from pathlib import Path

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

_GOOD_BUNDLE = "var __GameBundle=(function(){return{bootGameHost(){}}})();"


def _tmp() -> Path:
    return Path(tempfile.mkdtemp(prefix="ro-test-"))


def _make_game_dir(tmp: Path, *, bundle=_GOOD_BUNDLE, src_files=None) -> Path:
    """造 amgen-<id>/ 形态产物目录(bundle.iife.js + src/)。"""
    gd = tmp / "amgen-t1"
    (gd / "src").mkdir(parents=True)
    if bundle is not None:
        (gd / "bundle.iife.js").write_text(bundle, encoding="utf-8")
    for name, content in (src_files or {"game-logic.js": "// logic\n", "core.js": "// core\n"}).items():
        (gd / "src" / name).write_text(content, encoding="utf-8")
    return gd


def test_succeeded_shape():
    """成功路:verdict.pass ∧ bundle 含 __GameBundle → status=succeeded + gameConfig 占位 + engineBundle 承重。"""
    gd = _make_game_dir(_tmp())
    job = {"job_id": "trace-abc", "traceId": "trace-abc", "templateId": "generic", "brief": "点点乐"}
    summary = {"ok": True, "finished": True, "verdict": {"pass": True, "failedGates": []}, "costRmb": 0.05}

    out = R.build_result_out(job, summary, gd)

    assert out["traceId"] == "trace-abc"
    assert out["status"] == "succeeded"
    assert out["templateId"] == "generic"
    # gameConfig 占位,绝不含源工程(BLOCKER 修正)
    assert out["gameConfig"]["templateId"] == "generic"
    assert out["gameConfig"]["theme"] == "generic"
    assert out["gameConfig"]["engineDriven"] is True
    assert "files" not in out["gameConfig"] and "entry" not in out["gameConfig"]
    # engineBundle 承重
    assert "__GameBundle" in out["engineBundle"]
    assert out.get("failureReason") is None


def test_failed_when_gates_not_pass():
    """verdict.pass=False → status=failed + 七值 failureReason,不带 engineBundle。"""
    gd = _make_game_dir(_tmp())
    job = {"traceId": "t2", "templateId": "generic", "brief": "x"}
    summary = {"finished": True, "verdict": {"pass": False, "failedGates": ["E_live"]}}

    out = R.build_result_out(job, summary, gd)

    assert out["status"] == "failed"
    assert out.get("engineBundle") is None
    assert out["failureReason"] in R._FAILURE_REASONS  # 对齐后端 FailureReasonEnum 真七值


def test_failed_when_engine_bundle_missing___gamebundle():
    """即使 verdict.pass,bundle 无 __GameBundle → status=failed(承重门:engineBundle 含 __GameBundle)。"""
    gd = _make_game_dir(_tmp(), bundle="var notTheGlobal=1;")
    job = {"traceId": "t3", "templateId": "generic", "brief": "x"}
    summary = {"finished": True, "verdict": {"pass": True, "failedGates": []}}

    out = R.build_result_out(job, summary, gd)

    assert out["status"] == "failed"
    assert out.get("engineBundle") is None


def test_failed_breaker_maps_llm_error():
    """预算熔断 / 内部真因不在七值内 → failureReason=llm_error(契约#6 catch-all;枚举无 budget 专项值)。"""
    gd = _make_game_dir(_tmp())
    job = {"traceId": "t4", "templateId": "generic", "brief": "x"}
    summary = {"finished": False, "breaker": True, "verdict": {"pass": None, "failedGates": None}}

    out = R.build_result_out(job, summary, gd)

    assert out["status"] == "failed"
    assert out["failureReason"] == "llm_error"


def test_explicit_failure_reason_wins():
    """显式 failure_reason(在七值内)优先于推断。"""
    gd = _make_game_dir(_tmp())
    job = {"traceId": "t5", "templateId": "generic", "brief": "x"}
    summary = {"finished": True, "verdict": {"pass": False, "failedGates": ["A_boot"]}}

    out = R.build_result_out(job, summary, gd, failure_reason="timeout")

    assert out["status"] == "failed"
    assert out["failureReason"] == "timeout"


def test_traceid_fallback_to_job_id():
    """traceId 缺 → 用 job_id(贯穿键同值)。"""
    gd = _make_game_dir(_tmp())
    job = {"job_id": "jid-9", "templateId": "generic", "brief": "x"}
    summary = {"verdict": {"pass": True}}

    out = R.build_result_out(job, summary, gd)

    assert out["traceId"] == "jid-9"


def test_cost_goes_into_trace_not_toplevel():
    """成本落 trace.cost.totalRmb——顶层不得有 costRmb(DifyCallbackReqVO 无此字段、后端严格拒未知)。"""
    gd = _make_game_dir(_tmp())
    job = {"traceId": "t6", "templateId": "generic", "brief": "x"}
    summary = {"verdict": {"pass": True}, "costRmb": 0.07}

    out = R.build_result_out(job, summary, gd)

    assert "costRmb" not in out, "顶层不得有 costRmb(会被后端 ObjectMapper 拒)"
    assert out["trace"]["cost"]["totalRmb"] == 0.07


def test_result_out_keys_within_vo_known_set():
    """result-out 顶层键严格 ⊆ DifyCallbackReqVO 已知字段集(后端拒未知字段的硬约束)。"""
    vo_known = {"traceId", "status", "templateId", "gameConfig", "assets",
                "engineBundle", "failureReason", "qualityScore", "sourceProject", "trace"}
    gd = _make_game_dir(_tmp())
    job = {"traceId": "t9", "templateId": "generic", "brief": "x"}
    out_ok = R.build_result_out(job, {"verdict": {"pass": True}, "costRmb": 0.07}, gd,
                               source_project='{"schemaVersion":"2.0"}')
    out_fail = R.build_result_out(job, {"verdict": {"pass": False}}, gd)
    assert set(out_ok) <= vo_known, f"成功路含未知字段:{set(out_ok) - vo_known}"
    assert set(out_fail) <= vo_known, f"失败路含未知字段:{set(out_fail) - vo_known}"


def test_build_source_project_valid_2_0():
    """profile 可派生 + src/ 有文件 → sourceProject 是合法 2.0(必填齐 + sourceHash 64-hex)。"""
    import json
    import re
    gd = _make_game_dir(_tmp(), src_files={"game-logic.js": "// l\n", "core.js": "// c\n"})
    profile = {"tickModel": "frame", "inputModel": "tap-targets", "progressModel": "score"}

    sp_str = R.build_source_project(gd, profile, entry="entry.js")

    assert sp_str is not None
    sp = json.loads(sp_str)
    assert sp["schemaVersion"] == "2.0"
    assert sp["globalName"] == "__GameBundle"
    assert sp["entry"] == "entry.js"
    assert sp["profile"] == profile
    assert "src/game-logic.js" in sp["files"] and "src/core.js" in sp["files"]
    assert re.fullmatch(r"[a-f0-9]{64}", sp["sourceHash"])


def test_build_source_project_omitted_when_profile_incomplete():
    """profile 派生不出(None / 三枚举不全)→ 返 None(省略、绝不伪造)。"""
    gd = _make_game_dir(_tmp())
    assert R.build_source_project(gd, None, entry="entry.js") is None
    assert R.build_source_project(gd, {"tickModel": "frame"}, entry="entry.js") is None


def test_build_source_project_omitted_when_no_src():
    """src/ 无文件 → 返 None。"""
    empty = _tmp() / "amgen-e"
    (empty / "src").mkdir(parents=True)
    profile = {"tickModel": "frame", "inputModel": "tap-targets", "progressModel": "score"}
    assert R.build_source_project(empty, profile, entry="entry.js") is None


def test_result_out_source_project_additive():
    """build_result_out 收到 source_project(JSON str)→ payload 带;缺省不带(additive)。"""
    gd = _make_game_dir(_tmp())
    job = {"traceId": "t7", "templateId": "generic", "brief": "x"}
    summary = {"verdict": {"pass": True}}

    out_with = R.build_result_out(job, summary, gd, source_project='{"schemaVersion":"2.0"}')
    assert out_with["sourceProject"] == '{"schemaVersion":"2.0"}'

    out_without = R.build_result_out(job, summary, gd)
    assert "sourceProject" not in out_without


def test_deterministic_zero_llm():
    """纯函数:同输入恒同输出(无网络无 LLM)。"""
    gd = _make_game_dir(_tmp())
    job = {"traceId": "t8", "templateId": "generic", "brief": "稳定"}
    summary = {"verdict": {"pass": True}, "costRmb": 0.01}
    assert R.build_result_out(job, summary, gd) == R.build_result_out(job, summary, gd)


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)