fix: add docstring/typehints; add exception in Workflow.get_topology()

Author: Matteo-Baussart-ANSYS

src/ansys/dpf/core/common.py

@@ -433,7 +433,16 @@ def type_to_special_dpf_constructors():
 _derived_class_name_to_type = None
 
 
-def derived_class_name_to_type():
+def derived_class_name_to_type() -> dict[str, type]:
+    """
+    Returns a mapping of derived class names to their corresponding Python classes.
+
+    Returns
+    -------
+    dict[str, type]
+        A dictionary mapping derived class names (str) to their corresponding
+        Python class objects.
+    """
     global _derived_class_name_to_type
     if _derived_class_name_to_type is None:
         from ansys.dpf.core.workflow_topology import WorkflowTopology

src/ansys/dpf/core/workflow.py

@@ -36,7 +36,7 @@
 from typing import Union
 
 from ansys import dpf
-from ansys.dpf.core import dpf_operator, inputs, outputs
+from ansys.dpf.core import dpf_operator, errors, inputs, outputs
 from ansys.dpf.core.check_version import (
     server_meet_version,
     version_requires,
@@ -969,7 +969,13 @@ def get_topology(self):
         Returns
         -------
         workflow_topology : workflow_topology.WorkflowTopology
+
+        Notes
+        -----
+        Available from 10.0 server version.
         """
+        if not self._server.meet_version("10.0"):
+            raise errors.DpfVersionNotSupported("10.0")
         workflow_to_workflow_topology_op = dpf_operator.Operator(
             "workflow_to_workflow_topology", server=self._server
         )

tests/conftest.py

@@ -391,6 +391,8 @@ def decorator(func):
             if version == "5.0"
             else not SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_6_0
             if version == "6.0"
+            else not SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0
+            if version == "10.0"
             else True,
             reason=f"Requires server version greater than or equal to {version}",
             raises=core.errors.DpfVersionNotSupported,

tests/test_any.py

@@ -24,6 +24,7 @@
 
 import conftest
 from ansys.dpf import core as dpf
+from conftest import raises_for_servers_version_under
 
 
 @conftest.raises_for_servers_version_under("7.0")
@@ -134,10 +135,7 @@ def test_cast_workflow_any(server_type):
     assert new_entity.input_names == []
 
 
-@pytest.mark.skipif(
-    not conftest.SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0,
-    reason="any does not support operator below 8.0",
-)
+@raises_for_servers_version_under("10.0")
 def test_cast_operator_any(server_type):
     entity = dpf.Operator(server=server_type, name="U")
     any_dpf = dpf.Any.new_from(entity)

tests/test_operator.py

@@ -45,6 +45,7 @@
     SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_7_0,
     SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_8_0,
 )
+from conftest import raises_for_servers_version_under
 
 # Check for ANSYS installation env var
 HAS_AWP_ROOT212 = os.environ.get("AWP_ROOT212", False) is not False
@@ -1427,10 +1428,7 @@ def test_operator_input_output_streams(server_in_process, simple_bar):
     assert times
 
 
-@pytest.mark.skipif(
-    not conftest.SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0,
-    reason="Operator `workflow_to_workflow_topology` does not exist below 10.0",
-)
+@raises_for_servers_version_under("10.0")
 def test_operator_outputs_derived_class(server_type):
     workflow = dpf.core.Workflow(server=server_type)
 
@@ -1443,10 +1441,7 @@ def test_operator_outputs_derived_class(server_type):
     assert workflow_topology
 
 
-@pytest.mark.skipif(
-    not conftest.SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0,
-    reason="Operator `workflow_to_workflow_topology` does not exist below 10.0",
-)
+@raises_for_servers_version_under("10.0")
 def test_operator_get_output_derived_class(server_type):
     workflow = dpf.core.Workflow(server=server_type)
 

tests/test_workflow.py

@@ -31,6 +31,7 @@
 import conftest
 from ansys import dpf
 from ansys.dpf.core import misc
+from conftest import raises_for_servers_version_under
 
 if misc.module_exists("graphviz"):
     HAS_GRAPHVIZ = True
@@ -1031,10 +1032,7 @@ def test_workflow_input_output_streams(server_in_process, simple_bar):
     assert times
 
 
-@pytest.mark.skipif(
-    not conftest.SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0,
-    reason="Operator `workflow_to_workflow_topology` does not exist below 10.0",
-)
+@raises_for_servers_version_under("10.0")
 def test_workflow_get_output_derived_class(server_type):
     workflow = dpf.core.Workflow(server=server_type)
 

tests/test_workflow_topology.py

@@ -24,7 +24,7 @@
 
 from ansys import dpf
 import ansys.dpf.core.operators as op
-import conftest
+from conftest import raises_for_servers_version_under
 
 
 def workflow_forward(server_type) -> dpf.core.Workflow:
@@ -156,10 +156,7 @@ def expected_workflow_topology(workflow):
     return workflow_topologies[workflow.name]
 
 
-@pytest.mark.skipif(
-    not conftest.SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0,
-    reason="Operator `workflow_to_workflow_topology` does not exist below 10.0",
-)
+@raises_for_servers_version_under("10.0")
 def test_instantiate_workflow_to_workflow_topology_op(server_type):
     workflow_to_workflow_topology_op = dpf.core.Operator(
         "workflow_to_workflow_topology", server=server_type
@@ -168,20 +165,14 @@ def test_instantiate_workflow_to_workflow_topology_op(server_type):
     assert workflow_to_workflow_topology_op
 
 
-@pytest.mark.skipif(
-    not conftest.SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0,
-    reason="Operator `workflow_to_workflow_topology` does not exist below 10.0",
-)
+@raises_for_servers_version_under("10.0")
 def test_workflow_get_topology(workflow):
     workflow_topology = workflow.get_topology()
 
     assert workflow_topology
 
 
-@pytest.mark.skipif(
-    not conftest.SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0,
-    reason="Operator `workflow_to_workflow_topology` does not exist below 10.0",
-)
+@raises_for_servers_version_under("10.0")
 def test_workflow_topology_sizes(workflow, expected_workflow_topology):
     workflow_topology = workflow.get_topology()
 
@@ -195,10 +186,7 @@ def test_workflow_topology_sizes(workflow, expected_workflow_topology):
     assert len(workflow_topology.exposed_outputs) == expected_workflow_topology["exposed_outputs"]
 
 
-@pytest.mark.skipif(
-    not conftest.SERVERS_VERSION_GREATER_THAN_OR_EQUAL_TO_10_0,
-    reason="Operator `workflow_to_workflow_topology` does not exist below 10.0",
-)
+@raises_for_servers_version_under("10.0")
 def test_workflow_topology_str(workflow):
     workflow_topology = workflow.get_topology()