games-development-ai/wg1/gen-worker/wg1_groupb_test.py

# -*- coding: utf-8 -*-
"""wg1_groupb_test.py —— W-G1 组B worker 侧自动化用例（9d trace 抽取完整率 + D9 反同质化）。

【为什么 __main__ 可跑而非 pytest（对齐 m3_player_test.py 范式）】worker venv 在 mini-desktop；本测试不依赖
  agentscope/openai（service.py 顶部会拉 openai，6c6g 无），故用 AST 隔离加载 service._extract_trace
  （函数体零外部依赖，纯 dict 操作），dedup.py 只用 stdlib——本测试在 6c6g 裸 python3 即可跑。

【覆盖执行版 §8 worker 侧 4 条】
  ① 9d trace 完整率：抽现成 run_studio 样本 → 必填子集七项完整率 100%（按必填子集口径）。
  ② D9 完全相同 config 撞重：同 config×2 → 第二次命中 → dupHit=true（仍照常，不阻断由 studio 侧保证）。
  ③ D9 归一撞重：title 空白差异 → norm_text 归一后相等 → 撞重判定一致。
  ④ D9 不同 config 不撞：异 config → sig 不同 → dupHit=false。

运行：python3 wg1/gen-worker/wg1_groupb_test.py（退出码 0=全过，非 0=有失败）。
"""

import ast
import glob
import json
import os
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent  # wg1/gen-worker
RESULTS = ROOT / "results"

# 必填子集（执行版 §5.3）：run_studio 当前路径无条件产出 = 完整率分母（camelCase 抽取后键名）
REQUIRED_SUBSET = ["pass", "repairs", "wallS", "models", "attempts", "gameId", "stage"]

_passed = 0
_failed = 0


def _check(name, cond, detail=""):
    """轻量断言：记录通过/失败并打印（不中断，跑完全部用例再汇总退出码）。"""
    global _passed, _failed
    if cond:
        _passed += 1
        print(f"  [PASS] {name}  {detail}")
    else:
        _failed += 1
        print(f"  [FAIL] {name}  {detail}")


def _load_extract_trace():
    """AST 隔离加载 service._extract_trace（绕开 service.py 顶部 openai/agentscope import）。"""
    src = (ROOT / "worker" / "service.py").read_text(encoding="utf-8")
    tree = ast.parse(src)
    for node in tree.body:
        if isinstance(node, ast.FunctionDef) and node.name == "_extract_trace":
            ns = {}
            exec(ast.get_source_segment(src, node), ns)  # 函数体零外部依赖
            return ns["_extract_trace"]
    raise RuntimeError("service._extract_trace 未找到")


