diff --git a/src/ansys/workbench/core/workbench_launcher.py b/src/ansys/workbench/core/workbench_launcher.py index 94e1f700..5e040fa2 100644 --- a/src/ansys/workbench/core/workbench_launcher.py +++ b/src/ansys/workbench/core/workbench_launcher.py @@ -319,24 +319,55 @@ def __getenv(self, key): def exit(self): """End the launched Workbench server.""" if self._process: + launched_process_description = self.__describe_process( + self._process, fallback_process_id=self._process_id + ) try: if self._wmi: for p in self.__collect_process_tree(): - logging.info("Shutting down " + p.Name + " ...") + logging.info("Shutting down " + self.__describe_process(p) + " ...") try: p.Terminate() except Exception as ex: - logging.info(f"Failed to terminate process {p.Name}: {ex}") + logging.info( + "Failed to terminate process " + + self.__describe_process(p) + + f": {ex}" + ) self._process.Terminate() else: self._process.terminate() except Exception as ex: - logging.info(f"Failed to terminate process {self._process.pid}: {ex}") + logging.info(f"Failed to terminate process {launched_process_description}: {ex}") self._wmi_connection = None self._process_id = -1 self._process = None + def __describe_process(self, process, fallback_process_id=None): + """Return a stable process description for logging.""" + process_name = self.__get_process_attribute(process, "Name") + process_id = self.__get_process_attribute(process, "ProcessId") + if process_id is None: + process_id = self.__get_process_attribute(process, "pid") + + if process_name and process_id is not None: + return f"{process_name} ({process_id})" + if process_name: + return process_name + if process_id is not None: + return str(process_id) + if fallback_process_id is not None and fallback_process_id >= 0: + return str(fallback_process_id) + return "unknown process" + + def __get_process_attribute(self, process, attribute_name): + """Read a process attribute without letting property getters abort shutdown.""" + try: + return getattr(process, attribute_name, None) + except Exception: + return None + def __collect_process_tree(self): # collect parent-children mapping children = {self._process_id: []} diff --git a/tests/test_workbench_launcher.py b/tests/test_workbench_launcher.py new file mode 100644 index 00000000..66612a7a --- /dev/null +++ b/tests/test_workbench_launcher.py @@ -0,0 +1,89 @@ + # Copyright (C) 2023 - 2026 ANSYS, Inc. and/or its affiliates. + # SPDX-License-Identifier: MIT + # + # + # Permission is hereby granted, free of charge, to any person obtaining a copy + # of this software and associated documentation files (the "Software"), to deal + # in the Software without restriction, including without limitation the rights + # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + # copies of the Software, and to permit persons to whom the Software is + # furnished to do so, subject to the following conditions: + # + # The above copyright notice and this permission notice shall be included in all + # copies or substantial portions of the Software. + # + # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + # SOFTWARE. + +"""Tests for workbench launcher.""" + +from ansys.workbench.core.workbench_launcher import Launcher + + +class MissingWmiProcess: + """Represent a terminated WMI process object.""" + + def __init__(self, process_name=None): + self.Name = process_name + + @property + def ProcessId(self): + """Raise because the WMI process no longer exists.""" + raise RuntimeError("ProcessId is unavailable") + + @property + def pid(self): + """Raise because WMI objects do not expose pid safely here.""" + raise RuntimeError("pid is unavailable") + + def Terminate(self): + """Raise because the process is already gone.""" + raise RuntimeError("process not found") + + +def test_exit_handles_missing_wmi_process(): + """Ensure exit does not fail when the launched WMI process already exited.""" + launcher = object.__new__(Launcher) + launcher._wmi = object() + launcher._libc = None + launcher._wmi_connection = object() + launcher._process_id = 61358 + launcher._process = MissingWmiProcess("RunWB2.exe") + launcher._Launcher__collect_process_tree = lambda: [] + + launcher.exit() + + assert launcher._wmi_connection is None + assert launcher._process_id == -1 + assert launcher._process is None + + +def test_describe_process_uses_available_name_when_ids_fail(): + """Use the process name when id lookups fail with runtime exceptions.""" + launcher = object.__new__(Launcher) + launcher._process_id = 61358 + + assert launcher._Launcher__describe_process(MissingWmiProcess("RunWB2.exe")) == "RunWB2.exe" + + +def test_describe_process_does_not_use_parent_id_for_child_process(): + """Return an unknown label when a child process cannot describe itself.""" + launcher = object.__new__(Launcher) + launcher._process_id = 61358 + + assert launcher._Launcher__describe_process(MissingWmiProcess()) == "unknown process" + + +def test_describe_process_uses_explicit_fallback_process_id(): + """Use the explicit fallback for the launched process description only.""" + launcher = object.__new__(Launcher) + launcher._process_id = 61358 + + assert launcher._Launcher__describe_process( + MissingWmiProcess(), fallback_process_id=launcher._process_id + ) == "61358"