Monkey patch environment variable.

src/ansys/fluent/core/meshing/meshing_workflow.py

@@ -31,7 +31,6 @@
 from ansys.fluent.core._types import PathType
 from ansys.fluent.core.services.datamodel_se import PyMenuGeneric
 from ansys.fluent.core.utils.fluent_version import FluentVersion
-from ansys.fluent.core.workflow_new import Workflow
 
 name_to_identifier_map = {
     "Watertight Geometry": "EnableCleanCAD",
@@ -41,7 +40,18 @@
 }
 
 
-class MeshingWorkflow(Workflow):
+def resolve_workflow_base():
+    if os.getenv("USE_SERVER_MW") == "1":
+        from ansys.fluent.core.workflow_new import Workflow
+    else:
+        from ansys.fluent.core.workflow import Workflow
+    return Workflow
+
+
+WorkflowBase = resolve_workflow_base()
+
+
+class MeshingWorkflow(WorkflowBase):
     """Provides meshing specialization of the workflow wrapper that extends the core
     functionality in an object-oriented manner."""
 
@@ -77,23 +87,25 @@ def __init__(
         self._meshing = meshing
         self._name = name
         self._identifier = identifier
-        # self._unsubscribe_root_affected_callback()
+        if os.getenv("USE_SERVER_MW") != "1":
+            self._unsubscribe_root_affected_callback()
         if initialize:
             self._new_workflow(name=self._name)
         else:
             self._activate_dynamic_interface(dynamic_interface=True)
         self._initialized = True
 
-    # def __getattribute__(self, item: str):
-    #     if (
-    #         not item.startswith("_")
-    #         and super().__getattribute__("_initialized")
-    #         and not getattr(self._meshing.GlobalSettings, self._identifier)()
-    #     ):
-    #         raise RuntimeError(
-    #             f"'{self._name}' objects are inaccessible from other workflows."
-    #         )
-    #     return super().__getattribute__(item)
+    def __getattribute__(self, item: str):
+        if os.getenv("USE_SERVER_MW") != "1":
+            if (
+                not item.startswith("_")
+                and super().__getattribute__("_initialized")
+                and not getattr(self._meshing.GlobalSettings, self._identifier)()
+            ):
+                raise RuntimeError(
+                    f"'{self._name}' objects are inaccessible from other workflows."
+                )
+        return super().__getattribute__(item)
 
 
 class WatertightMeshingWorkflow(MeshingWorkflow):
@@ -267,7 +279,7 @@ class WorkflowMode(Enum):
     TOPOLOGY_BASED_MESHING_MODE = TopologyBasedMeshingWorkflow
 
 
-class LoadWorkflow(Workflow):
+class LoadWorkflow(WorkflowBase):
     """Provides a specialization of the workflow wrapper for a loaded workflow."""
 
     def __init__(
@@ -298,7 +310,7 @@ def __init__(
         self._load_workflow(file_path=os.fspath(file_path))
 
 
-class CreateWorkflow(Workflow):
+class CreateWorkflow(WorkflowBase):
     """Provides a specialization of the workflow wrapper for a newly created
     workflow."""
 

src/ansys/fluent/core/session_base_meshing.py

@@ -138,8 +138,11 @@ def meshing_workflow(self):
 
     def watertight_workflow(self, initialize: bool = True):
         """Datamodel root of workflow."""
+        root_module = (
+            "meshing_workflow" if os.getenv("USE_SERVER_MW") == "1" else "workflow"
+        )
         self._current_workflow = WorkflowMode.WATERTIGHT_MESHING_MODE.value(
-            _make_datamodel_module(self, "meshing_workflow"),
+            _make_datamodel_module(self, root_module),
             self.meshing,
             self.get_fluent_version(),
             initialize,
@@ -148,8 +151,11 @@ def watertight_workflow(self, initialize: bool = True):
 
     def fault_tolerant_workflow(self, initialize: bool = True):
         """Datamodel root of workflow."""
+        root_module = (
+            "meshing_workflow" if os.getenv("USE_SERVER_MW") == "1" else "workflow"
+        )
         self._current_workflow = WorkflowMode.FAULT_TOLERANT_MESHING_MODE.value(
-            _make_datamodel_module(self, "meshing_workflow"),
+            _make_datamodel_module(self, root_module),
             self.meshing,
             self.PartManagement,
             self.PMFileManagement,
@@ -160,8 +166,11 @@ def fault_tolerant_workflow(self, initialize: bool = True):
 
     def two_dimensional_meshing_workflow(self, initialize: bool = True):
         """Data model root of the workflow."""
+        root_module = (
+            "meshing_workflow" if os.getenv("USE_SERVER_MW") == "1" else "workflow"
+        )
         self._current_workflow = WorkflowMode.TWO_DIMENSIONAL_MESHING_MODE.value(
-            _make_datamodel_module(self, "meshing_workflow"),
+            _make_datamodel_module(self, root_module),
             self.meshing,
             self.get_fluent_version(),
             initialize,
@@ -170,8 +179,11 @@ def two_dimensional_meshing_workflow(self, initialize: bool = True):
 
     def topology_based_meshing_workflow(self, initialize: bool = True):
         """Datamodel root of workflow."""
+        root_module = (
+            "meshing_workflow" if os.getenv("USE_SERVER_MW") == "1" else "workflow"
+        )
         self._current_workflow = WorkflowMode.TOPOLOGY_BASED_MESHING_MODE.value(
-            _make_datamodel_module(self, "meshing_workflow"),
+            _make_datamodel_module(self, root_module),
             self.meshing,
             self.get_fluent_version(),
             initialize,
@@ -180,8 +192,11 @@ def topology_based_meshing_workflow(self, initialize: bool = True):
 
     def load_workflow(self, file_path: PathType):
         """Datamodel root of workflow."""
+        root_module = (
+            "meshing_workflow" if os.getenv("USE_SERVER_MW") == "1" else "workflow"
+        )
         self._current_workflow = LoadWorkflow(
-            _make_datamodel_module(self, "meshing_workflow"),
+            _make_datamodel_module(self, root_module),
             self.meshing,
             os.fspath(file_path),
             self.get_fluent_version(),
@@ -190,8 +205,11 @@ def load_workflow(self, file_path: PathType):
 
     def create_workflow(self, initialize: bool = True):
         """Datamodel root of the workflow."""
+        root_module = (
+            "meshing_workflow" if os.getenv("USE_SERVER_MW") == "1" else "workflow"
+        )
         self._current_workflow = CreateWorkflow(
-            _make_datamodel_module(self, "meshing_workflow"),
+            _make_datamodel_module(self, root_module),
             self.meshing,
             self.get_fluent_version(),
             initialize,

tests/conftest.py

@@ -491,3 +491,8 @@ def datamodel_api_version_all(request, monkeypatch: pytest.MonkeyPatch) -> None:
 @pytest.fixture
 def datamodel_api_version_new(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("REMOTING_NEW_DM_API", "1")
+
+
+@pytest.fixture
+def use_server_meshing_workflow(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("USE_SERVER_MW", "1")

tests/test_server_meshing_workflow.py

@@ -893,7 +893,9 @@ def test_insert_delete_and_rename_task(new_meshing_session):
 @pytest.mark.nightly
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_new_watertight_workflow_enhanced_meshing(new_meshing_session_wo_exit):
+def test_new_watertight_workflow_enhanced_meshing(
+    new_meshing_session_wo_exit, use_server_meshing_workflow
+):
     # Import geometry
     import_file_name = examples.download_file(
         "mixing_elbow.pmdb", "pyfluent/mixing_elbow"
@@ -959,7 +961,9 @@ def test_new_watertight_workflow_enhanced_meshing(new_meshing_session_wo_exit):
 @pytest.mark.nightly
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_new_fault_tolerant_workflow_enhanced_meshing(new_meshing_session_wo_exit):
+def test_new_fault_tolerant_workflow_enhanced_meshing(
+    new_meshing_session_wo_exit, use_server_meshing_workflow
+):
     meshing = new_meshing_session_wo_exit
 
     # Import CAD and part management
@@ -1222,7 +1226,9 @@ def test_new_fault_tolerant_workflow_enhanced_meshing(new_meshing_session_wo_exi
 @pytest.mark.nightly
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_new_2d_meshing_workflow_enhanced_meshing(new_meshing_session_wo_exit):
+def test_new_2d_meshing_workflow_enhanced_meshing(
+    new_meshing_session_wo_exit, use_server_meshing_workflow
+):
     # Import geometry
     import_file_name = examples.download_file("NACA0012.fmd", "pyfluent/airfoils")
     two_dim_mesh = new_meshing_session_wo_exit.two_dimensional_meshing()
@@ -1325,7 +1331,9 @@ def test_new_2d_meshing_workflow_enhanced_meshing(new_meshing_session_wo_exit):
 
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_workflow_and_data_model_methods_new_meshing_workflow(new_meshing_session):
+def test_workflow_and_data_model_methods_new_meshing_workflow(
+    new_meshing_session, use_server_meshing_workflow
+):
     meshing = new_meshing_session
     watertight = meshing.watertight()
     _next_possible_tasks = [
@@ -1347,7 +1355,7 @@ def test_workflow_and_data_model_methods_new_meshing_workflow(new_meshing_sessio
 
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_duplicate_tasks(new_meshing_session):
+def test_duplicate_tasks(new_meshing_session, use_server_meshing_workflow):
     meshing = new_meshing_session
     watertight = meshing.watertight()
 
@@ -1386,7 +1394,9 @@ def test_duplicate_tasks(new_meshing_session):
 
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_watertight_workflow(mixing_elbow_geometry_filename, new_meshing_session):
+def test_watertight_workflow(
+    mixing_elbow_geometry_filename, new_meshing_session, use_server_meshing_workflow
+):
     watertight = new_meshing_session.watertight()
     watertight.import_geometry.file_name = mixing_elbow_geometry_filename
     watertight.import_geometry()
@@ -1402,7 +1412,7 @@ def test_watertight_workflow(mixing_elbow_geometry_filename, new_meshing_session
 
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_delete_interface(new_meshing_session):
+def test_delete_interface(new_meshing_session, use_server_meshing_workflow):
     watertight = new_meshing_session.watertight()
 
     watertight.import_geometry.insertable_tasks.import_body_of_influence_geometry.insert()
@@ -1441,7 +1451,7 @@ def test_delete_interface(new_meshing_session):
 
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_ordering_of_tasks(new_meshing_session):
+def test_ordering_of_tasks(new_meshing_session, use_server_meshing_workflow):
     watertight = new_meshing_session.watertight()
     assert len(watertight.children()) == 7
     _watertight_tasks = [
@@ -1495,7 +1505,7 @@ def test_ordering_of_tasks(new_meshing_session):
 
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_workflow_type_checking(new_meshing_session):
+def test_workflow_type_checking(new_meshing_session, use_server_meshing_workflow):
     meshing = new_meshing_session
     watertight = meshing.watertight()
 
@@ -1514,7 +1524,7 @@ def test_workflow_type_checking(new_meshing_session):
 
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_workflow_traversal(new_meshing_session):
+def test_workflow_traversal(new_meshing_session, use_server_meshing_workflow):
     meshing = new_meshing_session
     watertight = meshing.watertight()
 
@@ -1568,7 +1578,9 @@ def test_workflow_traversal(new_meshing_session):
 
 @pytest.mark.codegen_required
 @pytest.mark.fluent_version(">=26.1")
-def test_new_watertight_workflow_using_traversal(new_meshing_session_wo_exit):
+def test_new_watertight_workflow_using_traversal(
+    new_meshing_session_wo_exit, use_server_meshing_workflow
+):
     # Import geometry
     import_file_name = examples.download_file(
         "mixing_elbow.pmdb", "pyfluent/mixing_elbow"