def test_trace_completeness(extract):
    """① 9d trace 完整率：抽现成 run_studio 样本 → 必填子集七项 100% 非空（执行版 §8）。"""
    print("\n=== 用例①：9d trace 抽取完整率（必填子集口径）===")
    # run_studio 产物形态样本 = *.stage*.json（result 字面量恒含必填子集；裸 *.json 是旧 run.py 形态，路径未定不纳入）
    samples = sorted(glob.glob(str(RESULTS / "*.stage*.json")))
    _check("run_studio 形态样本存在", len(samples) >= 1, f"样本数={len(samples)}")
    ok = 0
    for f in samples:
        d = json.load(open(f, encoding="utf-8"))
        t = extract(d)
        miss = [k for k in REQUIRED_SUBSET if t.get(k) is None]
        if not (isinstance(t.get("attempts"), list) and len(t["attempts"]) >= 1):
            miss.append("attempts(empty)")
        # 每条 attempt 须带 guards 逐门（verdict.guards 真身）
        a0 = t["attempts"][0] if t.get("attempts") else {}
        if not isinstance(a0.get("guards"), dict):
            miss.append("attempts[0].guards")
        if not miss:
            ok += 1
        else:
            print(f"    {os.path.basename(f)} 缺={miss}")
    rate = (100 * ok // len(samples)) if samples else 0
    _check("必填子集完整率 100%", samples and ok == len(samples), f"{ok}/{len(samples)} = {rate}%")
    # camelCase 化抽样校验（snake→camel：wall_s→wallS / game_id→gameId / stage_fail→stageFail）
    if samples:
        t = extract(json.load(open(samples[0], encoding="utf-8")))
        _check("camelCase 化（wallS/gameId）", "wallS" in t and "gameId" in t)
        _check("attempts.guards 逐门保留",
               isinstance(t["attempts"][0].get("guards"), dict))
    # 空/None 兜底
    _check("空 result 返回 None", extract(None) is None and extract({}) is None)


def test_dedup():
    """②③④ D9 反同质化：完全相同撞重 / 归一撞重 / 不同不撞（执行版 §8）。"""
    print("\n=== 用例②③④：D9 反同质化告警 ===")
    sys.path.insert(0, str(ROOT / "worker"))
    from agent_loop import dedup  # 只用 stdlib，无 agentscope

    # 隔离登记表（不污染真 results/_dedup_registry.jsonl）
    tmp = tempfile.mkdtemp()
    dedup.DEDUP_REGISTRY = Path(tmp) / "_dedup_registry.jsonl"

    # 用例④（先建基线）：不同 config 不撞 + sig 不同
    s1 = dedup.check_similarity("躲避陨石的游戏", "generic", "trace-1", "stage1")
    s2 = dedup.check_similarity("消除方块的游戏", "generic", "trace-2", "stage1")
    _check("D9 首款不撞", s1["dupHit"] is False, f"dupHit={s1['dupHit']}")
    _check("D9 不同 config 不撞", s2["dupHit"] is False)
    _check("D9 不同 config sig 不同", s1["sig"] != s2["sig"])

    # 用例②：完全相同 config 撞重 → dupHit=true + dupWith 指向首款
    s3 = dedup.check_similarity("躲避陨石的游戏", "generic", "trace-3", "stage1")
    _check("D9 完全相同 config 撞重", s3["dupHit"] is True, f"dupWith={s3['dupWith']}")
    _check("D9 dupWith 指向首款 trace-1", s3["dupWith"] == "trace-1")

    # 用例③：归一撞重（title 尾空白差异，norm_text 归一后相等）
    s4 = dedup.check_similarity("躲避陨石的游戏 ", "generic", "trace-4", "stage1")
    _check("D9 归一撞重（空白差异判定一致）", s4["dupHit"] is True)
    _check("D9 norm_text 去空白", dedup.norm_text("点点乐 ") == dedup.norm_text("点点乐"))
    _check("D9 norm_text 去英文标点+casefold",
           dedup.norm_text("Hello, World!") == dedup.norm_text("helloworld"))

    # 自身重投不撞自己（同 traceId 排除）
    s5 = dedup.check_similarity("消除方块的游戏", "generic", "trace-2", "stage1")
    _check("D9 同 traceId 重投不撞自己", s5["dupHit"] is False)

    # 登记表坏行容错（降级不抛 → dupHit=False/None，不阻断）
    dedup.DEDUP_REGISTRY.write_text("not-json\n{bad\n", encoding="utf-8")
    s6 = dedup.check_similarity("某游戏", "generic", "trace-9", "stage1")
    _check("D9 登记表坏行容错（降级不阻断）", s6["dupHit"] in (False, None), f"dupHit={s6['dupHit']}")

    # compute_sig 可复现 + 64 位 hex
    _check("compute_sig 可复现", dedup.compute_sig("a", "b", "c") == dedup.compute_sig("a", "b", "c"))
    _check("compute_sig 64 位 hex", len(dedup.compute_sig("a", "b", "c")) == 64)


def main():
    print("W-G1 组B worker 侧自动化用例（9d trace 抽取 + D9 反同质化）")
    extract = _load_extract_trace()
    test_trace_completeness(extract)
    test_dedup()
    print(f"\n=== 汇总：通过 {_passed} / 失败 {_failed} ===")
    sys.exit(0 if _failed == 0 else 1)


if __name__ == "__main__":
    main()