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