"""渲染联测：Python ↔ Java 后端。 需要： - 已构建的 client-named.jar + McfoldRenderServer.class - 环境变量 MCFOLD_RENDER_CLASSPATH、MCFOLD_NATIVES_DIR - Xvfb DISPLAY 缺任一项则跳过（不误报失败）。 ★ 架构口径：Python 只发 RLE（set_blocks），协议格式由 Java 侧组装。 """ import os import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from mcfold.backends import Gen_1_20, backend_class from mcfold.backends.base import BackendError, rle_encode from mcfold.render import SECTION_VOLUME, WorldRenderer, parse_save_chunk SAVE = Path("/home/mc-test/mc-test") STORE = Path("/home/newqqagent/mcview/store/vanilla-1.20.1") def _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 缺编译产物" if not os.environ.get("DISPLAY"): return False, "无 DISPLAY" return True, "" def _open_backend(): ok, why = _available() if not ok: return None, why br = Gen_1_20( game_dir=str(STORE / "instances" / "default"), assets=str(STORE / "assets"), asset_index="1.20.1", ) try: br.start() except BackendError as e: return None, f"启动失败：{str(e)[:100]}" return br, "" # ---------------------------------------------------------------- 世代分派 def test_generation_dispatch(): """判据是 DeltaTracker 是否存在（不是硬编码版本号）。""" jars = { "1.20.1": "/tmp/client-1.20.1.jar", "1.21.1": "/tmp/genb/client-1.21.1-named.jar", } present = {v: p for v, p in jars.items() if Path(p).is_file()} if not present: print(" - 无 named jar，跳过") return for ver, jar in present.items(): cls = backend_class(jar) if ver == "1.20.1": assert cls is Gen_1_20, f"{ver} 应选 Gen_1_20，得到 {cls.__name__}" else: # 1.21+ 必须有 DeltaTracker ⇒ 选 gen_1_21 from mcfold.backends import Gen_1_21 assert cls is Gen_1_21, f"{ver} 应选 Gen_1_21，得到 {cls.__name__}" print(f" ✓ 世代分派 {ver} → {cls.__name__}") def test_generation_metadata(): """每个后端要能说清自己的 API 特征（加新世代时的对照依据）。""" notes = Gen_1_20.api_notes() assert "setLevel" in notes and "render" in notes assert notes.get("deltaTracker") is False from mcfold.backends import Gen_1_21 n21 = Gen_1_21.api_notes() assert n21.get("deltaTracker") is True, "1.21+ 用 DeltaTracker" assert "DeltaTracker" in n21["render"] print(f" ✓ 世代元信息（1.20: {notes['render']}）") # ---------------------------------------------------------------- 协议层 def test_capacity_from_runtime(): """★ 世界尺寸必须问后端，不是 Python 硬编码。""" br, why = _open_backend() if br is None: print(f" - 跳过（{why}）") return try: cap = br._request({"cmd": "capacity"}) assert cap["sectionsCount"] > 0, cap assert "minSection" in cap and "maxSection" in cap # 1.20.1 实测 24 / -4 / 20 assert cap["sectionsCount"] == 24, cap assert cap["minSection"] == -4, cap assert cap["maxSection"] == 20, cap print(f" ✓ 运行时世界尺寸（sections={cap['sectionsCount']}, " f"min={cap['minSection']}, biome={cap.get('defaultBiome')}）") finally: br.stop() def test_resolve_and_lookup(): """方块名↔state id 是渲染器侧权威映射（Python 不复制这张表）。""" br, why = _open_backend() if br is None: print(f" - 跳过（{why}）") return try: reg = br.registry_info() assert reg["blockCount"] > 500 and reg["stateCount"] > 1000 assert br.lookup_state(1)["block"] == "minecraft:stone" ids = br.resolve_blocks([ "minecraft:stone", "minecraft:oak_log[axis=y]", "minecraft:grass_block[snowy=false]", "minecraft:no_such_block", ]) assert ids[0] > 0 and ids[1] > 0 and ids[2] > 0 assert ids[3] == -1, "未知方块应返回 -1" print(f" ✓ 名字↔id（blocks={reg['blockCount']}, states={reg['stateCount']}）") finally: br.stop() # ---------------------------------------------------------------- 渲染（新路径） def test_set_blocks_render(): """新路径：Python 只发 RLE，Java 组装协议 + 官方序列化 + 入 cache。""" br, why = _open_backend() if br is None: print(f" - 跳过（{why}）") return try: cap = br._request({"cmd": "capacity"}) n_sec = cap["sectionsCount"] stone, grass, log = br.resolve_blocks([ "minecraft:stone", "minecraft:grass_block[snowy=false]", "minecraft:oak_log[axis=y]", ]) assert stone > 0 and grass > 0 and log > 0 br.set_viewport(0, 0, radius=12) sections = {} for idx in range(n_sec): v = [0] * SECTION_VOLUME if idx < 4: v = [stone] * SECTION_VOLUME # 地基 elif idx == 4: v = [grass] * SECTION_VOLUME # 地表 elif idx == 5: for x in range(7, 9): # 索引 = x + z*16 + y*256 for z in range(7, 9): v[x + z * 16] = log sections[idx] = v total_placed = 0 for cx in range(-3, 4): for cz in range(-3, 4): r = br.set_blocks(cx, cz, sections) assert r["ok"], r total_placed += r["placed"] assert total_placed > 0, "应写入方块" wr = WorldRenderer(br) wr._capacity = cap dbg = br.debug() assert dbg["loadedChunks"] > 0, f"仓库为空: {dbg}" br.set_camera(8.0, 72.0, 8.0, yaw=45.0, pitch=30.0) res = br.render(frames=80) assert res.chunks_rendered > 0, f"没有区块被渲染: {res.raw}" assert res.distinct_colors > 400, f"画面过单调: {res.distinct_colors}" assert res.has_world_content print(f" ✓ 新路径渲染（placed={total_placed}, chunks={res.chunks_rendered}, " f"colors={res.distinct_colors}）") finally: br.stop() def test_real_save_render(): """真实存档经新路径（RLE 路由）渲染 —— 最终验收。""" ok, why = _available() if not ok: print(f" - 跳过（{why}）") return if not (SAVE / "level.dat").is_file(): print(" - 跳过（无测试存档）") return wr = WorldRenderer.open(store=str(STORE), java="java") try: info = wr.load_save(SAVE, center=(8, -7), radius=3) 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) d = wr.debug() assert d["loadedChunks"] > 0, d res = wr.screenshot(frames=90) assert res.chunks_rendered > 0, res.raw assert res.distinct_colors > 2000, f"真实地形应色彩丰富: {res.distinct_colors}" assert res.has_world_content 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)