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

"""
test_compare_multi.py — U2 三品类对照 harness 单测(mock verdict + 注入路径,不依赖真跑)。

覆盖:判据两层 + 逐品类绝对地板 + 双低标红(judge_genre)/ 金标 spec 注入真路径(staged 非 evidence)+ hash 一致 /
n 上限夹 5 / gameId 前缀隔离 / 报告序号不覆盖。真跑实跑在 U3(门控)。

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

import hashlib
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # → cheap-worker/
import compare_node as C
import cheap_run


def _run(gates: dict, real: bool = True) -> dict:
    return {"gates": gates, "realSrc": real}


_ALL = {"A_boot": True, "H_progress": True, "E_live": True}     # 真玩、全过
_HALF = {"A_boot": True, "H_progress": False, "E_live": False}  # 没真玩(H 不过)


# ── judge_genre:两层 + 绝对地板 + 双低标红 ──

def test_judge_equivalent():
    """Python 每局真 src/ + 过门率不低于 Node + 真玩比例≥0.6 → 坐实等价。"""
    j = C.judge_genre([_run(_ALL)] * 3, [_run(_ALL)] * 3)
    assert j["status"] == "equivalent", j
    assert j["pyPlayRatio"] == 1.0 and j["doubleLowRed"] is False


def test_judge_double_low_red():
    """两路真玩比例都<0.5 → inconclusive + 标红(防两路同卡被判等价)。"""
    j = C.judge_genre([_run(_HALF)] * 3, [_run(_HALF)] * 3)
    assert j["status"] == "inconclusive" and j["doubleLowRed"] is True, j


def test_judge_node_fails_python_plays():
    """Node 玩不动(H 全不过)、Python 真玩(过门率更高+达地板)→ 绝对层:Python 真过即坐实。"""
    j = C.judge_genre([_run(_ALL)] * 3, [_run(_HALF)] * 3)
    assert j["status"] == "equivalent", j
    assert j["doubleLowRed"] is False  # Node 低但 Python 高 → 非双低


def test_judge_absolute_floor_blocks():
    """Python 过门率不低于 Node,但真玩比例<0.6 地板(且 Node 真玩→非双低)→ regressed,不蒙混坐实。"""
    py = [_run(_ALL), _run(_HALF), _run(_HALF)]   # py_play=1/3≈0.33 < 0.6
    node = [_run(_ALL)] * 3                          # node_play=1.0(非双低)
    j = C.judge_genre(py, node)
    assert j["status"] == "regressed" and "地板" in j["reason"], j


def test_judge_regressed_rate():
    """Python 过门率低于 Node → regressed。"""
    j = C.judge_genre([_run(_HALF)] * 3, [_run(_ALL)] * 3)
    assert j["status"] == "regressed", j
    assert j["rateDelta"] < 0


def test_judge_not_real_src():
    """Python 非每局真 src/ → 即使过门率好也 regressed(doc↔code 兑现)。"""
    j = C.judge_genre([_run(_ALL, real=False)] * 3, [_run(_ALL)] * 3)
    assert j["status"] == "regressed", j


# ── 金标 spec 注入:真路径(staged 非 evidence)+ hash 一致 ──

def test_inject_target_is_staged_not_evidence():
    """inject 目标 = _wg1-gen/<id>/play-spec.json(staged 真实读取处),不是 evidence 目录。"""
    dst = cheap_run.wg1_game_dir("X") / "play-spec.json"
    assert dst.parent.name == "X" and dst.parent.parent.name == "_wg1-gen"
    assert "evidence" not in dst.parts, "金标 spec 不能注入 evidence 目录(Codex C1/C4)"


def test_write_spec_atomic_hash_match():
    """原子写后内容与金标 fixture 逐字节一致(verdict 用的 spec 与 fixture hash 对得上)。"""
    golden = C.golden_spec_path(C._GENRE_BY_KEY["whack-mole"])
    content = golden.read_text(encoding="utf-8")
    with tempfile.TemporaryDirectory() as d:
        dst = Path(d) / "play-spec.json"
        C._write_spec_atomic(dst, content)
        h_src = hashlib.sha256(content.encode("utf-8")).hexdigest()
        h_dst = hashlib.sha256(dst.read_bytes()).hexdigest()
        assert h_src == h_dst, "注入后 spec 与金标 hash 不一致"


def test_golden_paths_resolve():
    """三品类金标 spec 都在 fixtures/golden-specs/。"""
    for g in C.GENRES:
        assert C.golden_spec_path(g).exists(), f"{g['key']} 金标 spec 缺失"


# ── 编排辅助:n 上限 / 前缀隔离 / 报告序号 ──

def test_clamp_n():
    assert C._clamp_n(7) == 5 and C._clamp_n(3) == 3 and C._clamp_n(0) == 1
    assert C._clamp_n("x") == 3  # 非法回落默认 3


def test_pair_ids_isolation():
    """品类×k 的 gameId 前缀两两不撞、cheap-/node- 分离。"""
    seen = set()
    for key in ("click-score", "whack-mole", "shop-serve"):
        for k in range(3):
            py, node = C._pair_ids(key, k)
            assert py.startswith("cheap-") and node.startswith("node-")
            assert py not in seen and node not in seen
            seen.update([py, node])
    assert len(seen) == 3 * 3 * 2


def test_port_pool_distinct_non_overlapping():
    """并发端口池:每槽一对,所有 server/cdp 端口两两不撞、server 段与 cdp 段不重叠。"""
    pool = C._port_pool(4)
    assert len(pool) == 4
    all_ports = [p for pair in pool for p in pair]
    assert len(set(all_ports)) == len(all_ports), "端口有重复(并发会撞)"
    server_ports = {pair[0] for pair in pool}
    cdp_ports = {pair[1] for pair in pool}
    assert not (server_ports & cdp_ports), "server 段与 cdp 段重叠"
    assert C._port_pool(1) == [(4320, 9222)]  # 串行=单槽默认端口
    assert C._port_pool(0) == [(4320, 9222)]  # conc<1 兜底至少一槽


def test_next_index_no_overwrite():
    assert C._next_index([]) == 1
    assert C._next_index(["compare-multi-1", "compare-multi-2"]) == 3
    assert C._next_index(["compare-multi-1", "foo", "compare-multi-5"]) == 6


def test_aggregate_buckets_by_status():
    """聚合按 status 分桶,坐实品类进退役就绪范围。"""
    genre_runs = {
        "click-score": [{"pyId": "p", "nodeId": "n", "py": _run(_ALL), "node": _run(_ALL)}] * 3,
        "whack-mole": [{"pyId": "p", "nodeId": "n", "py": _run(_HALF), "node": _run(_HALF)}] * 3,
    }
    rep = C.aggregate(genre_runs)
    assert rep["summary"]["equivalentGenres"] == ["click-score"]
    assert rep["summary"]["inconclusiveGenres"] == ["whack-mole"]
    assert rep["summary"]["retireReadyUnderGolden"] == ["click-score"]


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)