from __future__ import annotations import importlib.util import json import math import tempfile import unittest from pathlib import Path from uuid import uuid4 from core.software_contracts import SoftwareContractError, get_contract ROOT = Path(__file__).resolve().parents[1] WORKER_PATH = ( ROOT / "windows-node" / "adapters" / "cad.geometry.prepare@v1" / "worker.py" ) SPEC = importlib.util.spec_from_file_location("zcbot_cad_worker", WORKER_PATH) assert SPEC and SPEC.loader worker = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(worker) HAS_CADQUERY = importlib.util.find_spec("cadquery") is not None def _request(feature: str, definition: dict, *, regions: list[dict] | None = None) -> dict: prepare = { "type": feature, "title": f"CAD {feature}", "canonical_unit": "mm", "regions": regions or [], } prepare.update(definition) return { "schema_version": 1, "inputs": [], "operation": {"prepare": prepare}, "outputs": [], } def _primitive(shape: str, **parameters) -> dict: return _request("primitive", {"primitive": {"shape": shape, **parameters}}) class CadContractTests(unittest.TestCase): def test_contract_exposes_fixed_features_outputs_and_quotas(self) -> None: contract = get_contract("cad.geometry.prepare@v1") self.assertTrue(contract.ordinarily_visible) self.assertEqual(set(contract.features), {"inspect", "primitive", "recipe"}) self.assertEqual(contract.input_policy["suffixes"], [".step", ".stp"]) request = _primitive("box", length=10, width=20, height=30) normalized, digest = contract.normalize_request(request) self.assertEqual(contract.feature(normalized), "primitive") self.assertEqual(len(digest), 64) self.assertEqual( set(contract.expected_outputs(normalized)), {"geometry", "geometry_recipe", "geometry_manifest", "geometry_preview", "provenance"}, ) geometry_output = contract.outputs["geometry"] self.assertEqual(geometry_output.filename, "geometry.step") ansys = get_contract("ansys.mechanical.static_structural@v2") self.assertIn(Path(geometry_output.filename).suffix, ansys.input_policy["suffixes"]) def test_contract_rejects_scripts_paths_nonfinite_and_missing_units(self) -> None: contract = get_contract("cad.geometry.prepare@v1") for key, value in (("script", "anything"), ("path", "C:/secret.step")): request = _primitive("box", length=10, width=20, height=30) request["operation"]["prepare"][key] = value with self.assertRaises(SoftwareContractError): contract.normalize_request(request) request = _primitive("box", length=math.inf, width=20, height=30) with self.assertRaises(SoftwareContractError): contract.normalize_request(request) request = _primitive("box", length=math.nan, width=20, height=30) with self.assertRaisesRegex(SoftwareContractError, "non-finite"): contract.normalize_request(request) request = _primitive("box", length=10, width=20, height=30) del request["operation"]["prepare"]["canonical_unit"] with self.assertRaises(SoftwareContractError): contract.normalize_request(request) for primitive in ( {"shape": "box", "length": 10, "width": 20}, {"shape": "box", "length": 10, "width": 20, "height": 30, "radius": 2}, ): with self.assertRaises(SoftwareContractError): contract.normalize_request(_request("primitive", {"primitive": primitive})) def test_ansys_and_cad_planar_selector_contracts_stay_consistent(self) -> None: cad = get_contract("cad.geometry.prepare@v1").request_schema["$defs"] ansys = get_contract("ansys.mechanical.static_structural@v2").request_schema["$defs"] def resolved(value: object, definitions: dict) -> object: if isinstance(value, list): return [resolved(item, definitions) for item in value] if not isinstance(value, dict): return value if set(value) == {"$ref"}: return resolved(definitions[value["$ref"].split("/")[-1]], definitions) return { key: resolved(item, definitions) for key, item in value.items() if key != "description" } for name in ("extreme_face", "planar_faces"): ansys_name = f"{name}_scope" self.assertEqual( resolved(cad[name], cad), resolved(ansys[ansys_name], ansys), ) def test_ansys_inspect_is_hidden_diagnostic_with_replacement(self) -> None: contract = get_contract("ansys.geometry.inspect@v1") self.assertFalse(contract.ordinarily_visible) self.assertEqual(contract.lifecycle.status, "deprecated") self.assertEqual(contract.lifecycle.visibility, "diagnostic_only") self.assertEqual(contract.lifecycle.replacement, "cad.geometry.prepare@v1") @unittest.skipUnless(HAS_CADQUERY, "fixed CAD runtime is not installed in project test environment") class CadKernelTests(unittest.TestCase): def test_all_required_primitives_have_numeric_oracles(self) -> None: cases = [ ({"shape": "box", "length": 2, "width": 3, "height": 5}, 30.0), ({"shape": "cylinder", "radius": 2, "height": 5}, 20 * math.pi), ({"shape": "tube", "radius": 3, "inner_radius": 2, "height": 5}, 25 * math.pi), ({"shape": "cone", "radius": 3, "height": 4}, 12 * math.pi), ({"shape": "frustum", "radius1": 3, "radius2": 1, "height": 4}, 52 * math.pi / 3), ({"shape": "sphere", "radius": 3}, 36 * math.pi), ({"shape": "wedge", "x_min": 0, "x_max": 4, "y_min": 0, "y_max": 3, "z_min": 0, "z_max": 2}, 12.0), ({"shape": "torus", "major_radius": 5, "minor_radius": 2}, 40 * math.pi**2), ] for spec, expected in cases: with self.subTest(spec["shape"]): shape = worker._primitive(spec) self.assertAlmostEqual(shape.Volume(), expected, delta=max(1e-8, expected * 1e-8)) def test_recipe_builds_plate_hole_notch_i_beam_and_multi_body(self) -> None: plate = { "profiles": [], "steps": [ {"id": "plate", "op": "box", "primitive": {"shape": "box", "length": 100, "width": 60, "height": 10}}, {"id": "hole", "op": "cylinder", "primitive": {"shape": "cylinder", "radius": 10, "height": 10}}, {"id": "centered_hole", "op": "translate", "target": "hole", "vector": [50, 30, 0]}, {"id": "perforated", "op": "cut", "target": "plate", "tools": ["centered_hole"]}, {"id": "notch", "op": "box", "primitive": {"shape": "box", "length": 20, "width": 20, "height": 10}}, {"id": "placed_notch", "op": "translate", "target": "notch", "vector": [80, 40, 0]}, {"id": "result", "op": "cut", "target": "perforated", "tools": ["placed_notch"]}, ], "result": "result", } shape = worker._recipe(plate, None, "mm") self.assertAlmostEqual( shape.Volume(), 100 * 60 * 10 - math.pi * 10**2 * 10 - 20 * 20 * 10, delta=1e-6, ) i_beam = { "profiles": [{"id": "i", "points": [[0, 0, 0], [60, 0, 0], [60, 10, 0], [35, 10, 0], [35, 90, 0], [60, 90, 0], [60, 100, 0], [0, 100, 0], [0, 90, 0], [25, 90, 0], [25, 10, 0], [0, 10, 0]]}], "steps": [{"id": "beam", "op": "extrude", "profile": "i", "vector": [0, 0, 200]}], "result": "beam", } beam = worker._recipe(i_beam, None, "mm") self.assertAlmostEqual(beam.Volume(), 2000 * 200, delta=1e-6) multi = { "profiles": [], "steps": [ {"id": "coupon", "op": "box", "primitive": {"shape": "box", "length": 10, "width": 10, "height": 20}}, {"id": "top", "op": "translate", "target": "coupon", "vector": [0, 0, 30]}, {"id": "assembly", "op": "combine", "targets": ["coupon", "top"]}, ], "result": "assembly", } assembly = worker._recipe(multi, None, "mm") self.assertEqual(len(assembly.Solids()), 2) def test_regions_manifest_and_step_roundtrip_are_deterministic(self) -> None: shape = worker._primitive({"shape": "box", "length": 10, "width": 20, "height": 30}) regions = [{"name": "fixed", "topology": "face", "selector": {"type": "extreme_face", "axis": "z", "side": "min", "expected_count": 1}}] resolved = worker._regions(shape, regions) prepare = _primitive("box", length=10, width=20, height=30)["operation"]["prepare"] first = worker._manifest(shape, prepare, resolved, None) second = worker._manifest(shape, prepare, resolved, None) self.assertEqual(first, second) self.assertEqual(first["topology"]["solids"], 1) self.assertEqual(first["regions"][0]["count"], 1) self.assertEqual(first["bounding_box"]["size"], [10.0, 20.0, 30.0]) self.assertAlmostEqual(first["area"], 2200.0, delta=1e-8) self.assertEqual(first["center_of_mass"], [5.0, 10.0, 15.0]) with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "geometry.step" worker._export_step(shape, path, "mm") worker._require_step_unit(path) import cadquery as cq reopened = cq.importers.importStep(str(path), unit="MM").val() self.assertAlmostEqual(reopened.Volume(), 6000.0, delta=1e-6) def test_advanced_recipe_operations_fail_loudly_or_produce_valid_solids(self) -> None: cases = { "revolve": { "profiles": [{"id": "section", "points": [[2, 0, 0], [3, 0, 0], [3, 0, 5], [2, 0, 5]]}], "steps": [{"id": "result", "op": "revolve", "profile": "section", "axis_start": [0, 0, 0], "axis_end": [0, 0, 1], "angle_degrees": 360}], "result": "result", }, "sweep": { "profiles": [{"id": "section", "points": [[0, -1, -1], [0, 1, -1], [0, 1, 1], [0, -1, 1]]}], "steps": [{"id": "result", "op": "sweep", "profile": "section", "path": [[0, 0, 0], [10, 0, 0]]}], "result": "result", }, "loft": { "profiles": [ {"id": "a", "points": [[-2, -2, 0], [2, -2, 0], [2, 2, 0], [-2, 2, 0]]}, {"id": "b", "points": [[-1, -1, 10], [1, -1, 10], [1, 1, 10], [-1, 1, 10]]}, ], "steps": [{"id": "result", "op": "loft", "profiles": ["a", "b"]}], "result": "result", }, "fillet": { "profiles": [], "steps": [ {"id": "base", "op": "box", "primitive": {"shape": "box", "length": 10, "width": 10, "height": 10}}, {"id": "result", "op": "fillet", "target": "base", "selector": {"type": "curve_edges", "curve": "line", "expected_count": 12}, "radius": 1}, ], "result": "result", }, "chamfer": { "profiles": [], "steps": [ {"id": "base", "op": "box", "primitive": {"shape": "box", "length": 10, "width": 10, "height": 10}}, {"id": "result", "op": "chamfer", "target": "base", "selector": {"type": "curve_edges", "curve": "line", "expected_count": 12}, "distance": 0.5}, ], "result": "result", }, "shell": { "profiles": [], "steps": [ {"id": "base", "op": "box", "primitive": {"shape": "box", "length": 10, "width": 10, "height": 10}}, {"id": "result", "op": "shell", "target": "base", "selector": {"type": "extreme_face", "axis": "z", "side": "max", "expected_count": 1}, "thickness": 1}, ], "result": "result", }, "mirror": { "profiles": [], "steps": [ {"id": "base", "op": "box", "primitive": {"shape": "box", "length": 2, "width": 3, "height": 4}}, {"id": "result", "op": "mirror", "target": "base", "plane": "YZ"}, ], "result": "result", }, "linear_pattern": { "profiles": [], "steps": [ {"id": "base", "op": "box", "primitive": {"shape": "box", "length": 2, "width": 2, "height": 2}}, {"id": "result", "op": "linear_pattern", "target": "base", "vector": [1, 0, 0], "count": 3, "spacing": 4}, ], "result": "result", }, } for name, recipe in cases.items(): with self.subTest(name): shape = worker._recipe(recipe, None, "mm") self.assertTrue(shape.isValid()) self.assertGreater(shape.Volume(), 0) invalid = cases["fillet"] invalid = json.loads(json.dumps(invalid)) invalid["steps"][1]["selector"]["expected_count"] = 11 with self.assertRaisesRegex(RuntimeError, "GEOMETRY_SCOPE_COUNT_MISMATCH"): worker._recipe(invalid, None, "mm") def test_flange_circular_holes_and_circular_pattern(self) -> None: recipe = { "profiles": [], "steps": [ {"id": "flange", "op": "cylinder", "primitive": {"shape": "cylinder", "radius": 20, "height": 5}}, {"id": "hole", "op": "cylinder", "primitive": {"shape": "cylinder", "radius": 2, "height": 5}}, {"id": "offset_hole", "op": "translate", "target": "hole", "vector": [12, 0, 0]}, {"id": "holes", "op": "circular_pattern", "target": "offset_hole", "axis_start": [0, 0, 0], "axis_end": [0, 0, 1], "count": 6, "angle_degrees": 360}, {"id": "result", "op": "cut", "target": "flange", "tools": ["holes"]}, ], "result": "result", } shape = worker._recipe(recipe, None, "mm") expected = math.pi * 20**2 * 5 - 6 * math.pi * 2**2 * 5 self.assertAlmostEqual(shape.Volume(), expected, delta=expected * 1e-8) def test_full_job_emits_fixed_package_and_preview(self) -> None: request = _request( "primitive", {"primitive": {"shape": "cylinder", "radius": 4, "height": 12}}, regions=[{"name": "ends", "topology": "face", "selector": {"type": "surface_faces", "surface": "plane", "expected_count": 2}}], ) get_contract("cad.geometry.prepare@v1").normalize_request(request) def execute() -> dict: with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "request").mkdir() (root / "request" / "request.json").write_text( json.dumps( { "job_id": str(uuid4()), "lease_id": str(uuid4()), "request_digest": "a" * 64, "request": request, } ), encoding="utf-8", ) artifacts = worker.run(root) self.assertEqual( {item["filename"] for item in artifacts}, set(worker.OUTPUT_MEDIA) ) self.assertGreater( (root / "output" / "geometry-preview.png").stat().st_size, 1000, ) return json.loads( (root / "output" / "geometry-manifest.json").read_text( encoding="utf-8" ) ) first = execute() second = execute() self.assertEqual(first, second) self.assertAlmostEqual(first["volume"], math.pi * 4**2 * 12, delta=1e-6) def test_external_step_inspect_and_recipe_import(self) -> None: import cadquery as cq source = cq.Solid.makeBox(7, 11, 13) with tempfile.TemporaryDirectory() as directory: root = Path(directory) input_dir = root / "input" / "geometry" request_dir = root / "request" input_dir.mkdir(parents=True) request_dir.mkdir() source_path = input_dir / "external.step" worker._export_step(source, source_path, "mm") for feature, definition in ( ("inspect", {"geometry": {"input": "geometry"}}), ( "recipe", { "recipe": { "profiles": [], "steps": [ {"id": "imported", "op": "import_step", "input": "geometry"}, {"id": "moved", "op": "translate", "target": "imported", "vector": [20, 0, 0]}, {"id": "result", "op": "combine", "targets": ["imported", "moved"]}, ], "result": "result", } }, ), ): request = _request(feature, definition) request["inputs"] = [ {"key": "geometry", "artifact_id": str(uuid4())} ] (request_dir / "request.json").write_text( json.dumps( { "job_id": str(uuid4()), "lease_id": str(uuid4()), "request_digest": "b" * 64, "request": request, } ), encoding="utf-8", ) artifacts = worker.run(root) self.assertEqual(len(artifacts), 5) manifest = json.loads( (root / "output" / "geometry-manifest.json").read_text( encoding="utf-8" ) ) expected_volume = 7 * 11 * 13 * (2 if feature == "recipe" else 1) self.assertAlmostEqual( manifest["volume"], expected_volume, delta=1e-6 ) if __name__ == "__main__": unittest.main()