✨ feat: WP-09 Live2D/3D——模型导入校验与非破坏命令历史

- 迁移 18:avatar_model(live2d/vrm/gltf、validated/rejected、校验报告、来源许可、当前命令指针)+ avatar_command(十类命令检查约束、(avatar,sequence) 唯一、undone 标记)
- GLB/glTF 解析器(纯 Python):GLB magic/版本/chunk 结构校验、JSON 场景图引用一致性(scene→node→mesh→material 逐层索引越界检测)、VRM 0.x/1.0 扩展检测与 meta 提取
- 资源上限(恶意模型文件防线):文件 200MB、节点 10 万、网格 1 万、纹理 2 千、材质 5 千、动画 1 千——超限入库 rejected 并保留报告
- Cubism model3.json 校验:Version/FileReferences 必需、moc3 引用、纹理列表、表情/动作/HitArea 计数
- 上传 MIME 白名单扩展 model/ 与 application/json
- 命令历史:十类命令(param_map/expression/physics/motion/material/texture/bone/attachment/lighting/camera)追加(sequence 单调)、撤销(标记 undone 非破坏)、rejected 模型不接受编辑(422 前置校验)
- 测试 111 项全绿(+9):合法 GLB/VRM 检测/损坏文件/引用越界/超限/Cubism 正反例/命令历史与撤销幂等/rejected 拒编
This commit is contained in:
chenxin committed 2026-09-14 10:34:52 +08:00
1 parent 9dafb8fe72
commit 7c57d27494
7 files changed
+839

No files matched your search

