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( 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()