zcbot/tests/test_blender_adapter.py

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@v2"
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": 2,
        "inputs": [],
        "operation": {
            "scene": {
                "type": "rotary_kiln",
                "title": "水泥回转窑",
                "kiln": {
                    "length_m": 60,
                    "outer_diameter_m": 4,
                    "shell_thickness_m": 0.04,
                    "lining_thickness_m": 0.2,
                    "slope_percent": 3.5,
                    "support_stations": 3,
                    "rotational_speed_rpm": 3,
                },
                "components": {
                    "drive_system": True,
                    "burner": True,
                    "hoods": True,
                    "inspection_platform": True,
                    "material_bed": True,
                },
                "appearance": {
                    "shell_color": "#707780",
                    "metal_color": "#303840",
                    "lining_color": "#B56A3B",
                    "material_color": "#E35D20",
                    "background_color": "#20252B",
                },
                "render": {
                    "width": 1280,
                    "height": 720,
                    "transparent_background": False,
                    "views": ["isometric", "longitudinal_cutaway"],
                },
            }
        },
        "outputs": [
            {"key": "preview_cutaway", "type": "figure", "format": "png"},
            {"key": "scene_glb", "type": "model", "format": "glb"},
        ],
    }


class BlenderContractTests(unittest.TestCase):
    def test_rotary_kiln_contract_is_workspace_backed(self) -> None:
        contract = get_contract("blender.scene.author@v2")
        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,
            (
                "preview_isometric",
                "preview_side",
                "preview_kiln_head",
                "preview_kiln_tail",
                "preview_cutaway",
                "preview_cross_section",
            ),
        )
        self.assertEqual(contract.required_adapter_version(normalized), "0.2.0")
        self.assertEqual(
            set(contract.expected_outputs(normalized)),
            {
                "project",
                "preview_isometric",
                "preview_cutaway",
                "scene_manifest",
                "scene_glb",
                "provenance",
            },
        )

    def test_contract_rejects_arbitrary_execution_and_unsafe_scale(self) -> None:
        contract = get_contract("blender.scene.author@v2")
        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_geometry = request()
        invalid_geometry["operation"]["scene"]["kiln"]["support_stations"] = 1  # type: ignore[index]
        with self.assertRaises(SoftwareContractError):
            contract.normalize_request(invalid_geometry)


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

    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_contains_rotary_kiln_outputs_without_dynamic_code(self) -> None:
        source = (ADAPTER / "blender_worker.py").read_text(encoding="utf-8")
        for marker in (
            '"KilnShell"',
            '"GirthGear"',
            '"SupportRoller_',
            '"scene.blend"',
            '"preview-isometric.png"',
            '"CutawayShell"',
            '"CutawayLining"',
            '"longitudinal_cutaway"',
            '"scene.glb"',
            "use_renderable=True",
            "bpy.ops.wm.open_mainfile",
        ):
            self.assertIn(marker, source)
        self.assertNotIn("exec(", source)
        self.assertNotIn("eval(", source)


if __name__ == "__main__":
    unittest.main()