feat(mcfold): P4 区块解析（位打包/三种调色板分支/注册表显式降级）+ prismarine 真实数据交叉验证
This commit is contained in:
pi
@ -41,3 +41,12 @@
|
||||
| 协议**不需重写**，只需解释声明式 JSON | 111 个 play 包由同一份 `protocol.json` 驱动 |
|
||||
| Python 客户端能真连真服 | Node 侧日志 `SERVER_GOT_LOGIN username= PyProbe` |
|
||||
| 协议号由实例文件夹决定 | `1.20.1/version.json` → 763 |
|
||||
| 区块格式**必须自己解**（协议层是声明式，但 chunkData 是不透明 buffer） | `protocol.json` 里 chunkData 只声明为 buffer |
|
||||
| Python 解码器能解独立实现产出的区块 | prismarine-chunk 产样本 → 4/4 方块命中 |
|
||||
| mod 方块在本机注册表里**显式 unknown**，而非静默错配 | `state 99999 -> unknown(99999)` |
|
||||
|
||||
## 进度（续）
|
||||
|
||||
- [x] **P4** 区块解析（`mcfold/chunk.py`，11 项回归全绿）
|
||||
位打包 / 单值·本地·全局调色板 / 生物群系 section / 世界边界
|
||||
**关键行为**：越界 state id ⇒ `unknown(id)`（对照 mineflayer 的静默错配）
|
||||
|
||||
BIN
mcfold/__pycache__/chunk.cpython-312.pyc
Normal file
BIN
mcfold/__pycache__/chunk.cpython-312.pyc
Normal file
Binary file not shown.
265
mcfold/chunk.py
Normal file
265
mcfold/chunk.py
Normal file
@ -0,0 +1,265 @@
|
||||
"""区块解析：chunkData buffer → 方块（state id / 方块名）。
|
||||
|
||||
为什么这段要自己写（对照 P2 的协议层）：
|
||||
协议层是**声明式**的（protodef 解释 JSON 即可），
|
||||
但 `packet_map_chunk` 里的 `chunkData` 是一个 **不透明 buffer**
|
||||
—— protodef 只声明它是 buffer，内部结构不在 protocol.json 里。
|
||||
所以区块格式必须按 MC 的固定布局自己解。
|
||||
|
||||
1.20 的区块格式（实测自 prismarine-chunk src/pc/1.18/ChunkColumn.js:253）：
|
||||
minY = -64, worldHeight = 384 ⇒ 24 个 section
|
||||
每个 section 交替出现：
|
||||
[方块 section] [生物群系 section]
|
||||
方块 section:
|
||||
solidBlockCount : i16 (大端)
|
||||
bitsPerBlock : u8
|
||||
== 0 -> 单一值调色板: varint(stateId) + varint(0) ← 后一个是 longs 数
|
||||
> MAX_BITS(8) -> 全局调色板: 直接是 state id，16 位/值
|
||||
否则 -> 本地调色板: varint(paletteLen) + palette[] + 位数组
|
||||
位数组（1.20 有 size 前缀；1.21.5+ 才去掉）:
|
||||
longs : varint
|
||||
data : longs 个 64 位值；每个 long 按 [高 u32, 低 u32] 存储
|
||||
|
||||
关键坑：**值不跨 long 边界**（valuesPerLong = floor(64/bitsPerValue)），
|
||||
即一个 64 位 long 里整齐放若干个值。这与我最初以为的"紧密打包"不同。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from .protodef import ProtoError, Reader
|
||||
|
||||
# ---- 1.20 区块常量（实测自 prismarine-chunk constants.js + 1.18/ChunkColumn.js）----
|
||||
MIN_Y = -64
|
||||
WORLD_HEIGHT = 384
|
||||
NUM_SECTIONS = WORLD_HEIGHT >> 4 # 24
|
||||
SECTION_VOLUME = 16 * 16 * 16 # 4096
|
||||
BIOME_SECTION_VOLUME = SECTION_VOLUME // (4 * 4 * 4) # 64
|
||||
|
||||
MIN_BITS_PER_BLOCK = 4
|
||||
MAX_BITS_PER_BLOCK = 8 # 超过它就转全局调色板
|
||||
GLOBAL_BITS_PER_BLOCK = 16
|
||||
|
||||
MIN_BITS_PER_BIOME = 1
|
||||
MAX_BITS_PER_BIOME = 3
|
||||
GLOBAL_BITS_PER_BIOME = 6
|
||||
|
||||
|
||||
class ChunkError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class BitArray:
|
||||
"""MC 的位打包数组（每个 long 内整齐放 floor(64/bits) 个值，不跨边界）。"""
|
||||
|
||||
__slots__ = ("data", "capacity", "bits_per_value", "values_per_long", "value_mask")
|
||||
|
||||
def __init__(self, data: list[int], capacity: int, bits_per_value: int):
|
||||
self.data = data # 每个元素是一个 64 位 long
|
||||
self.capacity = capacity
|
||||
self.bits_per_value = bits_per_value
|
||||
self.values_per_long = 64 // bits_per_value
|
||||
self.value_mask = (1 << bits_per_value) - 1
|
||||
|
||||
@classmethod
|
||||
def from_buffer(cls, r: Reader, capacity: int, bits_per_value: int,
|
||||
size_prefix: bool = True) -> "BitArray":
|
||||
"""按 MC 的在线格式读位数组。"""
|
||||
if bits_per_value <= 0 or bits_per_value > 32:
|
||||
raise ChunkError(f"非法 bitsPerValue: {bits_per_value}")
|
||||
vpl = 64 // bits_per_value
|
||||
if size_prefix:
|
||||
longs = r.varint()
|
||||
else:
|
||||
longs = (capacity + vpl - 1) // vpl
|
||||
if longs < 0 or longs > 1_000_000:
|
||||
raise ChunkError(f"位数组长度异常: {longs}")
|
||||
data: list[int] = []
|
||||
for _ in range(longs):
|
||||
# JS 侧：先读高 u32 再读低 u32 → long = (high<<32)|low
|
||||
high = int.from_bytes(r.read(4), "big")
|
||||
low = int.from_bytes(r.read(4), "big")
|
||||
data.append((high << 32) | low)
|
||||
return cls(data, capacity, bits_per_value)
|
||||
|
||||
def get(self, index: int) -> int:
|
||||
if index < 0 or index >= self.capacity:
|
||||
return 0
|
||||
start = index // self.values_per_long
|
||||
off = (index - start * self.values_per_long) * self.bits_per_value
|
||||
if start >= len(self.data):
|
||||
return 0
|
||||
# 值不跨 long 边界 ⇒ 单个 long 内移位即可
|
||||
return (self.data[start] >> off) & self.value_mask
|
||||
|
||||
def to_list(self) -> list[int]:
|
||||
return [self.get(i) for i in range(self.capacity)]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Section:
|
||||
"""一个 16x16x16 的方块 section。"""
|
||||
y_index: int # 0..23（对应世界 y = MIN_Y + 16*y_index）
|
||||
solid_block_count: int = 0
|
||||
bits_per_block: int = 0
|
||||
palette: list[int] = field(default_factory=list) # 本地调色板（indirect 时用）
|
||||
bits: BitArray | None = None
|
||||
single_value: int | None = None # bits==0 时的单一 state id
|
||||
|
||||
def state_id_at(self, x: int, y: int, z: int) -> int:
|
||||
"""section 内 (x,y,z) 的全局 state id。索引顺序 = (y<<8)|(z<<4)|x。"""
|
||||
idx = (y << 8) | (z << 4) | x
|
||||
if self.bits_per_block == 0:
|
||||
return self.single_value or 0
|
||||
raw = self.bits.get(idx) if self.bits else 0
|
||||
if self.palette:
|
||||
# 本地调色板：raw 是调色板下标
|
||||
return self.palette[raw] if 0 <= raw < len(self.palette) else 0
|
||||
return raw # 全局调色板：raw 就是 state id
|
||||
|
||||
|
||||
@dataclass
|
||||
class BiomeSection:
|
||||
bits_per_biome: int = 0
|
||||
palette: list[int] = field(default_factory=list)
|
||||
bits: BitArray | None = None
|
||||
single_value: int | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Chunk:
|
||||
"""一个区块列（16 x 384 x 16）。"""
|
||||
x: int
|
||||
z: int
|
||||
sections: list[Section] = field(default_factory=list)
|
||||
biomes: list[BiomeSection] = field(default_factory=list)
|
||||
|
||||
def state_id_at(self, x: int, y: int, z: int) -> int:
|
||||
"""世界坐标 → state id（y 是绝对高度）。"""
|
||||
if y < MIN_Y or y >= MIN_Y + WORLD_HEIGHT:
|
||||
return 0
|
||||
sec = (y - MIN_Y) >> 4
|
||||
if sec >= len(self.sections):
|
||||
return 0
|
||||
return self.sections[sec].state_id_at(x & 15, y & 15, z & 15)
|
||||
|
||||
|
||||
def _read_block_section(r: Reader) -> Section:
|
||||
solid = int.from_bytes(r.read(2), "big", signed=True)
|
||||
bits = r.read(1)[0]
|
||||
if bits > 16:
|
||||
raise ChunkError(f"bitsPerBlock 过大: {bits}")
|
||||
|
||||
sec = Section(y_index=-1, solid_block_count=solid, bits_per_block=bits)
|
||||
|
||||
if bits == 0:
|
||||
# 单一值：varint(stateId) 后跟 varint(0)（longs 数）
|
||||
sec.single_value = r.varint()
|
||||
r.varint() # longs = 0
|
||||
return sec
|
||||
|
||||
if bits > MAX_BITS_PER_BLOCK:
|
||||
# 全局调色板：直接 state id
|
||||
sec.bits = BitArray.from_buffer(r, SECTION_VOLUME, GLOBAL_BITS_PER_BLOCK,
|
||||
size_prefix=True)
|
||||
sec.bits_per_block = GLOBAL_BITS_PER_BLOCK
|
||||
return sec
|
||||
|
||||
# 本地调色板
|
||||
n = r.varint()
|
||||
if n < 0 or n > 4096:
|
||||
raise ChunkError(f"调色板长度异常: {n}")
|
||||
sec.palette = [r.varint() for _ in range(n)]
|
||||
sec.bits = BitArray.from_buffer(r, SECTION_VOLUME, bits, size_prefix=True)
|
||||
return sec
|
||||
|
||||
|
||||
def _read_biome_section(r: Reader) -> BiomeSection:
|
||||
bits = r.read(1)[0]
|
||||
if bits > 8:
|
||||
raise ChunkError(f"bitsPerBiome 过大: {bits}")
|
||||
sec = BiomeSection(bits_per_biome=bits)
|
||||
|
||||
if bits == 0:
|
||||
sec.single_value = r.varint()
|
||||
r.varint() # longs = 0
|
||||
return sec
|
||||
|
||||
if bits > MAX_BITS_PER_BIOME:
|
||||
sec.bits = BitArray.from_buffer(r, BIOME_SECTION_VOLUME, GLOBAL_BITS_PER_BIOME,
|
||||
size_prefix=True)
|
||||
sec.bits_per_biome = GLOBAL_BITS_PER_BIOME
|
||||
return sec
|
||||
|
||||
n = r.varint()
|
||||
if n < 0 or n > 1024:
|
||||
raise ChunkError(f"生物群系调色板长度异常: {n}")
|
||||
sec.palette = [r.varint() for _ in range(n)]
|
||||
sec.bits = BitArray.from_buffer(r, BIOME_SECTION_VOLUME, bits, size_prefix=True)
|
||||
return sec
|
||||
|
||||
|
||||
def decode_chunk(x: int, z: int, chunk_data: bytes,
|
||||
num_sections: int = NUM_SECTIONS) -> Chunk:
|
||||
"""把 packet_map_chunk 的 chunkData 解成区块列。
|
||||
|
||||
num_sections 由世界的 worldHeight 决定（1.20 = 24）；传入以适配不同版本。
|
||||
"""
|
||||
r = Reader(chunk_data)
|
||||
chunk = Chunk(x=x, z=z)
|
||||
|
||||
for i in range(num_sections):
|
||||
if r.remaining() <= 0:
|
||||
break
|
||||
sec = _read_block_section(r)
|
||||
sec.y_index = i
|
||||
chunk.sections.append(sec)
|
||||
# 每个方块 section 后紧跟一个生物群系 section
|
||||
if r.remaining() > 0:
|
||||
chunk.biomes.append(_read_biome_section(r))
|
||||
|
||||
return chunk
|
||||
|
||||
|
||||
# ---------------- state id → 方块名（方块注册表） ----------------
|
||||
|
||||
class BlockRegistry:
|
||||
"""方块注册表：state id ↔ 方块名。
|
||||
|
||||
mod 支持的落点正在这里：原版表是内置的（minecraft-data 的 blocks.json），
|
||||
而 mod 方块需要另外的途径补进来（见文档附录 D/E）。找不到时必须**显式**报
|
||||
"unknown"，而不是像 mineflayer 那样静默映射成别的方块。
|
||||
"""
|
||||
|
||||
def __init__(self, blocks: list[dict[str, Any]]):
|
||||
self.blocks = blocks
|
||||
self.by_state: dict[int, str] = {}
|
||||
self.by_name: dict[str, dict[str, Any]] = {}
|
||||
for b in blocks:
|
||||
name = b["name"]
|
||||
self.by_name[name] = b
|
||||
lo = b.get("minStateId", 0)
|
||||
hi = b.get("maxStateId", lo)
|
||||
for sid in range(lo, hi + 1):
|
||||
self.by_state[sid] = name
|
||||
self.max_state_id = max(self.by_state) if self.by_state else -1
|
||||
|
||||
@classmethod
|
||||
def from_json_file(cls, path: str) -> "BlockRegistry":
|
||||
import json
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
return cls(json.load(f))
|
||||
|
||||
def name_of(self, state_id: int) -> str | None:
|
||||
"""state id → 方块名；越界返回 None（**不**静默回退）。"""
|
||||
return self.by_state.get(state_id)
|
||||
|
||||
def name_or_unknown(self, state_id: int) -> str:
|
||||
n = self.name_of(state_id)
|
||||
return n if n is not None else f"unknown({state_id})"
|
||||
|
||||
def is_out_of_range(self, state_id: int) -> bool:
|
||||
"""越界 ⇒ 该 id 极可能来自 mod 方块（本机注册表没有它）。"""
|
||||
return state_id > self.max_state_id
|
||||
9
tests/support_testserver.js
Normal file
9
tests/support_testserver.js
Normal file
@ -0,0 +1,9 @@
|
||||
// 供测试用的 Node 服务端（独立实现）。
|
||||
// 用法: node support_testserver.js <port>
|
||||
const mc = require('minecraft-protocol')
|
||||
const port = parseInt(process.argv[2] || '25599')
|
||||
const server = mc.createServer({ 'online-mode': false, port, version: '1.20.1', motd: 'mcfold-test' })
|
||||
server.on('listening', () => console.log('SERVER_READY'))
|
||||
server.on('login', (c) => console.log('SERVER_GOT_LOGIN username=', c.username))
|
||||
server.on('error', e => console.log('SERVER_ERR', e.message))
|
||||
setTimeout(() => process.exit(0), 120000)
|
||||
155
tests/test_chunk.py
Normal file
155
tests/test_chunk.py
Normal file
@ -0,0 +1,155 @@
|
||||
"""chunk 解析回归。
|
||||
|
||||
核心验证方式：用 **prismarine-chunk（独立实现）产出的真实格式数据**
|
||||
交给 Python 解 —— 两边独立，对得上才算格式对。
|
||||
样本由 /home/newqqagent/plugin_data/mc/runtime/chunkgen*.js 生成。
|
||||
"""
|
||||
import sys, os, json, subprocess, tempfile
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
from mcfold.chunk import (decode_chunk, BlockRegistry, MIN_Y, NUM_SECTIONS,
|
||||
BitArray, Reader, ChunkError)
|
||||
|
||||
RUNTIME = "/home/newqqagent/plugin_data/mc/runtime"
|
||||
BLOCKS = f"{RUNTIME}/node_modules/minecraft-data/minecraft-data/data/pc/1.20/blocks.json"
|
||||
|
||||
def gen(script):
|
||||
"""跑 Node 生成独立实现产出的样本。"""
|
||||
r = subprocess.run(["node", f"{RUNTIME}/{script}"], capture_output=True, text=True,
|
||||
cwd=RUNTIME, timeout=120)
|
||||
if r.returncode != 0:
|
||||
return None, r.stderr[:300]
|
||||
return json.loads(r.stdout), None
|
||||
|
||||
def reg():
|
||||
return BlockRegistry.from_json_file(BLOCKS)
|
||||
|
||||
def test_bitarray():
|
||||
"""位打包：值不跨 long 边界（这是最容易写错的地方）。"""
|
||||
# 4 位/值，一个 long 放 16 个值
|
||||
ba = BitArray(data=[0x3210], capacity=16, bits_per_value=4)
|
||||
assert [ba.get(i) for i in range(4)] == [0, 1, 2, 3], [ba.get(i) for i in range(4)]
|
||||
print(" ✓ 位打包：值不跨 long 边界（4位/值 → 16 值/long）")
|
||||
# 5 位/值 → floor(64/5)=12 值/long
|
||||
ba2 = BitArray(data=[0x1], capacity=12, bits_per_value=5)
|
||||
assert ba2.values_per_long == 12, ba2.values_per_long
|
||||
assert ba2.get(0) == 1
|
||||
print(" ✓ 位打包：5位/值 → 12 值/long（余位丢弃）")
|
||||
|
||||
def test_real_chunk_from_prismarine():
|
||||
"""核心回归：解独立实现产出的区块。"""
|
||||
d, err = gen("chunkgen.js")
|
||||
if d is None:
|
||||
print(f" - 跳过（Node 生成失败: {err}）"); return
|
||||
exp = d["expect"]
|
||||
ch = decode_chunk(0, 0, bytes.fromhex(d["hex"]))
|
||||
assert len(ch.sections) == NUM_SECTIONS, len(ch.sections)
|
||||
r = reg()
|
||||
checks = [((0,-60,0),"stone"), ((1,-59,2),"gold_block"),
|
||||
((5,-59,7),"diamond_block"), ((3,20,4),"diamond_block")]
|
||||
for (x,y,z), want in checks:
|
||||
sid = ch.state_id_at(x,y,z)
|
||||
got = r.name_or_unknown(sid)
|
||||
assert got == want, f"({x},{y},{z}) {sid} -> {got}, 期望 {want}"
|
||||
nonempty = [s.y_index for s in ch.sections if s.solid_block_count > 0]
|
||||
assert nonempty == [0,5], nonempty
|
||||
print(f" ✓ 解 prismarine 真实区块: 4/4 方块命中, 非空 section={nonempty}")
|
||||
|
||||
def test_all_encoding_branches():
|
||||
"""覆盖单值/本地调色板/高 section 三种编码分支。"""
|
||||
d, err = gen("chunkgen2.js")
|
||||
if d is None:
|
||||
print(f" - 跳过（Node 生成失败: {err}）"); return
|
||||
r = reg()
|
||||
# A: 全空气 → 应无实心方块
|
||||
chA = decode_chunk(0,0,bytes.fromhex(d["A"]["hex"]))
|
||||
assert all(s.solid_block_count == 0 for s in chA.sections), "全空气却有实心方块"
|
||||
print(" ✓ 分支 A（全空气）: solidBlockCount 全 0")
|
||||
# B: 一行石头
|
||||
chB = decode_chunk(0,0,bytes.fromhex(d["B"]["hex"]))
|
||||
names = {r.name_or_unknown(chB.state_id_at(i,0,0)) for i in range(16)}
|
||||
assert names == {"stone"}, names
|
||||
print(f" ✓ 分支 B（一行石头）: y=0 的 16 个方块全是 stone")
|
||||
# C: 四种混合 → 验证本地调色板
|
||||
chC = decode_chunk(0,0,bytes.fromhex(d["C"]["hex"]))
|
||||
got = [r.name_or_unknown(chC.state_id_at(i,60,0)) for i in range(4)]
|
||||
assert got == ["stone","dirt","gold_block","diamond_block"], got
|
||||
print(f" ✓ 分支 C（本地调色板）: {got}")
|
||||
# D: 高 section
|
||||
chD = decode_chunk(0,0,bytes.fromhex(d["D"]["hex"]))
|
||||
assert r.name_or_unknown(chD.state_id_at(3,20,4)) == "diamond_block"
|
||||
print(" ✓ 分支 D（高 section y=20）")
|
||||
|
||||
def test_mod_block_explicit_unknown():
|
||||
"""★ 关键行为：本机注册表没有的 state id 必须显式 unknown，**不可静默错配**。
|
||||
|
||||
这正是 mineflayer 的缺陷（越界 id 静默变成 note_block），我们要把它变成可见降级。
|
||||
|
||||
说明：格式正确性已由 test_real_chunk_from_prismarine（独立实现数据，4/4）证明；
|
||||
这里聚焦**注册表行为**，故用 Python 侧的 writer 构造样本（自洽即可）。
|
||||
"""
|
||||
from mcfold.protodef import Writer
|
||||
w = Writer()
|
||||
# section 0: solidCount=1, bits=4, palette=[99999], longs=256, 全 0 位
|
||||
w.write((1).to_bytes(2, "big"))
|
||||
w.write(bytes([4]))
|
||||
w.varint(1); w.varint(99999)
|
||||
w.varint(256)
|
||||
w.write(b"\x00" * (256 * 8))
|
||||
# biome section（单值）
|
||||
w.write(bytes([0])); w.varint(0); w.varint(0)
|
||||
# 其余 23 个 section：单值空气 + biome
|
||||
for _ in range(23):
|
||||
w.write((0).to_bytes(2, "big")); w.write(bytes([0]))
|
||||
w.varint(0); w.varint(0)
|
||||
w.write(bytes([0])); w.varint(0); w.varint(0)
|
||||
|
||||
ch = decode_chunk(0, 0, w.getvalue())
|
||||
# 末尾 section 属半写状态，取已完整解出的 section 0（y=-64..-49）
|
||||
sid = ch.state_id_at(0, -64, 0)
|
||||
assert sid == 99999, f"state id 解码错: {sid}"
|
||||
|
||||
r = reg()
|
||||
got = r.name_or_unknown(sid)
|
||||
assert got.startswith("unknown("), f"越界 id 竟解析出了名字: {got}"
|
||||
assert r.is_out_of_range(sid), "越界判定失效"
|
||||
# 对照：mineflayer 的行为是静默错配 —— 我们明确不同
|
||||
assert r.name_of(sid) is None, "越界 id 不应能查到名字"
|
||||
print(f" ✓ ★ mod 方块显式降级: state {sid} -> {got}（而非静默错配）")
|
||||
|
||||
def test_registry_degradation_direct():
|
||||
"""直接验证注册表降级语义（不依赖线格式）。"""
|
||||
r = reg()
|
||||
assert r.name_or_unknown(0) == "air"
|
||||
assert r.name_or_unknown(24134) is not None # 原版上界内
|
||||
assert r.name_or_unknown(24135).startswith("unknown(") # 超出上界
|
||||
assert r.is_out_of_range(50000)
|
||||
print(" ✓ 注册表降级语义：原版内可查、越界显式 unknown")
|
||||
|
||||
def test_out_of_bounds_safe():
|
||||
"""世界边界外的查询必须安全返回空气，而不是崩。"""
|
||||
from mcfold.protodef import Writer
|
||||
w = Writer()
|
||||
for _ in range(NUM_SECTIONS):
|
||||
w.write((0).to_bytes(2, "big")); w.write(bytes([0]))
|
||||
w.varint(0); w.varint(0) # 方块 section（单值空气）
|
||||
w.write(bytes([0])); w.varint(0); w.varint(0) # biome section
|
||||
ch = decode_chunk(0, 0, w.getvalue())
|
||||
assert ch.state_id_at(0, -1000, 0) == 0
|
||||
assert ch.state_id_at(0, 1000, 0) == 0
|
||||
assert ch.state_id_at(0, MIN_Y, 0) == 0
|
||||
print(" ✓ 世界边界外查询安全返回空气")
|
||||
|
||||
def test_bad_data_raises():
|
||||
"""损坏数据必须报错，不能静默产出错数据。"""
|
||||
try:
|
||||
decode_chunk(0,0,b"\x00\x01\xff") # bitsPerBlock=255
|
||||
assert False, "应当报错"
|
||||
except (ChunkError, Exception) as e:
|
||||
print(f" ✓ 损坏数据显式报错: {str(e)[:45]}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
for fn in [test_bitarray, test_real_chunk_from_prismarine, test_all_encoding_branches,
|
||||
test_mod_block_explicit_unknown, test_registry_degradation_direct,
|
||||
test_out_of_bounds_safe, test_bad_data_raises]:
|
||||
fn()
|
||||
print("\nchunk 解析：全部通过")
|
||||
@ -19,9 +19,39 @@ def reachable():
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
def _start_server():
|
||||
"""自动起 Node 服务端（独立实现），让本测试**总能真跑**而不是跳过。"""
|
||||
import subprocess, time, shutil
|
||||
runtime = "/home/newqqagent/plugin_data/mc/runtime"
|
||||
script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "support_testserver.js")
|
||||
if not shutil.which("node"):
|
||||
return None
|
||||
shutil.copy(script, os.path.join(runtime, "support_testserver.js"))
|
||||
p = subprocess.Popen(["node", "support_testserver.js", str(PORT)], cwd=runtime,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
|
||||
for _ in range(50):
|
||||
if reachable(): return p
|
||||
if p.poll() is not None: return None
|
||||
time.sleep(0.2)
|
||||
return None
|
||||
|
||||
def test_interop():
|
||||
proc = None
|
||||
if not reachable():
|
||||
print(f" - 服务端 {HOST}:{PORT} 未运行，跳过（见本文件顶部用法）"); return
|
||||
proc = _start_server()
|
||||
if not reachable():
|
||||
print(f" - 服务端 {HOST}:{PORT} 不可用，跳过"); return
|
||||
try:
|
||||
_do_interop()
|
||||
finally:
|
||||
if proc:
|
||||
proc.terminate()
|
||||
try: proc.wait(timeout=5)
|
||||
except Exception: proc.kill()
|
||||
try: os.remove("/home/newqqagent/plugin_data/mc/runtime/support_testserver.js")
|
||||
except OSError: pass
|
||||
|
||||
def _do_interop():
|
||||
c = Client(HOST, PORT, "PyProbe", "1.20.1")
|
||||
try:
|
||||
r = c.run_smoke()
|
||||
|
||||
Reference in New Issue
Block a user