225 lines
9.3 KiB
Python
225 lines
9.3 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 unittest import mock
|
|
from uuid import uuid4
|
|
|
|
from core.software_contracts import SoftwareContractError, get_contract
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ADAPTER = (
|
|
ROOT / "windows-node" / "adapters" / "ansys.mechanical.static_structural@v1"
|
|
)
|
|
WORKER_PATH = ADAPTER / "worker.py"
|
|
|
|
|
|
def _load_worker():
|
|
spec = importlib.util.spec_from_file_location("zcbot_ansys_worker", WORKER_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 _request() -> dict:
|
|
return {
|
|
"schema_version": 1,
|
|
"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", "named_selection": "fixed_face"}
|
|
],
|
|
"loads": [
|
|
{
|
|
"type": "force",
|
|
"named_selection": "load_face",
|
|
"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_contract_is_not_enrolled_by_default_and_normalizes_static_job(self) -> None:
|
|
contract = get_contract("ansys.mechanical.static_structural@v1")
|
|
normalized, digest = contract.normalize_request(_request())
|
|
self.assertFalse(contract.default_enrollment)
|
|
self.assertEqual(contract.feature(normalized), "static_structural")
|
|
self.assertEqual(contract.required_adapter_version(normalized), "0.2.0")
|
|
self.assertEqual(len(digest), 64)
|
|
self.assertEqual(
|
|
set(contract.expected_outputs(normalized)),
|
|
{
|
|
"project",
|
|
"stress_image",
|
|
"summary",
|
|
"result_table",
|
|
"analysis_spec",
|
|
"provenance",
|
|
},
|
|
)
|
|
|
|
def test_contract_rejects_code_paths_and_unsupported_analysis(self) -> None:
|
|
contract = get_contract("ansys.mechanical.static_structural@v1")
|
|
for key, value in (
|
|
("script", "print('unsafe')"),
|
|
("apdl", "/SOLU"),
|
|
("path", r"C:\\private\\model.step"),
|
|
):
|
|
request = _request()
|
|
request[key] = value
|
|
with self.assertRaises(SoftwareContractError):
|
|
contract.normalize_request(request)
|
|
request = _request()
|
|
request["operation"]["analysis"]["type"] = "modal"
|
|
with self.assertRaises(SoftwareContractError):
|
|
contract.normalize_request(request)
|
|
|
|
|
|
class AnsysWorkerTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.worker = _load_worker()
|
|
|
|
def test_worker_semantics_reject_zero_force_and_reused_target(self) -> None:
|
|
request = _request()
|
|
request["operation"]["analysis"]["loads"][0]["components"] = [0, 0, 0]
|
|
with self.assertRaisesRegex(ValueError, "FORCE_VECTOR_MUST_BE_NONZERO"):
|
|
self.worker._validate_semantics(request)
|
|
|
|
request = _request()
|
|
request["operation"]["analysis"]["loads"][0]["named_selection"] = "FIXED_FACE"
|
|
with self.assertRaisesRegex(ValueError, "NAMED_SELECTION_TARGETS_MUST_BE_UNIQUE"):
|
|
self.worker._validate_semantics(request)
|
|
|
|
def test_probe_is_json_and_does_not_claim_ready_off_windows(self) -> None:
|
|
completed = subprocess.run(
|
|
[sys.executable, str(WORKER_PATH), "--probe"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
value = json.loads(completed.stdout)
|
|
self.assertEqual(value["adapter_version"], "0.2.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:
|
|
source = WORKER_PATH.read_text(encoding="utf-8")
|
|
self.assertIn('EXECUTION_GATE = "ZCBOT_ANSYS_242_VALIDATED"', source)
|
|
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
|
|
|
|
class Deformation:
|
|
Minimum = Quantity(0.0, "mm")
|
|
Maximum = Quantity(0.125, "mm")
|
|
|
|
class Reaction:
|
|
XAxis = Quantity(1.0, "N")
|
|
YAxis = Quantity(2.0, "N")
|
|
ZAxis = Quantity(3.0, "N")
|
|
Total = Quantity(3.741657, "N")
|
|
|
|
deformation = self.worker._result_rows("total_deformation", "all", Deformation())
|
|
reaction = self.worker._result_rows("reaction_force", "fixed_face", Reaction())
|
|
self.assertEqual([item["metric"] for item in deformation], ["minimum", "maximum"])
|
|
self.assertEqual(deformation[1]["value"], 0.125)
|
|
self.assertEqual(deformation[1]["unit"], "mm")
|
|
self.assertEqual([item["metric"] for item in reaction], ["x", "y", "z", "total"])
|
|
self.assertTrue(all(item["scope"] == "fixed_face" for item in reaction))
|
|
|
|
def test_named_selection_resolution_is_case_insensitive_and_rejects_missing(self) -> None:
|
|
class Selection:
|
|
def __init__(self, name: str) -> None:
|
|
self.Name = name
|
|
|
|
model = mock.Mock()
|
|
model.NamedSelections.Children = [Selection("fixed_face"), Selection("LOAD_FACE")]
|
|
resolved = self.worker._named_selections(model, ["FIXED_FACE", "load_face"])
|
|
self.assertEqual(resolved["FIXED_FACE"].Name, "fixed_face")
|
|
self.assertEqual(resolved["load_face"].Name, "LOAD_FACE")
|
|
with self.assertRaisesRegex(RuntimeError, "NAMED_SELECTION_NOT_FOUND:missing"):
|
|
self.worker._named_selections(model, ["missing"])
|
|
|
|
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": _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.worker, "run", return_value=[]),
|
|
mock.patch.object(sys, "argv", [str(WORKER_PATH), "--acceptance", str(job_dir)]),
|
|
):
|
|
os.environ.pop(self.worker.EXECUTION_GATE, None)
|
|
self.assertEqual(self.worker.main(), 0)
|
|
terminal = json.loads((job_dir / "terminal.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(terminal["status"], "succeeded")
|
|
self.assertNotIn(self.worker.EXECUTION_GATE, os.environ)
|
|
|
|
def test_worker_uses_v242_embedded_app_and_deterministic_disposal(self) -> None:
|
|
source = WORKER_PATH.read_text(encoding="utf-8")
|
|
self.assertIn("with App(version=ANSYS_REVISION, private_appdata=True) as app:", source)
|
|
self.assertIn("preferences.ProcessNamedSelections = True", source)
|
|
self.assertIn(
|
|
'api["Ansys"].Mechanical.DataModel.Enums.GeometryImportPreference.Format.Automatic',
|
|
source,
|
|
)
|
|
self.assertNotIn('api["GeometryImportPreference"]', source)
|
|
self.assertIn("solution.Solve(True)", source)
|
|
self.assertIn('Path(str(analysis.WorkingDir)) / "solve.out"', source)
|
|
|
|
def test_acceptance_uses_bundled_nist_geometry_and_creates_named_selections(self) -> None:
|
|
fixture = WORKER_PATH.with_name("acceptance-coupon.step")
|
|
acceptance = WORKER_PATH.with_name("acceptance.py").read_text(encoding="utf-8")
|
|
source = WORKER_PATH.read_text(encoding="utf-8")
|
|
self.assertTrue(fixture.is_file())
|
|
self.assertIn("NIST MBE PMI Validation", fixture.read_text(encoding="utf-8"))
|
|
self.assertIn('BUILTIN_GEOMETRY = Path(__file__).with_name("acceptance-coupon.step")', acceptance)
|
|
self.assertIn("nargs=\"?\", default=BUILTIN_GEOMETRY", acceptance)
|
|
self.assertIn("_create_acceptance_named_selections", source)
|
|
self.assertIn('("fixed_face", -1.5)', source)
|
|
self.assertIn('("load_face", 0.425)', source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|