fix: Raise AttributeError for non-existing meshing objects after switch_to solver. (#3949)

Raise AttributeError for non-existing meshing objects after switch_to
solver.

Earlier instead of raising an Attribute error (general python
behaviour), it was returning None. This has been corrected now.

There is an "is_active" method associated with session objects to check
it's state.

---------

Co-authored-by: pyansys-ci-bot <[email protected]>

Author: prmukherj

doc/changelog.d/3949.fixed.md

@@ -0,0 +1 @@
+Raise AttributeError for non-existing meshing objects after switch_to solver.

src/ansys/fluent/core/session.py

@@ -265,6 +265,10 @@ def __init__(self, _session):
         for obj in filter(None, (self._datamodel_events, self.transcript, self.events)):
             self._fluent_connection.register_finalizer_cb(obj.stop)
 
+    def is_active(self):
+        """Whether the current session is active."""
+        return True if self._fluent_connection else False
+
     @property
     def field_info(self):
         """Provides access to Fluent field information."""
@@ -370,6 +374,7 @@ def exit(self, **kwargs) -> None:
         logger.debug("session.exit() called")
         if self._fluent_connection:
             self._fluent_connection.exit(**kwargs)
+            self._fluent_connection = None
 
     def force_exit(self) -> None:
         """Forces the Fluent session to exit, losing unsaved progress and data."""
@@ -449,6 +454,8 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any):
         self.exit()
 
     def __dir__(self):