+219
View File
@@ -1916,3 +1916,222 @@ class TestSandbox:
finally:
os.environ.pop("HUNJING_SANDBOX_ENABLED", None)
get_settings.cache_clear()
def _make_glb(gltf_dict, bin_data: bytes = b"") -> bytes:
import json as _json
import struct as _struct
json_bytes = _json.dumps(gltf_dict).encode() + b" "
while len(json_bytes) % 4:
json_bytes += b" "
chunks = _struct.pack("<II", len(json_bytes), 0x4E4F534A) + json_bytes
if bin_data:
pad = (-len(bin_data)) % 4
b = bin_data + b"\x00" * pad
chunks += _struct.pack("<II", len(b), 0x004E4942) + b
total = 12 + len(chunks)
return _struct.pack("<III", 0x46546C67, 2, total) + chunks
VALID_GLTF = {
"asset": {"version": "2.0"},
"scenes": [{"nodes": [0]}],
"nodes": [{"mesh": 0, "name": "root"}],
"meshes": [{"primitives": [{"material": 0}]}],
"materials": [{"name": "skin"}],
"textures": [],
}
@pytest.mark.usefixtures("client")
class TestAvatarImport:
async def _upload(self, client, headers, data: bytes, mime: str) -> str:
up = await client.post(
"/v1/uploads", json={"filename": "m.bin", "media_type": mime}, headers=headers
)
assert up.status_code == 201, up.text
await client.request(
"PUT", f"/v1/uploads/{up.json()['id']}/data", content=data, headers=headers
)
done = await client.post(f"/v1/uploads/{up.json()['id']}/completions", headers=headers)
return done.json()["id"]
async def test_valid_gltf_glb_import(self, client) -> None:
headers = await login(client, "av1@example.com")
fid = await self._upload(client, headers, _make_glb(VALID_GLTF), "model/gltf-binary")
resp = await client.post(
"/v1/avatars/imports",
json={"name": "测试模型", "format": "gltf", "file_id": fid}, headers=headers,
)
assert resp.status_code == 201, resp.text
body = resp.json()
assert body["status"] == "validated"
assert body["validation_report"]["stats"]["nodes"] == 1
async def test_vrm_detection(self, client) -> None:
headers = await login(client, "av2@example.com")
vrm = {**VALID_GLTF, "extensionsUsed": ["VRMC_vrm"],
"extensions": {"VRMC_vrm": {"meta": {"title": "测试 VRM"}}}}
fid = await self._upload(client, headers, _make_glb(vrm), "model/gltf-binary")
resp = await client.post(
"/v1/avatars/imports",
json={"format": "vrm", "file_id": fid}, headers=headers,
)
assert resp.json()["status"] == "validated"
report = resp.json()["validation_report"]
assert report["vrm_version"] == "1.0"
assert report["vrm_meta"]["title"] == "测试 VRM"
async def test_corrupt_glb_rejected_with_report(self, client) -> None:
headers = await login(client, "av3@example.com")
fid = await self._upload(client, headers, b"\x00" * 64, "model/gltf-binary")
resp = await client.post(
"/v1/avatars/imports",
json={"format": "gltf", "file_id": fid}, headers=headers,
)
assert resp.json()["status"] == "rejected"
assert any("magic" in e for e in resp.json()["validation_report"]["errors"])
async def test_broken_scene_reference_rejected(self, client) -> None:
headers = await login(client, "av4@example.com")
bad = {"asset": {"version": "2.0"}, "scenes": [{"nodes": [99]}], "nodes": []}
fid = await self._upload(client, headers, _make_glb(bad), "model/gltf-binary")
resp = await client.post(
"/v1/avatars/imports",
json={"format": "gltf", "file_id": fid}, headers=headers,
)
assert resp.json()["status"] == "rejected"
assert any("node" in e for e in resp.json()["validation_report"]["errors"])
async def test_resource_limit_rejection(self, client) -> None:
headers = await login(client, "av5@example.com")
huge = {
"asset": {"version": "2.0"},
"scenes": [{"nodes": []}],
"nodes": [{"name": f"n{i}"} for i in range(100_001)],
}
fid = await self._upload(client, headers, _make_glb(huge), "model/gltf-binary")
resp = await client.post(
"/v1/avatars/imports",
json={"format": "gltf", "file_id": fid}, headers=headers,
)
assert resp.json()["status"] == "rejected"
assert any("上限" in e for e in resp.json()["validation_report"]["errors"])
async def test_cubism_model3_import(self, client) -> None:
import json as _json
headers = await login(client, "av6@example.com")
cubism = {
"Version": 3,
"FileReferences": {
"moc": "model.moc3",
"Textures": ["tex0.png", "tex1.png"],
"Physics": "physics.json",
},
"HitAreas": [{"Id": "HitAreaHead", "Name": "头"}],
}
fid = await self._upload(
client, headers, _json.dumps(cubism).encode(), "application/json"
)
resp = await client.post(
"/v1/avatars/imports",
json={"format": "live2d", "file_id": fid}, headers=headers,
)
assert resp.json()["status"] == "validated"
stats = resp.json()["validation_report"]["stats"]
assert stats["textures"] == 2 and stats["hit_areas"] == 1
async def test_cubism_missing_moc_rejected(self, client) -> None:
import json as _json
headers = await login(client, "av7@example.com")
fid = await self._upload(
client, headers, _json.dumps({"Version": 3, "FileReferences": {}}).encode(),
"application/json",
)
resp = await client.post(
"/v1/avatars/imports",
json={"format": "live2d", "file_id": fid}, headers=headers,
)
assert resp.json()["status"] == "rejected"
assert any("moc3" in e for e in resp.json()["validation_report"]["errors"])
@pytest.mark.usefixtures("client")
class TestAvatarCommands:
async def _import_valid(self, client, headers) -> str:
up = await client.post(
"/v1/uploads",
json={"filename": "m.glb", "media_type": "model/gltf-binary"}, headers=headers
)
await client.request(
"PUT", f"/v1/uploads/{up.json()['id']}/data",
content=_make_glb(VALID_GLTF), headers=headers,
)
done = await client.post(
f"/v1/uploads/{up.json()['id']}/completions", headers=headers
)
resp = await client.post(
"/v1/avatars/imports",
json={"format": "gltf", "file_id": done.json()["id"]}, headers=headers,
)
return resp.json()["id"]
async def test_command_history_and_undo(self, client) -> None:
headers = await login(client, "av8@example.com")
aid = await self._import_valid(client, headers)
c1 = await client.post(
f"/v1/avatars/{aid}/commands",
json={"command_type": "param_map",
"payload": {"ParamAngleX": {"min": -30, "max": 30}}}, headers=headers,
)
assert c1.json()["sequence"] == 1
c2 = await client.post(
f"/v1/avatars/{aid}/commands",
json={"command_type": "expression",
"payload": {"name": "smile", "params": {}}}, headers=headers,
)
assert c2.json()["sequence"] == 2
# 非法命令类型
bad = await client.post(
f"/v1/avatars/{aid}/commands",
json={"command_type": "delete_file", "payload": {}}, headers=headers,
)
assert bad.status_code == 422
# 撤销:幂等 + 历史保留
undo = await client.post(
f"/v1/avatars/{aid}/commands/{c2.json()['id']}/undoings", headers=headers
)
assert undo.json()["undone"] is True
again = await client.post(
f"/v1/avatars/{aid}/commands/{c2.json()['id']}/undoings", headers=headers
)
assert again.json()["undone"] is True # 幂等
history = await client.get(f"/v1/avatars/{aid}/commands", headers=headers)
seqs = [c["sequence"] for c in history.json()["data"]]
assert seqs == [1, 2] # 非破坏:撤销不删除
async def test_rejected_model_not_editable(self, client) -> None:
headers = await login(client, "av9@example.com")
up = await client.post(
"/v1/uploads",
json={"filename": "bad.glb", "media_type": "model/gltf-binary"}, headers=headers
)
await client.request(
"PUT", f"/v1/uploads/{up.json()['id']}/data",
content=b"garbage", headers=headers,
)
done = await client.post(
f"/v1/uploads/{up.json()['id']}/completions", headers=headers
)
model = await client.post(
"/v1/avatars/imports",
json={"format": "gltf", "file_id": done.json()["id"]}, headers=headers,
)
cmd = await client.post(
f"/v1/avatars/{model.json()['id']}/commands",
json={"command_type": "param_map", "payload": {}}, headers=headers,
)
assert cmd.status_code == 422 # rejected 模型不接受编辑