feat: Send mapped option as grpc metadata (#3582)

* feat: Send mapped option as grpc metadata

* test: enable test_on_command_executed_lifetime

src/ansys/fluent/core/launcher/fluent_container.py

@@ -319,16 +319,10 @@ def configure_container_dict(
             container_dict["environment"] = {}
         container_dict["environment"]["FLUENT_NO_AUTOMATIC_TRANSCRIPT"] = "1"
 
-    if (
-        os.getenv("REMOTING_NEW_DM_API") == "1"
-        or os.getenv("REMOTING_MAPPED_NEW_DM_API") == "1"
-    ):
+    if os.getenv("REMOTING_NEW_DM_API") == "1":
         if "environment" not in container_dict:
             container_dict["environment"] = {}
-        if os.getenv("REMOTING_NEW_DM_API") == "1":
-            container_dict["environment"]["REMOTING_NEW_DM_API"] = "1"
-        if os.getenv("REMOTING_MAPPED_NEW_DM_API") == "1":
-            container_dict["environment"]["REMOTING_MAPPED_NEW_DM_API"] = "1"
+        container_dict["environment"]["REMOTING_NEW_DM_API"] = "1"
 
     if pyfluent.LAUNCH_FLUENT_IP or os.getenv("REMOTING_SERVER_ADDRESS"):
         if "environment" not in container_dict:

src/ansys/fluent/core/services/datamodel_se.py

@@ -175,6 +175,8 @@ def __init__(
         self._stub = DataModelGrpcModule.DataModelStub(intercept_channel)
         self._metadata = metadata
         self.file_transfer_service = file_transfer_service
+        if os.getenv("REMOTING_MAPPED_NEW_DM_API") == "1":
+            self._metadata.append(("mapped", "1"))
 
     # TODO: Remove it from the proto interface
     def initialize_datamodel(

tests/test_datamodel_api.py

@@ -51,13 +51,12 @@
 def test_env_var_setting(datamodel_api_version_all, request, new_solver_session):
     solver = new_solver_session
     test_name = request.node.name
-    for var in ["REMOTING_NEW_DM_API", "REMOTING_MAPPED_NEW_DM_API"]:
-        # TODO: It might be possible to check the param value in the fixture
-        # instead of checking the test name here.
-        if test_name.endswith("[old]"):
-            assert solver.scheme_eval.scheme_eval(f'(getenv "{var}")') is None
-        elif test_name.endswith("[new]"):
-            assert solver.scheme_eval.scheme_eval(f'(getenv "{var}")') == "1"
+    # TODO: It might be possible to check the param value in the fixture
+    # instead of checking the test name here.
+    if test_name.endswith("[old]"):
+        assert solver.scheme_eval.scheme_eval('(getenv "REMOTING_NEW_DM_API")') is None
+    elif test_name.endswith("[new]"):
+        assert solver.scheme_eval.scheme_eval('(getenv "REMOTING_NEW_DM_API")') == "1"
 
 
 @pytest.mark.fluent_version(">=25.2")

tests/test_datamodel_service.py

@@ -655,7 +655,6 @@ def test_on_affected_at_type_path_lifetime(new_solver_session):
     assert "/test/affected/A:A1/B-1" not in solver._se_service.subscriptions
 
 
-@pytest.mark.skip(reason="https://github.com/ansys/pyfluent/issues/3625")
 @pytest.mark.fluent_version(">=24.2")
 def test_on_command_executed_lifetime(new_solver_session):
     solver = new_solver_session
@@ -672,8 +671,7 @@ def test_on_command_executed_lifetime(new_solver_session):
     else:
         _ = root.A["A1"].add_on_command_executed(lambda *args: data.append(1))
         root.A["A1"].add_on_command_executed(lambda *args: data.append(2))
-        # TODO: path should be appended to the tag
-        tags = ["/test/command_executed", "/test/command_executed-1"]
+        tags = ["/test/command_executed/A:A1", "/test/command_executed/A:A1-1"]
     gc.collect()
     for tag in tags:
         assert tag in solver._se_service.subscriptions