from __future__ import annotations import importlib.util import json import os import subprocess import sys import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest import mock from uuid import uuid4 from core.software_contracts import SoftwareContractError, get_contract ROOT = Path(__file__).resolve().parents[1] ADAPTER_ROOT = ROOT / "windows-node" / "adapters" INSPECT_WORKER = ADAPTER_ROOT / "ansys.geometry.inspect@v1" / "worker.py" STATIC_WORKER = ADAPTER_ROOT / "ansys.mechanical.static_structural@v2" / "worker.py" sys.path.insert(0, str(ADAPTER_ROOT)) import ansys_geometry # noqa: E402 def _load_worker(path: Path, name: str): spec = importlib.util.spec_from_file_location(name, path) if spec is None or spec.loader is None: raise RuntimeError("could not load ANSYS worker") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _inspect_request() -> dict: return { "schema_version": 1, "inputs": [{"key": "geometry", "artifact_id": str(uuid4())}], "operation": { "inspect": { "type": "geometry", "title": "coupon geometry inspection", "geometry": {"input": "geometry"}, } }, "outputs": [], } def _static_request() -> dict: return { "schema_version": 2, "inputs": [{"key": "geometry", "artifact_id": str(uuid4())}], "operation": { "analysis": { "type": "static_structural", "title": "coupon static analysis", "unit_system": "mm_kg_s", "geometry": {"input": "geometry"}, "material": "structural_steel", "mesh": {"global_size": 2.0}, "boundary_conditions": [ { "type": "fixed_support", "name": "fixed_end", "scope": { "type": "extreme_face", "axis": "x", "side": "min", "expected_count": 1, }, } ], "loads": [ { "type": "force", "name": "tip_load", "scope": { "type": "planar_faces", "normal": [1, 0, 0], "offset": 200, "tolerance": 0.01, "expected_count": 1, }, "components": [0, 0, -1000], } ], "results": ["total_deformation", "equivalent_stress"], } }, "outputs": [ {"key": "project", "format": "mechdat"}, {"key": "stress_image", "format": "png"}, ], } class AnsysContractTests(unittest.TestCase): def test_inspect_contract_normalizes_and_always_emits_manifest_preview(self) -> None: contract = get_contract("ansys.geometry.inspect@v1") normalized, digest = contract.normalize_request(_inspect_request()) self.assertFalse(contract.default_enrollment) self.assertEqual(contract.feature(normalized), "geometry") self.assertEqual(contract.required_adapter_version(normalized), "0.3.0") self.assertEqual(len(digest), 64) self.assertEqual( set(contract.expected_outputs(normalized)), {"geometry_manifest", "geometry_preview", "provenance"}, ) def test_static_v2_contract_accepts_declarative_scopes_and_required_preview(self) -> None: contract = get_contract("ansys.mechanical.static_structural@v2") normalized, digest = contract.normalize_request(_static_request()) self.assertEqual(contract.feature(normalized), "static_structural") self.assertEqual(contract.required_adapter_version(normalized), "0.3.0") self.assertEqual(len(digest), 64) self.assertEqual( set(contract.expected_outputs(normalized)), { "project", "stress_image", "summary", "result_table", "selection_preview", "analysis_spec", "provenance", }, ) def test_static_v2_rejects_v1_named_selection_shape(self) -> None: contract = get_contract("ansys.mechanical.static_structural@v2") request = _static_request() request["operation"]["analysis"]["boundary_conditions"][0] = { "type": "fixed_support", "named_selection": "fixed_face", } with self.assertRaises(SoftwareContractError): contract.normalize_request(request) def test_static_v2_rejects_code_paths_and_zero_plane_normal(self) -> None: contract = get_contract("ansys.mechanical.static_structural@v2") request = _static_request() request["script"] = "print('unsafe')" with self.assertRaises(SoftwareContractError): contract.normalize_request(request) request = _static_request() request["operation"]["analysis"]["loads"][0]["scope"]["normal"] = [0, 0, 0] normalized, _ = contract.normalize_request(request) worker = _load_worker(STATIC_WORKER, "zcbot_ansys_static_v2_semantics") with self.assertRaisesRegex(ValueError, "PLANE_NORMAL_MUST_BE_NONZERO"): worker._validate_semantics(normalized) class Face: def __init__( self, face_id: int, centroid: tuple[float, float, float], normal: tuple[float, float, float], surface_type: str = "GeoSurfacePlane", ) -> None: self.Id = face_id self.Centroid = centroid self.Area = 20.0 self.SurfaceType = surface_type self._normal = normal def ParamAtPoint(self, point): return (0.5, 0.5) def NormalAtParam(self, u, v): return self._normal class AnsysGeometryTests(unittest.TestCase): def setUp(self) -> None: self.inventory = { "geometry_unit": "mm", "bounding_box": {"min": [0, 0, 0], "max": [200, 100, 20]}, "faces": [ { "id": 1, "surface_type": "plane", "centroid": [0, 50, 10], "normal": [-1, 0, 0], }, { "id": 2, "surface_type": "plane", "centroid": [200, 50, 10], "normal": [1, 0, 0], }, { "id": 3, "surface_type": "plane", "centroid": [100, 50, 0], "normal": [0, 0, -1], }, ], } def test_extreme_face_only_uses_axis_aligned_planar_faces(self) -> None: ids = ansys_geometry._scope_face_ids( {}, self.inventory, { "type": "extreme_face", "axis": "x", "side": "min", "expected_count": 1, }, "mm", ) self.assertEqual(ids, [1]) def test_planar_faces_uses_plane_equation_and_count_assertion(self) -> None: ids = ansys_geometry._scope_face_ids( {}, self.inventory, { "type": "planar_faces", "normal": [1, 0, 0], "offset": 200, "tolerance": 0.01, "expected_count": 1, }, "mm", ) self.assertEqual(ids, [2]) with self.assertRaisesRegex(RuntimeError, "GEOMETRY_SCOPE_COUNT_MISMATCH"): ansys_geometry._scope_face_ids( {}, self.inventory, { "type": "planar_faces", "normal": [1, 0, 0], "offset": 200, "tolerance": 0.01, "expected_count": 2, }, "mm", ) def test_resolve_scope_creates_named_selection_from_geometry_ids(self) -> None: selection_info = SimpleNamespace(Ids=[]) created = SimpleNamespace(Name="", Location=None) model = SimpleNamespace( NamedSelections=SimpleNamespace(Children=[]), AddNamedSelection=mock.Mock(return_value=created), ) app = SimpleNamespace( Model=model, ExtAPI=SimpleNamespace( SelectionManager=SimpleNamespace( CreateSelectionInfo=mock.Mock(return_value=selection_info) ) ), ) api = {"SelectionTypeEnum": SimpleNamespace(GeometryEntities="geometry")} resolved = ansys_geometry.resolve_scope( app, api, self.inventory, {1: object(), 2: object(), 3: object()}, {"type": "extreme_face", "axis": "x", "side": "max"}, "mm", "tip_load", ) self.assertEqual(selection_info.Ids, [2]) self.assertEqual(created.Name, "tip_load") self.assertIs(created.Location, selection_info) self.assertEqual(resolved["source"], "geometry_query") def test_geometry_inventory_reports_faces_units_bounds_and_named_selections(self) -> None: faces = [ Face(11, (0, 5, 2), (-1, 0, 0)), Face(12, (10, 5, 2), (1, 0, 0)), ] body = SimpleNamespace( Id=7, Name="plate", Volume=400.0, Vertices=[], Faces=faces, ) part = SimpleNamespace(Name="part", Bodies=[body]) assembly = SimpleNamespace(Name="assembly", Parts=[part]) imported = SimpleNamespace( Name="fixture", Location=SimpleNamespace(Ids=[11]), ) app = SimpleNamespace( ExtAPI=SimpleNamespace( DataModel=SimpleNamespace( GeoData=SimpleNamespace(Unit="mm", Assemblies=[assembly]) ) ), Model=SimpleNamespace( NamedSelections=SimpleNamespace(Children=[imported]) ), ) inventory, entities = ansys_geometry.geometry_inventory(app) self.assertEqual(inventory["geometry_unit"], "mm") self.assertEqual(inventory["counts"]["faces"], 2) self.assertEqual(inventory["bounding_box"], {"min": [0, 5, 2], "max": [10, 5, 2]}) self.assertEqual(inventory["named_selections"][0]["entity_ids"], [11]) self.assertEqual(set(entities), {11, 12}) class AnsysWorkerTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.inspect = _load_worker(INSPECT_WORKER, "zcbot_ansys_inspect_worker") cls.static = _load_worker(STATIC_WORKER, "zcbot_ansys_static_v2_worker") def test_static_worker_rejects_zero_force_and_duplicate_names(self) -> None: request = _static_request() request["operation"]["analysis"]["loads"][0]["components"] = [0, 0, 0] with self.assertRaisesRegex(ValueError, "FORCE_VECTOR_MUST_BE_NONZERO"): self.static._validate_semantics(request) request = _static_request() request["operation"]["analysis"]["loads"][0]["name"] = "FIXED_END" with self.assertRaisesRegex(ValueError, "CONDITION_NAMES_MUST_BE_UNIQUE"): self.static._validate_semantics(request) def test_probes_are_json_and_do_not_claim_ready_off_windows(self) -> None: for worker in (INSPECT_WORKER, STATIC_WORKER): completed = subprocess.run( [sys.executable, str(worker), "--probe"], check=True, capture_output=True, text=True, ) value = json.loads(completed.stdout) self.assertEqual(value["adapter_version"], "0.3.0") self.assertEqual(value["software"], "ANSYS Mechanical") if sys.platform != "win32": self.assertEqual(value["health"], "unavailable") def test_execution_is_guarded_until_real_license_acceptance(self) -> None: for worker in (INSPECT_WORKER, STATIC_WORKER): source = worker.read_text(encoding="utf-8") self.assertIn("ANSYS_242_REAL_LICENSE_ACCEPTANCE_REQUIRED", source) for forbidden in ("subprocess", "eval(", "exec(", "os.system"): self.assertNotIn(forbidden, source) def test_result_rows_preserve_values_units_and_reaction_components(self) -> None: class Quantity: def __init__(self, value: float, unit: str) -> None: self.Value = value self.Unit = unit deformation = SimpleNamespace( Minimum=Quantity(0.0, "mm"), Maximum=Quantity(0.125, "mm") ) reaction = SimpleNamespace( XAxis=Quantity(1.0, "N"), YAxis=Quantity(2.0, "N"), ZAxis=Quantity(3.0, "N"), Total=Quantity(3.741657, "N"), ) self.assertEqual( self.static._result_rows("total_deformation", "all", deformation)[1]["value"], 0.125, ) rows = self.static._result_rows("reaction_force", "fixed_end", reaction) self.assertEqual([item["metric"] for item in rows], ["x", "y", "z", "total"]) def test_acceptance_entry_bypasses_gate_without_enabling_scheduled_execution(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": _static_request(), } (job_dir / "request" / "request.json").write_text( json.dumps(record), encoding="utf-8" ) with ( mock.patch.dict(os.environ, {}, clear=False), mock.patch.object(self.static, "run", return_value=[]), mock.patch.object( sys, "argv", [str(STATIC_WORKER), "--acceptance", str(job_dir)] ), ): os.environ.pop(ansys_geometry.EXECUTION_GATE, None) self.assertEqual(self.static.main(), 0) result = json.loads((job_dir / "terminal.json").read_text(encoding="utf-8")) self.assertEqual(result["status"], "succeeded") self.assertNotIn(ansys_geometry.EXECUTION_GATE, os.environ) def test_worker_uses_v242_geodata_and_geometry_entity_selection(self) -> None: common = (ADAPTER_ROOT / "ansys_geometry.py").read_text(encoding="utf-8") static = STATIC_WORKER.read_text(encoding="utf-8") self.assertIn("GeoData", common) self.assertIn("face.ParamAtPoint(centroid)", common) self.assertIn("face.NormalAtParam", common) self.assertIn("SelectionTypeEnum", common) self.assertIn("location.Ids = face_ids", common) self.assertIn("resolved_scopes", static) self.assertIn("selection-preview.png", static) self.assertIn("solution.Solve(True)", static) def test_acceptance_uses_bundled_geometry_and_v2_planar_scopes(self) -> None: adapter = STATIC_WORKER.parent fixture = adapter / "acceptance-coupon.step" acceptance = (adapter / "acceptance.py").read_text(encoding="utf-8") self.assertTrue(fixture.is_file()) self.assertIn("NIST MBE PMI Validation", fixture.read_text(encoding="utf-8")) self.assertIn('"schema_version": 2', acceptance) self.assertIn('"type": "planar_faces"', acceptance) self.assertIn('"offset": -1.5', acceptance) self.assertIn('"offset": 0.425', acceptance) self.assertIn('"selection_preview"', acceptance) if __name__ == "__main__": unittest.main()