+        if self._fluent_connection is None:
+            return ["is_active"]
         dir_list = set(list(self.__dict__.keys()) + dir(type(self))) - {
             "field_data",
             "field_info",

src/ansys/fluent/core/session_meshing.py

@@ -64,7 +64,6 @@ def __init__(
             transcript can be subsequently started and stopped
             using method calls on the ``Session`` object.
         """
-        self._switched = False
         super(Meshing, self).__init__(
             fluent_connection=fluent_connection,
             scheme_eval=scheme_eval,
@@ -89,15 +88,19 @@ def switch_to_solver(self) -> Any:
             scheme_eval=self.scheme_eval,
             file_transfer_service=self._file_transfer_service,
         )
-        self._switched = True
+        self._fluent_connection = None
         return solver_session
 
     def __getattribute__(self, item: str):
-        if item == "_switched":
-            return super(Meshing, self).__getattribute__(item)
-
-        if self._switched and item != "exit":
-            return None
+        if super(Meshing, self).__getattribute__(
+            "_fluent_connection"
+        ) is None and item not in [
+            "is_active",
+            "_fluent_connection",
+        ]:
+            raise AttributeError(
+                f"'{__class__.__name__}' object has no attribute '{item}'"
+            )
 
         return super(Meshing, self).__getattribute__(item)
 

src/ansys/fluent/core/session_solver.py

@@ -366,6 +366,8 @@ def __getattr__(self, attr):
         return getattr(self._settings_api_root, attr)
 
     def __dir__(self):
+        if self._fluent_connection is None:
+            return ["is_active"]
         settings_dir = []
         if self.get_fluent_version() <= FluentVersion.v242:
             self._populate_settings_api_root()

tests/conftest.py

@@ -257,6 +257,13 @@ def create_session(**kwargs):
         return pyfluent.launch_fluent(**kwargs)
 
 
+@pytest.fixture
+def new_meshing_session_wo_exit():
+    meshing = create_session(mode=pyfluent.FluentMode.MESHING)
+    yield meshing
+    # Exit is intentionally avoided here. Please exit from the method using this.
+
+
 @pytest.fixture
 def new_meshing_session():
     meshing = create_session(mode=pyfluent.FluentMode.MESHING)
@@ -277,6 +284,14 @@ def watertight_workflow_session(new_meshing_session):
     return new_meshing_session
 
 
+@pytest.fixture
+def watertight_workflow_session_wo_exit(new_meshing_session_wo_exit):
+    new_meshing_session_wo_exit.workflow.InitializeWorkflow(
+        WorkflowType="Watertight Geometry"
+    )
+    return new_meshing_session_wo_exit
+
+
 @pytest.fixture
 def fault_tolerant_workflow_session(new_meshing_session):
     new_meshing_session.workflow.InitializeWorkflow(
@@ -285,6 +300,14 @@ def fault_tolerant_workflow_session(new_meshing_session):
     return new_meshing_session
 
 
+@pytest.fixture
+def fault_tolerant_workflow_session_wo_exit(new_meshing_session_wo_exit):
+    new_meshing_session_wo_exit.workflow.InitializeWorkflow(
+        WorkflowType="Fault-tolerant Meshing"
+    )
+    return new_meshing_session_wo_exit
+
+
 @pytest.fixture
 def mixing_elbow_watertight_pure_meshing_session(
     new_pure_meshing_session, mixing_elbow_geometry_filename

tests/test_cad_to_post_ftm.py

@@ -49,9 +49,9 @@
 @pytest.mark.nightly
 @pytest.mark.codegen_required
 def test_exhaust_system(
-    fault_tolerant_workflow_session, exhaust_system_geometry_filename
+    fault_tolerant_workflow_session_wo_exit, exhaust_system_geometry_filename
 ):
-    meshing_session = fault_tolerant_workflow_session
+    meshing_session = fault_tolerant_workflow_session_wo_exit
     workflow = meshing_session.workflow
 
     _ = partial(assign_task_arguments, workflow=workflow, check_state=True)
@@ -545,3 +545,5 @@ def test_exhaust_system(
         )
 
         ###############################################################################
+
+    solver_session.exit()

tests/test_cad_to_post_wtm.py

@@ -50,8 +50,10 @@
 
 @pytest.mark.nightly
 @pytest.mark.codegen_required
-def test_mixing_elbow(watertight_workflow_session, mixing_elbow_geometry_filename):
-    meshing_session = watertight_workflow_session
+def test_mixing_elbow(
+    watertight_workflow_session_wo_exit, mixing_elbow_geometry_filename
+):
+    meshing_session = watertight_workflow_session_wo_exit
     workflow = meshing_session.workflow
 
     assign_task_args = partial(
@@ -270,3 +272,4 @@ def test_mixing_elbow(watertight_workflow_session, mixing_elbow_geometry_filenam
         )
 
         ###############################################################################
+    solver_session.exit()

tests/test_fluent_session.py

@@ -231,13 +231,14 @@ def _freeze_fluent(s):
     tmp_thread = threading.Thread(target=_freeze_fluent, args=(session,), daemon=True)
     tmp_thread.start()
     tmp_thread.join(5)
+    fl_connection = session._fluent_connection
     if tmp_thread.is_alive():
         session.exit(timeout=1, timeout_force=True)
         tmp_thread.join()
     else:
         raise Exception("Test should have temporarily frozen Fluent, but did not.")
 
-    assert session._fluent_connection.wait_process_finished(wait=5)
+    assert fl_connection.wait_process_finished(wait=5)
 
 
 @pytest.mark.fluent_version(">=23.1")
@@ -276,16 +277,19 @@ def test_fluent_exit(monkeypatch: pytest.MonkeyPatch):
 
 def test_fluent_exit_wait():
     session1 = pyfluent.launch_fluent()
+    fl_connection1 = session1._fluent_connection
     session1.exit()
-    assert not session1._fluent_connection.wait_process_finished(wait=0)
+    assert not fl_connection1.wait_process_finished(wait=0)
 
     session2 = pyfluent.launch_fluent()
+    fl_connection2 = session2._fluent_connection
     session2.exit(wait=60)
-    assert session2._fluent_connection.wait_process_finished(wait=0)
+    assert fl_connection2.wait_process_finished(wait=0)
 
     session3 = pyfluent.launch_fluent()
+    fl_connection3 = session3._fluent_connection
     session3.exit(wait=True)
-    assert session3._fluent_connection.wait_process_finished(wait=0)
+    assert fl_connection3.wait_process_finished(wait=0)
 
     with pytest.raises(WaitTypeError):
         session4 = pyfluent.launch_fluent()

tests/test_meshing_workflow.py

@@ -441,3 +441,22 @@ def test_setting_none_type_tasks(new_meshing_session):
         meshing.workflow.TaskObject["Describe Overset Features"].CommandName()
         == "DescribeOversetFeatures"
     )
+
+
+def test_inaccessible_meshing_attributes_after_switching_to_solver(
+    new_meshing_session_wo_exit,
+):
+    meshing = new_meshing_session_wo_exit
+    assert meshing.is_active() is True
+    solver = meshing.switch_to_solver()
+    assert solver.is_active() is True
+    assert meshing.is_active() is False
+    with pytest.raises(AttributeError):
+        # 'switched' attribute is not there in Meshing.
+        assert meshing.switched
+    assert dir(meshing) == ["is_active"]
+    del meshing
+    assert solver.is_active() is True
+    solver.exit()
+    assert solver.is_active() is False
+    assert dir(solver) == ["is_active"]

tests/test_meshingmode/test_meshing_launch.py

@@ -102,20 +102,20 @@ def test_launch_pure_meshing(mixing_elbow_watertight_pure_meshing_session):
 
 @pytest.mark.fluent_version("latest")
 @pytest.mark.codegen_required
-def test_launch_meshing_and_switch(new_meshing_session):
-    meshing = new_meshing_session
-    assert not meshing._switched
-    _ = meshing.switch_to_solver()
-    assert meshing._switched
-    assert not meshing.tui
-    assert not meshing.meshing
-    assert not meshing.workflow
-    assert not meshing.watertight
+def test_launch_meshing_and_switch(new_meshing_session_wo_exit):
+    meshing = new_meshing_session_wo_exit
+    assert meshing.is_active() is True
+    solver = meshing.switch_to_solver()
+    assert meshing.is_active() is False
+    for attr in ("tui", "meshing", "workflow", "watertight"):
+        with pytest.raises(AttributeError):
+            getattr(meshing, attr)
+    solver.exit()
 
 
 @pytest.mark.fluent_version("latest")
 @pytest.mark.codegen_required
-def test_meshing_streaming_and_switch(new_meshing_session):
+def test_meshing_streaming_and_switch(new_meshing_session_wo_exit):
 
     def on_case_loaded(session, event_info):
         on_case_loaded.loaded = True
@@ -127,7 +127,7 @@ def on_trancript(transcript):
 
     on_trancript.called = False
 
-    meshing = new_meshing_session
+    meshing = new_meshing_session_wo_exit
 
     meshing.events.register_callback(MeshingEvent.CASE_LOADED, on_case_loaded)
     meshing.transcript.register_callback(on_trancript)
@@ -145,6 +145,7 @@ def on_trancript(transcript):
 
     assert not on_trancript.called
     assert not on_case_loaded.loaded
+    solver.exit()
 
 
 @pytest.mark.fluent_version("latest")