agentmc/tests/test_render_bridge.py

"""渲染桥联测：Python ↔ Java 渲染器。

需要：
  - 已构建的 client-named.jar + McfoldRenderServer.class
  - 环境变量 MCFOLD_RENDER_CLASSPATH、MCFOLD_NATIVES_DIR
  - Xvfb DISPLAY
缺任一项则跳过（不误报失败）。
"""
import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from mcfold.render import (
    SECTION_VOLUME, _encode_paletted, _varint, parse_save_chunk,
)
from mcfold.render_bridge import RenderBridge, RenderBridgeError

SAVE = Path("/home/mc-test/mc-test")
STORE = Path("/home/newqqagent/mcview/store/vanilla-1.20.1")


def _bridge_available() -> tuple[bool, str]:
    cp = os.environ.get("MCFOLD_RENDER_CLASSPATH")
    if not cp:
        return False, "未设 MCFOLD_RENDER_CLASSPATH"
    paths = cp.split(os.pathsep)
    if not any("client-named" in p for p in paths):
        return False, "classpath 缺 client-named.jar"
    if not any("McfoldRenderServer" in p or "classes" in p for p in paths):
        return False, "classpath 缺 McfoldRenderServer.class"
    if not os.environ.get("DISPLAY"):
        return False, "无 DISPLAY"
    return True, ""


def test_bridge_protocol():
    """桥的协议层：registry / lookup / resolve —— 不涉及渲染。"""
    ok, why = _bridge_available()
    if not ok:
        print(f"  - 跳过（{why}）")
        return
    br = RenderBridge(
        game_dir=str(STORE / "instances" / "default"),
        assets=str(STORE / "assets"), asset_index="1.20.1",
    )
    try:
        br.start()
    except RenderBridgeError as e:
        print(f"  - 跳过（启动失败：{str(e)[:80]}）")
        return
    try:
        reg = br.registry_info()
        assert reg["ok"] and reg["blockCount"] > 500, reg
        assert reg["stateCount"] > 1000, reg

        # state id → 方块名（含属性）
        st = br.lookup_state(1)
        assert st["block"] == "minecraft:stone", st
        assert st["inRange"] is True

        # 方块名(+属性) → state id
        r = br._request({"cmd": "resolve", "blocks": [
            "minecraft:stone",
            "minecraft:oak_log[axis=y]",
            "minecraft:grass_block[snowy=false]",
            "minecraft:this_block_does_not_exist",
        ]})
        ids = r["ids"]
        assert ids[0] > 0, "stone 应有有效 id"
        assert ids[1] > 0, "oak_log[axis=y] 应可解析"
        assert ids[2] > 0, "grass_block 应可解析"
        assert ids[3] == -1, "不存在的方块应返回 -1"
        assert r["unknown"] == 1
        print(f"  ✓ 桥协议（blocks={reg['blockCount']}, states={reg['stateCount']}, "
              f"resolve 3 中 / 1 未知）")
    finally:
        br.stop()


def test_bridge_render_synthetic():
    """合成数据渲染：验证「注入 → 网格重建 → 出图」全链路。"""
    ok, why = _bridge_available()
    if not ok:
        print(f"  - 跳过（{why}）")
        return
    br = RenderBridge(
        game_dir=str(STORE / "instances" / "default"),
        assets=str(STORE / "assets"), asset_index="1.20.1",
    )
    try:
        br.start()
    except RenderBridgeError as e:
        print(f"  - 跳过（启动失败：{str(e)[:80]}）")
        return
    try:
        # 先设视野中心（否则区块被丢弃）
        br._request({"cmd": "viewport", "cx": 0, "cz": 0, "radius": 12})

        # 造：地板 stone（索引 4 = y=0 = 世界 y=64）+ 中间一根原木
        stone = br._request({"cmd": "resolve", "blocks": ["minecraft:stone"]})["ids"][0]
        log = br._request({"cmd": "resolve", "blocks": ["minecraft:oak_log[axis=y]"]})["ids"][0]
        assert stone > 0 and log > 0

        payload = bytearray()
        for sy in range(24):
            ids = [0] * SECTION_VOLUME
            if sy == 4:
                ids = [stone] * SECTION_VOLUME          # 地板（世界 y=64）
            elif sy in (5, 6, 7):
                for x in range(7, 9):                   # 中间 2x2 柱子
                    for z in range(7, 9):
                        ids[x + z * 16] = log
            non_empty = sum(1 for v in ids if v != 0)
            payload += (non_empty & 0xFFFF).to_bytes(2, "big")
            payload += _encode_paletted(ids, SECTION_VOLUME)
            payload += bytes([0]) + _varint(1) + _varint(0)

        items = [(cx, cz, bytes(payload)) for cx in range(-3, 4) for cz in range(-3, 4)]
        fed = br._request({"cmd": "chunks", "items": [
            {"cx": cx, "cz": cz, "data": __import__("base64").b64encode(d).decode()}
            for cx, cz, d in items
        ]}, timeout=180)
        assert fed["failed"] == 0, f"注入失败: {fed.get('firstError')}"
        assert fed["loadedChunks"] > 0, f"仓库为空: {fed}"

        br.set_camera(8.0, 72.0, 8.0, yaw=45.0, pitch=25.0)
        res = br.render(frames=80)
        assert res.width > 0 and res.height > 0
        assert res.chunks_rendered > 0, f"没有区块被渲染: {res.raw}"
        assert res.distinct_colors > 400, f"画面过单调: {res.distinct_colors}"
        assert res.has_world_content is True
        print(f"  ✓ 合成数据渲染（chunks={res.chunks_rendered}, "
              f"colors={res.distinct_colors}, {len(res.image)}B PNG）")
    finally:
        br.stop()


def test_bridge_render_real_save():
    """真实存档渲染：最终验收判据。"""
    ok, why = _bridge_available()
    if not ok:
        print(f"  - 跳过（{why}）")
        return
    if not (SAVE / "level.dat").is_file():
        print("  - 跳过（无测试存档）")
        return
    from mcfold.render import WorldRenderer
    wr = WorldRenderer.open(store=str(STORE), java="java")
    try:
        info = wr.load_save(SAVE, center=(8, -7), radius=4)
        assert info["injected"] > 0, info
        assert info["unknownBlocks"] == 0, f"有未解析方块: {info}"

        wr.camera(x=130.0, y=95.0, z=-110.0, yaw=135.0, pitch=35.0)
        dbg = wr.debug()
        assert dbg["loadedChunks"] > 0, f"仓库为空: {dbg}"
        assert dbg["playerChunk"] == "8,-7", f"相机区块不对: {dbg}"

        res = wr.screenshot(frames=100)
        assert res.chunks_rendered > 0, res.raw
        assert res.distinct_colors > 2000, f"真实地形应色彩丰富: {res.distinct_colors}"
        assert res.has_world_content is True
        print(f"  ✓ 真实存档渲染（injected={info['injected']}, "
              f"specs={info['uniqueSpecs']}, chunks={res.chunks_rendered}, "
              f"colors={res.distinct_colors}）")
    finally:
        wr.close()


if __name__ == "__main__":
    tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
    failed = 0
    for t in tests:
        try:
            t()
        except Exception as e:
            failed += 1
            import traceback
            print(f"  ✗ {t.__name__}: {e}")
            traceback.print_exc()
    print(f"\n渲染桥联测: " + ("全部通过" if failed == 0 else f"{len(tests)-failed}/{len(tests)} 通过"))
    sys.exit(1 if failed else 0)