423 lines
16 KiB
Python
423 lines
16 KiB
Python
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()
|