zcbot/tests/test_blender_adapter.py

270 lines
11 KiB
Python

from __future__ import annotations
import importlib.util
import io
import json
import tempfile
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
from uuid import uuid4
from core.software_contracts import SoftwareContractError, get_contract
ROOT = Path(__file__).resolve().parents[1]
ADAPTER = ROOT / "windows-node" / "adapters" / "blender.scene.author@v3"
LAUNCHER_PATH = ADAPTER / "worker.py"
spec = importlib.util.spec_from_file_location("blender_adapter_launcher", LAUNCHER_PATH)
assert spec is not None and spec.loader is not None
launcher = importlib.util.module_from_spec(spec)
spec.loader.exec_module(launcher)
def request() -> dict[str, object]:
return {
"schema_version": 3,
"inputs": [],
"operation": {
"scene": {
"type": "recipe",
"recipe_version": 1,
"title": "水泥实验室设备布置",
"units": "meters",
"materials": [
{
"id": "concrete",
"base_color": "#B8BCC2",
"metallic": 0,
"roughness": 0.8,
},
{
"id": "steel",
"base_color": "#55606A",
"metallic": 0.75,
"roughness": 0.28,
},
],
"objects": [
{
"id": "floor",
"kind": "geometry",
"geometry": {"type": "box", "size": [12, 8, 0.2]},
"transform": {"position": [0, 0, -0.1]},
"material": "concrete",
"modifiers": [{"type": "bevel", "width": 0.03, "segments": 2}],
},
{
"id": "process_pipe",
"kind": "geometry",
"geometry": {
"type": "pipe",
"start": [-2, 0, 1],
"end": [2, 0, 1],
"outer_radius": 0.12,
"wall_thickness": 0.01,
},
"material": "steel",
},
{
"id": "kiln",
"kind": "generator",
"generator": "rotary_kiln",
"parameters": {
"length_m": 6,
"outer_diameter_m": 0.8,
"shell_thickness_m": 0.02,
"lining_thickness_m": 0.08,
"slope_percent": 3.5,
"support_stations": 3,
"drive_system": True,
},
"transform": {"position": [0, 0, 1.2]},
},
],
"lights": [
{
"id": "key",
"type": "area",
"position": [3, -5, 7],
"target": [0, 0, 0],
"energy": 1200,
"size": 5,
}
],
"render": {
"width": 1280,
"height": 720,
"background_color": "#20252B",
"views": [
{
"name": "总览",
"output_key": "preview_01",
"position": [10, -12, 8],
"target": [0, 0, 1],
"projection": "perspective",
"lens_mm": 50,
},
{
"name": "正视",
"output_key": "preview_02",
"position": [0, -15, 3],
"target": [0, 0, 1],
"projection": "orthographic",
"ortho_scale": 12,
},
],
},
}
},
"outputs": [
{"key": "preview_02", "type": "figure", "format": "png"},
{"key": "scene_glb", "type": "model", "format": "glb"},
],
}
class BlenderContractTests(unittest.TestCase):
def test_scene_recipe_contract_is_workspace_backed(self) -> None:
contract = get_contract("blender.scene.author@v3")
normalized, digest = contract.normalize_request(request())
self.assertEqual(normalized, request())
self.assertEqual(len(digest), 64)
self.assertEqual(contract.workspace.state_output, "project")
self.assertEqual(
contract.workspace.preview_outputs,
tuple(f"preview_{index:02d}" for index in range(1, 9)),
)
self.assertEqual(contract.required_adapter_version(normalized), "1.0.0")
self.assertEqual(
contract.summarize(normalized)["title"],
"水泥实验室设备布置",
)
self.assertEqual(
set(contract.expected_outputs(normalized)),
{
"project",
"preview_01",
"preview_02",
"scene_manifest",
"scene_glb",
"provenance",
},
)
def test_contract_supports_all_managed_geometry_kinds(self) -> None:
value = request()
objects = value["operation"]["scene"]["objects"] # type: ignore[index]
objects[:] = [
{"id": "box", "kind": "geometry", "geometry": {"type": "box", "size": [1, 1, 1]}},
{"id": "cylinder", "kind": "geometry", "geometry": {"type": "cylinder", "radius": 1, "depth": 2}},
{"id": "cone", "kind": "geometry", "geometry": {"type": "cone", "radius1": 1, "radius2": 0, "depth": 2}},
{"id": "sphere", "kind": "geometry", "geometry": {"type": "sphere", "radius": 1}},
{"id": "torus", "kind": "geometry", "geometry": {"type": "torus", "major_radius": 1, "minor_radius": 0.2}},
{"id": "plane", "kind": "geometry", "geometry": {"type": "plane", "size": [2, 2]}},
{"id": "extrude", "kind": "geometry", "geometry": {"type": "extrude", "profile": [[0, 0], [1, 0], [0, 1]], "depth": 1}},
{"id": "revolve", "kind": "geometry", "geometry": {"type": "revolve", "profile": [[0.5, 0], [1, 1]]}},
{"id": "sweep", "kind": "geometry", "geometry": {"type": "sweep", "path": [[0, 0, 0], [1, 0, 1]], "radius": 0.1}},
{"id": "pipe", "kind": "geometry", "geometry": {"type": "pipe", "start": [0, 0, 0], "end": [1, 0, 0], "outer_radius": 0.2, "wall_thickness": 0.02}},
]
get_contract("blender.scene.author@v3").normalize_request(value)
def test_contract_rejects_arbitrary_execution_and_unsafe_scale(self) -> None:
contract = get_contract("blender.scene.author@v3")
malicious = request()
malicious["operation"]["scene"]["python"] = "import os" # type: ignore[index]
with self.assertRaises(SoftwareContractError):
contract.normalize_request(malicious)
oversized = request()
oversized["operation"]["scene"]["render"]["width"] = 8192 # type: ignore[index]
with self.assertRaises(SoftwareContractError):
contract.normalize_request(oversized)
invalid_kind = request()
invalid_kind["operation"]["scene"]["objects"][0]["kind"] = "script" # type: ignore[index]
with self.assertRaises(SoftwareContractError):
contract.normalize_request(invalid_kind)
class BlenderLauncherTests(unittest.TestCase):
def test_probe_reports_detected_blender(self) -> None:
with patch.object(
launcher,
"_find_blender",
return_value=(Path(r"C:\Program Files\Blender\blender.exe"), "5.2.0"),
):
output = io.StringIO()
with redirect_stdout(output):
self.assertEqual(launcher.probe(), 0)
payload = json.loads(output.getvalue())
self.assertEqual(payload["health"], "ready")
self.assertEqual(payload["software_version"], "5.2.0")
self.assertEqual(payload["adapter_version"], "1.0.0")
def test_launcher_uses_fixed_background_and_autoexec_flags(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
job_dir = Path(temporary)
(job_dir / "request").mkdir()
record = {
"job_id": str(uuid4()),
"lease_id": str(uuid4()),
"request_digest": "a" * 64,
"request": request(),
}
(job_dir / "request" / "request.json").write_text(
json.dumps(record), encoding="utf-8"
)
executable = Path(r"C:\Program Files\Blender\blender.exe")
with (
patch.object(launcher, "_find_blender", return_value=(executable, "4.5.3")),
patch.object(
launcher.subprocess,
"run",
return_value=SimpleNamespace(returncode=9),
) as process,
):
self.assertEqual(launcher.run(job_dir), 9)
arguments = process.call_args.args[0]
self.assertEqual(arguments[0], str(executable))
self.assertIn("--background", arguments)
self.assertIn("--factory-startup", arguments)
self.assertIn("--disable-autoexec", arguments)
self.assertIn("--python-exit-code", arguments)
self.assertEqual(arguments[arguments.index("--python-exit-code") + 1], "1")
self.assertIn("--python", arguments)
terminal = json.loads((job_dir / "terminal.json").read_text(encoding="utf-8"))
self.assertEqual(terminal["status"], "failed")
self.assertEqual(terminal["error"]["code"], "BLENDER_NO_TERMINAL")
def test_bpy_worker_is_a_bounded_scene_recipe_compiler(self) -> None:
sources = {
name: (ADAPTER / name).read_text(encoding="utf-8")
for name in ("blender_worker.py", "scene_recipe.py", "rotary_kiln.py")
}
combined = "\n".join(sources.values())
for marker in (
'"scene.blend"',
'f"preview-{index:02d}.png"',
'"scene.glb"',
"read_factory_settings",
"MAX_EXPANDED_OBJECTS",
"MAX_MESH_POLYGONS",
"primitive_cube_add",
"primitive_cylinder_add",
"primitive_cone_add",
"primitive_uv_sphere_add",
"primitive_torus_add",
"build_rotary_kiln",
"use_renderable=True",
):
self.assertIn(marker, combined)
for forbidden in ("exec(", "eval(", "os.system", "bpy.ops.script"):
self.assertNotIn(forbidden, combined)
if __name__ == "__main__":
unittest.main()