diff --git a/doc/changelog.d/5892.miscellaneous.md b/doc/changelog.d/5892.miscellaneous.md new file mode 100644 index 00000000000..46625c05697 --- /dev/null +++ b/doc/changelog.d/5892.miscellaneous.md @@ -0,0 +1 @@ +Improve API and security in Desktop \ No newline at end of file diff --git a/doc/source/User_guide/index.rst b/doc/source/User_guide/index.rst index 4fc34833d2b..f5d54104ddf 100644 --- a/doc/source/User_guide/index.rst +++ b/doc/source/User_guide/index.rst @@ -81,6 +81,13 @@ For end-to-end examples, see `Examples ` How to create and analyze EMIT designs. + .. grid-item-card:: Security considerations + :link: security_consideration + :link-type: doc + :margin: 2 2 0 0 + + Information on security considerations. + .. toctree:: :hidden: :maxdepth: 2 @@ -95,3 +102,4 @@ For end-to-end examples, see `Examples ` files postprocessing emit_modeler + security_consideration diff --git a/doc/source/User_guide/security_consideration.rst b/doc/source/User_guide/security_consideration.rst new file mode 100644 index 00000000000..952dbe6d448 --- /dev/null +++ b/doc/source/User_guide/security_consideration.rst @@ -0,0 +1,38 @@ +Security considerations +======================= + +This section provides information on security considerations for the use +of PyAEDT. It is important to understand the capabilities which PyAEDT +provides, especially when using it to build applications or scripts that +accept untrusted input. + +.. _security_launch_aedt: + +Launching AEDT +-------------- + +The :py:func:`.launch_aedt` and :py:func:`.launch_aedt_in_lsf` functions can be used +to launch AEDT. The executable which is launched is configured with the function +parameters, environment variables and the +`settings `_. +This may allow an attacker to launch arbitrary executables on the system. When +exposing the launch function to untrusted users, it is important to validate that +the executable path, environment variables (for example ``"ANSYSEM_ROOT"``, +``ANSYSEM_PY_CLIENT_ROOT`` and ``ANSYSEMSV_ROOT``) and PyAEDT settings are safe. +Otherwise, hard-code them in the application. + +.. _security_ansys_cloud: + +Retrieving Ansys Cloud information +---------------------------------- + +The :py:func:`.get_cloud_job_info` and :py:func:`.get_available_cloud_config` +functions can be used to retrieve information related to Ansys Cloud. +The executable which is launched is configured with the function +parameters and the AEDT installation that is detected. Since finding the AEDT +installation path is based of environment variables, this may allow an attacker +to launch arbitrary executables on the system. When exposing the launch function +to untrusted users, it is important to validate that environment variables like +``"ANSYSEM_ROOT"``, ``ANSYSEM_PY_CLIENT_ROOT`` and ``ANSYSEMSV_ROOT`` are safe. +Otherwise, hard-code them in the application. + diff --git a/doc/styles/config/vocabularies/ANSYS/accept.txt b/doc/styles/config/vocabularies/ANSYS/accept.txt index d166d7beb89..7a5d99aeec3 100644 --- a/doc/styles/config/vocabularies/ANSYS/accept.txt +++ b/doc/styles/config/vocabularies/ANSYS/accept.txt @@ -14,6 +14,7 @@ busbars Bz CHECKIN [Cc]ircuit +[Cc]loud codecov (?i)Com COM interface @@ -97,6 +98,7 @@ subcircuit [Tt]oolkit [Mm]anager Twin Builder Uncomment +untrusted utils vias manual_settings diff --git a/src/ansys/aedt/core/application/design.py b/src/ansys/aedt/core/application/design.py index d5cde606662..3562d08a393 100644 --- a/src/ansys/aedt/core/application/design.py +++ b/src/ansys/aedt/core/application/design.py @@ -60,7 +60,7 @@ from ansys.aedt.core.application.design_solutions import solutions_defaults from ansys.aedt.core.application.variables import DataSet from ansys.aedt.core.application.variables import VariableManager -from ansys.aedt.core.desktop import _init_desktop_from_design +from ansys.aedt.core.desktop import Desktop from ansys.aedt.core.desktop import exception_to_desktop from ansys.aedt.core.generic.constants import AEDT_UNITS from ansys.aedt.core.generic.constants import unit_system @@ -196,7 +196,7 @@ def __init__( self._design_datasets: List = [] self.close_on_exit: bool = close_on_exit self._desktop_class = None - self._desktop_class = _init_desktop_from_design( + self._desktop_class = self.__init_desktop_from_design( version, non_graphical, new_desktop, @@ -4233,6 +4233,12 @@ def edit_notes(self, text): self.odesign.EditNotes(text) return True + @classmethod + def __init_desktop_from_design(cls, *args, **kwargs): + """Internal instantiation of the ``Desktop`` class.""" + Desktop._invoked_from_design = True + return Desktop(*args, **kwargs) + class DesignSettings: """Get design settings for the current AEDT app. @@ -4246,20 +4252,6 @@ def __init__(self, app): self._app: Any = app self.manipulate_inputs: Optional[DesignSettingsManipulation] = None - @property - def design_settings(self) -> Optional[Any]: - """Design settings.""" - try: - return self._app.odesign.GetChildObject("Design Settings") - except GrpcApiError: # pragma: no cover - self._app.logger.error("Failed to retrieve design settings.") - return None - - @property - def available_properties(self) -> List[str]: - """Available properties names for the current design.""" - return [prop for prop in self.design_settings.GetPropNames() if not prop.endswith("/Choices")] - def __repr__(self) -> str: lines = ["{"] for prop in self.available_properties: @@ -4294,6 +4286,20 @@ def __getitem__(self, key: str) -> Optional[Any]: def __contains__(self, item: str) -> bool: return item in self.available_properties + @property + def design_settings(self) -> Optional[Any]: + """Design settings.""" + try: + return self._app.odesign.GetChildObject("Design Settings") + except GrpcApiError: # pragma: no cover + self._app.logger.error("Failed to retrieve design settings.") + return None + + @property + def available_properties(self) -> List[str]: + """Available properties names for the current design.""" + return [prop for prop in self.design_settings.GetPropNames() if not prop.endswith("/Choices")] + class DesignSettingsManipulation: @abstractmethod diff --git a/src/ansys/aedt/core/desktop.py b/src/ansys/aedt/core/desktop.py index c18dd9bd626..9a6366f8763 100644 --- a/src/ansys/aedt/core/desktop.py +++ b/src/ansys/aedt/core/desktop.py @@ -38,43 +38,39 @@ import re import shutil import socket +import subprocess # nosec import sys import tempfile import time import traceback +from typing import Union import warnings -from ansys.aedt.core import __version__ as pyaedt_version +from ansys.aedt.core import __version__ from ansys.aedt.core.aedt_logger import AedtLogger from ansys.aedt.core.aedt_logger import pyaedt_logger -from ansys.aedt.core.generic.file_utils import generate_unique_name -from ansys.aedt.core.generic.general_methods import is_linux -from ansys.aedt.core.generic.general_methods import is_windows -import grpc - -if is_linux: - os.environ["ANS_NODEPCHECK"] = str(1) - -import subprocess - -from ansys.aedt.core import __version__ from ansys.aedt.core.generic.file_utils import available_license_feature +from ansys.aedt.core.generic.file_utils import generate_unique_name from ansys.aedt.core.generic.file_utils import open_file from ansys.aedt.core.generic.general_methods import active_sessions from ansys.aedt.core.generic.general_methods import com_active_sessions from ansys.aedt.core.generic.general_methods import get_string_version from ansys.aedt.core.generic.general_methods import grpc_active_sessions from ansys.aedt.core.generic.general_methods import inside_desktop +from ansys.aedt.core.generic.general_methods import is_linux +from ansys.aedt.core.generic.general_methods import is_windows from ansys.aedt.core.generic.general_methods import pyaedt_function_handler +from ansys.aedt.core.generic.settings import Settings from ansys.aedt.core.generic.settings import settings from ansys.aedt.core.internal.aedt_versions import aedt_versions +from ansys.aedt.core.internal.checks import min_aedt_version from ansys.aedt.core.internal.desktop_sessions import _desktop_sessions from ansys.aedt.core.internal.desktop_sessions import _edb_sessions +from ansys.aedt.core.internal.errors import AEDTRuntimeError +import grpc pathname = Path(__file__) - pyaedtversion = __version__ - modules = [tup[1] for tup in pkgutil.iter_modules()] @@ -88,31 +84,43 @@ def grpc_server_on(channel, timeout_sec=5) -> bool: @pyaedt_function_handler() -def launch_aedt(full_path, non_graphical, port, student_version, first_run=True): # pragma: no cover - """Launch AEDT in gRPC mode.""" +def launch_aedt( + full_path: Union[str, Path], non_graphical: bool, port: int, student_version: bool, first_run: bool = True +): # pragma: no cover + """Launch AEDT in gRPC mode. - def launch_desktop_on_port(): - command = [str(full_path), "-grpcsrv", str(port)] - if non_graphical: - command.append("-ng") - if settings.wait_for_license: - command.append("-waitforlicense") - if settings.aedt_log_file: - command.extend(["-Logfile", settings.aedt_log_file]) + .. warning:: - if is_linux: # pragma: no cover - command.append("&") - subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - else: - subprocess.Popen( - " ".join(command), - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - creationflags=subprocess.DETACHED_PROCESS, - ) + Do not execute this function with untrusted input parameters. + See the :ref:`security guide` for details. + """ - launch_desktop_on_port() + full_path = Path(full_path) + if not full_path.exists() or not full_path.name.lower() in { + "ansysedt", + "ansysedtsv", + "ansysedtsv.exe", + "ansysedt.exe", + }: + raise ValueError(f"The path {full_path} is not a valid executable.") + _check_port(port) + + command = [str(full_path), "-grpcsrv", str(port)] + if non_graphical: + command.append("-ng") + if settings.wait_for_license: + command.append("-waitforlicense") + if settings.aedt_log_file: + command.extend(["-Logfile", settings.aedt_log_file]) + + kwargs = { + "stdin": subprocess.DEVNULL, + "stdout": subprocess.DEVNULL, + "stderr": subprocess.DEVNULL, + } + if is_windows: + kwargs["creationflags"] = subprocess.DETACHED_PROCESS + subprocess.Popen(command, **kwargs) # nosec on_ci = os.getenv("ON_CI", "False") if not student_version and on_ci != "True" and not settings.skip_license_check: @@ -148,8 +156,18 @@ def launch_desktop_on_port(): return True, port +@pyaedt_function_handler() def launch_aedt_in_lsf(non_graphical, port): # pragma: no cover - """Launch AEDT in LSF in gRPC mode.""" + """Launch AEDT in LSF in gRPC mode. + + .. warning:: + + Do not execute this function with untrusted input parameters. + See the :ref:`security guide` for details. + """ + _check_port(port) + _check_settings(settings) + if not settings.custom_lsf_command: # pragma: no cover if hasattr(settings, "lsf_osrel") and hasattr(settings, "lsf_ui"): select_str = ( @@ -159,32 +177,19 @@ def launch_aedt_in_lsf(non_graphical, port): # pragma: no cover select_str = f'"select[(ui={settings.lsf_ui}) rusage[mem={settings.lsf_ram}]]"' else: select_str = f'"-R rusage[mem={settings.lsf_ram}"' + command = [ + "bsub", + "-n", + str(settings.lsf_num_cores), + "-R", + select_str, + "-Is", + settings.lsf_aedt_command, + "-grpcsrv", + str(port), + ] if settings.lsf_queue: - command = [ - "bsub", - "-n", - str(settings.lsf_num_cores), - "-R", - select_str, - f'"rusage[mem={settings.lsf_ram}]"', - f"-q {settings.lsf_queue}", - "-Is", - settings.lsf_aedt_command, - "-grpcsrv", - str(port), - ] - else: - command = [ - "bsub", - "-n", - str(settings.lsf_num_cores), - "-R", - select_str, - "-Is", - settings.lsf_aedt_command, - "-grpcsrv", - str(port), - ] + command.append(f"-q {settings.lsf_queue}") if non_graphical: command.append("-ng") if settings.wait_for_license: @@ -197,12 +202,12 @@ def launch_aedt_in_lsf(non_graphical, port): # pragma: no cover command.append(str(port)) command_str = " ".join(str(x) for x in command) pyaedt_logger.info("LSF Command: '" + command_str + "'") - lsf_message = lambda x: x.stderr.readline().strip().decode("utf-8", "replace") # nosec - try: # nosec - p = subprocess.Popen(command_str, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # nosec - except FileNotFoundError: # nosec - p = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # nosec - pyaedt_logger.error(lsf_message(p)) + lsf_message = lambda x: x.stderr.readline().strip().decode("utf-8", "replace") + try: + p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # nosec + except FileNotFoundError as e: + raise AEDTRuntimeError("Failed to start AEDT in LSF. Check the LSF configuration settings.") from e + timeout = settings.lsf_timeout i = 0 while i < timeout: @@ -225,6 +230,24 @@ def launch_aedt_in_lsf(non_graphical, port): # pragma: no cover return False, err +def _check_port(port): + """Check port.""" + try: + port = int(port) + except ValueError: + raise ValueError(f"The port {port} is not a valid integer.") + + +def _check_settings(settings: Settings): + """Check settings.""" + if not isinstance(settings.lsf_num_cores, int) or settings.lsf_num_cores <= 0: + raise ValueError("Invalid number of cores.") + if not isinstance(settings.lsf_ram, int) or settings.lsf_ram <= 0: + raise ValueError("Invalid memory value.") + if not settings.lsf_aedt_command: + raise ValueError("Invalid LSF AEDT command.") + + def _is_port_occupied(port, machine_name=""): if not port: return False @@ -278,124 +301,6 @@ def exception_to_desktop(ex_value, tb_data): # pragma: no cover pyaedt_logger.error(el) -def _delete_objects(): - settings.remote_api = False - pyaedt_logger.remove_all_project_file_logger() - try: - del sys.modules["glob"] - except Exception: - pass - gc.collect() - - -@pyaedt_function_handler() -def _close_aedt_application(desktop_class, close_desktop, pid, is_grpc_api): - """Release the AEDT API. - - Parameters - ---------- - desktop_class : :class:ansys.aedt.core.desktop.Desktop - Desktop class. - close_desktop : bool - Whether to close the active AEDT session. - pid : int - Process ID of the desktop app that is being closed. - is_grpc_api : bool - Whether the active AEDT session is gRPC or COM. - - Returns - ------- - bool - ``True`` when successful, ``False`` when failed. - - """ - if settings.remote_rpc_session or (settings.aedt_version >= "2022.2" and is_grpc_api): - if close_desktop and desktop_class.parent_desktop_id: # pragma: no cover - pyaedt_logger.error("A child desktop session is linked to this session.") - pyaedt_logger.error("Multiple desktop sessions must be released in reverse order.") - return False - elif close_desktop: - try: - if settings.use_multi_desktop: # pragma: no cover - os.kill(pid, 9) - else: - desktop_class.odesktop.QuitApplication() - if desktop_class.is_grpc_api: - desktop_class.grpc_plugin.Release() - timeout = 20 - while pid in active_sessions(): - time.sleep(1) - if timeout == 0: - os.kill(pid, 9) - break - timeout -= 1 - if _desktop_sessions: - for v in _desktop_sessions.values(): - if pid in v.parent_desktop_id: # pragma: no cover - del v.parent_desktop_id[v.parent_desktop_id.index(pid)] - return True - except Exception: # pragma: no cover - warnings.warn("Something went wrong closing AEDT. Exception in `_main.oDesktop.QuitApplication()`.") - else: # pragma: no cover - for k, d in _desktop_sessions.items(): - if k == pid: - d.grpc_plugin.recreate_application(True) - d.grpc_plugin.Release() - return True - elif not inside_desktop: # pragma: no cover - if close_desktop: - try: - if settings.use_multi_desktop: - desktop_class.odesktop.QuitApplication() - else: - os.kill(pid, 9) - except Exception: # pragma: no cover - warnings.warn("Something went wrong closing AEDT. Exception in `os.kill(pid, 9)`.") - return False - else: - try: - scopeID = 0 - while scopeID <= 5: - desktop_class.COMUtil.ReleaseCOMObjectScope(desktop_class.COMUtil.PInvokeProxyAPI, scopeID) - scopeID += 1 - except Exception: - pyaedt_logger.warning( - "Something went wrong releasing AEDT. Exception in `_main.COMUtil.ReleaseCOMObjectScope`." - ) - if not settings.remote_rpc_session and close_desktop: # pragma: no cover - timeout = 10 - while pid in active_sessions(): - time.sleep(1) - timeout -= 1 - if timeout == 0: - try: - os.kill(pid, 9) - return True - except Exception: # pragma: no cover - warnings.warn("Something went wrong closing AEDT. Exception in `os.kill(pid, 9)` after timeout.") - return False - break - - return True - - -def run_process(command, bufsize=None): - """Run a process with a subprocess. - - Parameters - ---------- - command : str - Command to execute. - bufsize : int, optional - Buffer size. The default is ``None``. - - """ - if bufsize: # pragma no cover - return subprocess.call(command, bufsize=bufsize) - else: - return subprocess.call(command) - - def is_student_version(oDesktop): edt_root = Path(oDesktop.GetExeDir()) if is_windows and Path(edt_root).is_dir(): @@ -404,15 +309,6 @@ def is_student_version(oDesktop): return False -def _init_desktop_from_design(*args, **kwargs): - """Distinguishes if the ``Desktop`` class is initialized internally. - - It can be initialized from the ``Design`` class or directly from the user. - For example, ``desktop=Desktop()``).""" - Desktop._invoked_from_design = True - return Desktop(*args, **kwargs) - - class Desktop(object): """Provides the Ansys Electronics Desktop (AEDT) interface. @@ -491,7 +387,7 @@ def __new__(cls, *args, **kwargs): aedt_process_id = kwargs.get("aedt_process_id") or None if (not args or len(args) < 8) else args[7] if not settings.remote_api: pyaedt_logger.info(f"Python version {sys.version}.") - pyaedt_logger.info(f"PyAEDT version {pyaedt_version}.") + pyaedt_logger.info(f"PyAEDT version {__version__}.") if settings.use_multi_desktop and not inside_desktop and new_desktop: pyaedt_logger.info("Initializing new Desktop session.") return object.__new__(cls) @@ -534,6 +430,7 @@ def __init__( port=0, aedt_process_id=None, ): + """Initialize desktop.""" # Used to track whether desktop release must be performed at exit or not. self.__closed = False if _desktop_sessions and version is None: @@ -553,8 +450,6 @@ def __init__( self.parent_desktop_id = [] self._odesktop = None self._connected_app_instances = 0 - - """Initialize desktop.""" self.launched_by_pyaedt = False # Used in unit tests. The ``PYAEDT_NON_GRAPHICAL`` environment variable overrides @@ -599,7 +494,7 @@ def __init__( self._logger.info("Debug logger is enabled. PyAEDT methods will be logged.") else: self._logger.info("Debug logger is disabled. PyAEDT methods will not be logged.") - student_version_flag, version_key, version = self._assert_version(version, student_version) + student_version_flag, version_key, version = self.__check_version(version, student_version) # start the AEDT opening decision tree # starting_mode can be one of these: "grpc", "com", "console_in", "console_out" @@ -665,7 +560,7 @@ def __init__( settings.aedt_version = version_key if starting_mode == "com": # pragma no cover self._logger.info("Launching PyAEDT with CPython and PythonNET.") - self._init_dotnet( + self.__init_dotnet( non_graphical, new_desktop, version, @@ -675,11 +570,11 @@ def __init__( ) elif starting_mode == "grpc": self._logger.info("Launching PyAEDT with gRPC plugin.") - self._init_grpc(non_graphical, new_desktop, version, student_version_flag, version_key) + self.__init_grpc(non_graphical, new_desktop, version, student_version_flag, version_key) - self._set_logger_file() + self.__set_logger_file() settings.enable_desktop_logs = not self.non_graphical - self._init_desktop() + self.__init_desktop() current_pid = int(self.odesktop.GetProcessID()) if aedt_process_id and not new_desktop and aedt_process_id != current_pid: # pragma no cover @@ -713,7 +608,7 @@ def __enter__(self): def __exit__(self, ex_type, ex_value, ex_traceback): # pragma no cover # Write the trace stack to the log file if an exception occurred in the main script. if ex_type: - err = self._exception(ex_value, ex_traceback) + self.__exception(ex_value, ex_traceback) if self.close_on_exit: self.release_desktop(close_projects=self.close_on_exit, close_on_exit=self.close_on_exit) self.__closed = True @@ -757,6 +652,147 @@ def __getitem__(self, project_design_name): return get_pyaedt_app(projectname, designname, self) + # ################################## # + # Properties # + # ################################## # + + @property + @min_aedt_version("2023.2") + def are_there_simulations_running(self): + """Check if there are simulation running. + + Returns + ------- + float + + """ + return self.odesktop.AreThereSimulationsRunning() + + @property + def current_version(self): + """Current AEDT version.""" + return aedt_versions.current_version + + @property + def current_student_version(self): + """Current AEDT student version.""" + return aedt_versions.current_student_version + + @property + def installed_versions(self): + """Dictionary of AEDT versions installed on the system and their installation paths.""" + return aedt_versions.installed_versions + + @property + def install_path(self): + """Installation path for AEDT.""" + version_key = settings.aedt_version + try: + return self.installed_versions[version_key] + except Exception: # pragma: no cover + return self.installed_versions[version_key + "CL"] + + @property + def logger(self): + """AEDT logger.""" + return self._logger + + @property + def odesktop(self): + """AEDT instance containing all projects and designs. + + Examples + -------- + Get the COM object representing the desktop. + + >>> from ansys.aedt.core import Desktop + >>> d = Desktop() + >>> d.odesktop + """ + if settings.use_grpc_api: + tries = 0 + while tries < 5: + try: + self._odesktop = self.grpc_plugin.odesktop + return self._odesktop + except Exception: + tries += 1 + time.sleep(1) + return self._odesktop + + @odesktop.setter + def odesktop(self, val): + self._odesktop = val + + @property + def messenger(self): + """Messenger manager for the AEDT logger.""" + return pyaedt_logger + + @property + def personallib(self): + """PersonalLib directory. + + Returns + ------- + str + Full absolute path for the ``PersonalLib`` directory. + + """ + return self.odesktop.GetPersonalLibDirectory() + + @property + def src_dir(self): + """Python source directory. + + Returns + ------- + str + Full absolute path for the ``python`` directory. + + """ + return Path(__file__) + + @property + def syslib(self): + """SysLib directory. + + Returns + ------- + str + Full absolute path for the ``SysLib`` directory. + + """ + return self.odesktop.GetLibraryDirectory() + + @property + def pyaedt_dir(self): + """PyAEDT directory. + + Returns + ------- + str + Full absolute path for the ``pyaedt`` directory. + + """ + return Path(__file__).parent + + @property + def userlib(self): + """UserLib directory. + + Returns + ------- + str + Full absolute path for the ``UserLib`` directory. + + """ + return self.odesktop.GetUserLibDirectory() + + # ############################################ # + # Public methods # + # ############################################ # + @pyaedt_function_handler() def active_design(self, project_object=None, name=None, design_type=None): """Get the active design. @@ -833,15 +869,6 @@ def active_project(self, name=None): self.close_windows() return active_project - @property - def install_path(self): - """Installation path for AEDT.""" - version_key = settings.aedt_version - try: - return self.installed_versions[version_key] - except Exception: # pragma: no cover - return self.installed_versions[version_key + "CL"] - @pyaedt_function_handler() def close_windows(self): """Close all windows. @@ -858,591 +885,212 @@ def close_windows(self): self.odesktop.CloseAllWindows() return True - @property - def current_version(self): - """Current AEDT version.""" - return aedt_versions.current_version + @pyaedt_function_handler() + def project_list(self): + """Get a list of projects. - @property - def current_student_version(self): - """Current AEDT student version.""" - return aedt_versions.current_student_version + Returns + ------- + List + List of projects. - @property - def installed_versions(self): - """Dictionary of AEDT versions installed on the system and their installation paths.""" - return aedt_versions.installed_versions + """ + return list(self.odesktop.GetProjectList()) - def _init_desktop(self): - # run it after the settings.non_graphical is set - self.pyaedt_version = pyaedtversion - settings.aedt_version = self.odesktop.GetVersion()[0:6] - self.odesktop.RestoreWindow() - settings.aedt_install_dir = self.odesktop.GetExeDir() + @pyaedt_function_handler() + def analyze_all(self, project=None, design=None): # pragma: no cover + """Analyze all setups in a project. - def _assert_version(self, specified_version, student_version): - if self.current_version == "" and aedt_versions.latest_version == "": - raise Exception("AEDT is not installed on your system. Install AEDT version 2022 R2 or higher.") - if not specified_version: - if student_version and self.current_student_version: - specified_version = self.current_student_version - elif student_version and self.current_version: - specified_version = self.current_version - student_version = False - self.logger.warning("AEDT Student Version not found on the system. Using regular version.") + Parameters + ---------- + project : str, optional + Project name. The default is ``None``, in which case the active project + is used. + design : str, optional + Design name. The default is ``None``, in which case all designs in + the project are analyzed. + + Returns + ------- + bool + ``True`` when successful, ``False`` when failed. + """ + if not project: + oproject = self.active_project() + else: + oproject = self.active_project(project) + if oproject: + if not design: + oproject.AnalyzeAll() else: - if self.current_version != "": - specified_version = self.current_version - else: - specified_version = aedt_versions.latest_version - if "SV" in specified_version: - student_version = True - self.logger.warning("Only AEDT Student Version found on the system. Using Student Version.") - elif student_version: - specified_version += "SV" - specified_version = get_string_version(specified_version) + odesign = self.active_design(oproject, design) + if odesign: + odesign.AnalyzeAll() + return True - if float(specified_version[0:6]) < 2019: - raise ValueError("PyAEDT supports AEDT version 2021 R1 and later. Recommended version is 2022 R2 or later.") - elif float(specified_version[0:6]) < 2022.2: - warnings.warn( - """PyAEDT has limited capabilities when used with an AEDT version earlier than 2022 R2. - Update your AEDT installation to 2022 R2 or later.""" - ) - if not (specified_version in self.installed_versions) and not ( - specified_version + "CL" in self.installed_versions - ): - raise ValueError( - f"Specified version {specified_version[0:6]}{' Student Version' if student_version else ''} is not " - f"installed on your system" - ) + @pyaedt_function_handler() + def clear_messages(self): + """Clear all AEDT messages. - version = "Ansoft.ElectronicsDesktop." + specified_version[0:6] - settings.aedt_install_dir = None - if specified_version in self.installed_versions: - settings.aedt_install_dir = self.installed_versions[specified_version] - if settings.remote_rpc_session: - try: - version = "Ansoft.ElectronicsDesktop." + settings.remote_rpc_session.aedt_version[0:6] - return settings.remote_rpc_session.student_version, settings.remote_rpc_session.aedt_version, version - except Exception: - return False, "", "" + Returns + ------- + bool + ``True`` when successful, ``False`` when failed. + """ + self.odesktop.ClearMessages("", "", 3) + return True - return student_version, specified_version, version + @pyaedt_function_handler() + def save_project(self, project_name=None, project_path=None): + """Save the project. - @staticmethod - def _run_student(): # pragma: no cover - DETACHED_PROCESS = 0x00000008 - path = Path(Path(settings.aedt_install_dir) / "ansysedtsv.exe").resolve(strict=True) - _ = subprocess.Popen([path], creationflags=DETACHED_PROCESS).pid - time.sleep(5) + Parameters + ---------- + project_name : str, optional + Project name. The default is ``None``, in which case the active project + is used. + project_path : str, optional + Full path to the project. The default is ``None``. If a path is + provided, ``save as`` is used. - def _dispatch_win32(self, version): # pragma: no cover - from ansys.aedt.core.internal.clr_module import win32_client + Returns + ------- + bool + ``True`` when successful, ``False`` when failed. + """ + if not project_name: + oproject = self.odesktop.GetActiveProject() + else: + oproject = self.odesktop.SetActiveProject(project_name) + if project_path: + oproject.SaveAs(project_path, True) + else: + oproject.Save() + return True - o_ansoft_app = win32_client.Dispatch(version) - self.odesktop = o_ansoft_app.GetAppDesktop() - self.isoutsideDesktop = True + @pyaedt_function_handler() + def copy_design(self, project_name=None, design_name=None, target_project=None): # pragma: no cover + """Copy a design and paste it in an existing project or new project. - def _init_dotnet( - self, - non_graphical, - new_aedt_session, - version, - student_version, - version_key, - aedt_process_id=None, - ): # pragma: no cover - import pythoncom + .. deprecated:: 0.6.31 + Use :func:`copy_design_from` instead. - pythoncom.CoInitialize() + Parameters + ---------- + project_name : str, optional + Project name. The default is ``None``, in which case the active project + is used. + design_name : str, optional + Design name. The default is ``None``. + target_project : str, optional + Target project. The default is ``None``. - if is_linux: - raise Exception( - "PyAEDT supports COM initialization in Windows only. To use in Linux, upgrade to AEDT 2022 R2 or later." - ) - base_path = settings.aedt_install_dir - sys.path.insert(0, base_path) - sys.path.insert(0, str(Path(base_path) / "PythonFiles" / "DesktopPlugin")) - launch_msg = f"AEDT installation Path {base_path}." - self.logger.info(launch_msg) - processID = [] - if is_windows: - processID = com_active_sessions(version, student_version, non_graphical) - if student_version and not processID: # Opens an instance if processID is an empty list - self._run_student() - elif non_graphical or new_aedt_session or not processID: - # Force new object if no non-graphical instance is running or if there is not an already existing process. - self._initialize(non_graphical=non_graphical, new_session=True, is_grpc=False, version=version_key) + Returns + ------- + bool + ``True`` when successful, ``False`` when failed. + """ + if not project_name: + oproject = self.active_project() else: - self._initialize(new_session=False, is_grpc=False, version=version_key) - processID2 = [] - if is_windows: - processID2 = com_active_sessions(version, student_version, non_graphical) - proc = [i for i in processID2 if i not in processID] # Looking for the "new" process - if ( - not proc and (not new_aedt_session) and aedt_process_id - ): # if it isn't a new aedt session and a process ID is given - proc = [aedt_process_id] - elif not proc: - proc = processID2 - if proc == processID2 and len(processID2) > 1: - self._dispatch_win32(version) - elif version_key >= "2021.2": + oproject = self.active_project(project_name) + if oproject: + if not design_name: + odesign = self.active_design(oproject) + else: + odesign = self.active_design(oproject, design_name) + if odesign: + oproject.CopyDesign(design_name) + if not target_project: + oproject.Paste() + return True + else: + oproject_target = self.active_project(target_project) + if not oproject_target: + oproject_target = self.odesktop.NewProject(target_project) + oproject_target.Paste() + return True + else: + return False - context = pythoncom.CreateBindCtx(0) - running_coms = pythoncom.GetRunningObjectTable() - monikiers = running_coms.EnumRunning() - for monikier in monikiers: - m = re.search(version[10:] + r"\.\d:" + str(proc[0]), monikier.GetDisplayName(context, monikier)) - if m: - obj = running_coms.GetObject(monikier) - self.isoutsideDesktop = True - from ansys.aedt.core.internal.clr_module import win32_client + @pyaedt_function_handler() + def project_path(self, project_name=None): + """Get the path to the project. - self.odesktop = win32_client.Dispatch(obj.QueryInterface(pythoncom.IID_IDispatch)) - if student_version: - self.logger.info(f"New AEDT {version_key} Student version process ID {proc[0]}.") - elif aedt_process_id: - self.logger.info(f"Existing AEDT session process ID {proc[0]}.") - else: - self.logger.info(f"New AEDT {version_key} Started process ID {proc[0]}.") - break - else: - self.logger.warning( - "PyAEDT is not supported in AEDT versions earlier than 2021 R2. Trying to launch PyAEDT with PyWin32." - ) - self._dispatch_win32(version) - # we should have a check here to see if AEDT is really started - self.is_grpc_api = False + Parameters + ---------- + project_name : str, optional + Project name. The default is ``None``, in which case the active + project is used. - def _initialize( - self, - machine="", - port=0, - non_graphical=False, - new_session=False, - version=None, - is_grpc=True, - ): - if not is_grpc: # pragma: no cover - from ansys.aedt.core.internal.clr_module import _clr + Returns + ------- + str + Path to the project. - _clr.AddReference("Ansys.Ansoft.CoreCOMScripting") - AnsoftCOMUtil = __import__("Ansys.Ansoft.CoreCOMScripting") - self.COMUtil = AnsoftCOMUtil.Ansoft.CoreCOMScripting.Util.COMUtil - StandalonePyScriptWrapper = AnsoftCOMUtil.Ansoft.CoreCOMScripting.COM.StandalonePyScriptWrapper - if non_graphical or new_session: - self.launched_by_pyaedt = True - return StandalonePyScriptWrapper.CreateObjectNew(non_graphical) - else: - return StandalonePyScriptWrapper.CreateObject(version) + """ + if not project_name: + oproject = self.active_project() else: - settings.use_grpc_api = True - self.is_grpc_api = True - base_path = settings.aedt_install_dir - sys.path.insert(0, base_path) - sys.path.insert(0, str(Path(base_path) / "PythonFiles" / "DesktopPlugin")) - if is_linux: - pyaedt_path = Path(__file__).parent - os.environ["PATH"] = str(pyaedt_path) + os.pathsep + os.environ["PATH"] - os.environ["DesktopPluginPyAEDT"] = str(Path(settings.aedt_install_dir) / "PythonFiles" / "DesktopPlugin") - launch_msg = f"AEDT installation Path {base_path}" - self.logger.info(launch_msg) - from ansys.aedt.core.internal.grpc_plugin_dll_class import AEDT - - if settings.use_multi_desktop: - os.environ["DesktopPluginPyAEDT"] = str( - Path(list(self.installed_versions.values())[0]) / "PythonFiles" / "DesktopPlugin" - ) - self.grpc_plugin = AEDT(os.environ["DesktopPluginPyAEDT"]) - oapp = self.grpc_plugin.CreateAedtApplication(machine, port, non_graphical, new_session) - if oapp: - self.isoutsideDesktop = True - self.aedt_process_id = self.odesktop.GetProcessID() - return True - - @property - def odesktop(self): - """AEDT instance containing all projects and designs. - - Examples - -------- - Get the COM object representing the desktop. - - >>> from ansys.aedt.core import Desktop - >>> d = Desktop() - >>> d.odesktop - """ - if settings.use_grpc_api: - tries = 0 - while tries < 5: - try: - self._odesktop = self.grpc_plugin.odesktop - return self._odesktop - except Exception: # pragma: no cover - tries += 1 - time.sleep(1) - return self._odesktop # pragma: no cover - - @odesktop.setter - def odesktop(self, val): # pragma: no cover - self._odesktop = val - - def _init_grpc(self, non_graphical, new_aedt_session, version, student_version, version_key): - if settings.remote_rpc_session: # pragma: no cover - settings.remote_api = True - if not self.machine: - try: - self.machine = settings.remote_rpc_session.server_name - except Exception: - pass - if not self.port: - try: - self.port = settings.remote_rpc_session.port - except Exception: - pass - if not self.machine or self.machine in [ - "localhost", - "127.0.0.1", - socket.getfqdn(), - socket.getfqdn().split(".")[0], - ]: - self.machine = "127.0.0.1" - else: # pragma: no cover - settings.remote_api = True - if not self.port: - if self.machine and self.machine != "127.0.0.1": # pragma: no cover - self.logger.error("New session of AEDT cannot be started on remote machine from Desktop Class.") - self.logger.error("Either use port argument or start an rpc session to start AEDT on remote machine.") - self.logger.error( - "Use client = ansys.aedt.core.common_rpc.client(machinename) to start a remote session." - ) - self.logger.error("Use client.aedt(port) to start aedt on remote machine before connecting.") - elif new_aedt_session: - self.port = _find_free_port() - self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") - else: - sessions = grpc_active_sessions( - version=version, student_version=student_version, non_graphical=non_graphical - ) - if sessions: - self.port = sessions[0] - if len(sessions) == 1: - self.logger.info(f"Found active AEDT gRPC session on port {self.port}.") - else: - self.logger.warning( - f"Multiple AEDT gRPC sessions are found. Setting the active session on port {self.port}." - ) - else: - if is_windows: # pragma: no cover - if com_active_sessions( - version=version, student_version=student_version, non_graphical=non_graphical - ): - # settings.use_grpc_api = False - self.logger.info("No AEDT gRPC found. Found active COM Sessions.") - return self._init_dotnet( - non_graphical, new_aedt_session, version, student_version, version_key - ) - self.port = _find_free_port() - self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") - new_aedt_session = True - elif new_aedt_session and not _is_port_occupied(self.port, self.machine): - self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") - elif new_aedt_session: - self.logger.warning("New Session of AEDT cannot be started on specified port because occupied.") - self.port = _find_free_port() - self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") - elif _is_port_occupied(self.port, self.machine): - self.logger.info(f"Connecting to AEDT session on gRPC port {self.port}.") - else: - self.logger.info(f"AEDT session is starting on gRPC port {self.port}.") - new_aedt_session = True - - if new_aedt_session and settings.use_lsf_scheduler and is_linux: # pragma: no cover - out, self.machine = launch_aedt_in_lsf(non_graphical, self.port) - if out: - self.launched_by_pyaedt = True - oApp = self._initialize( - is_grpc=True, machine=self.machine, port=self.port, new_session=False, version=version_key - ) - else: - self.logger.error(f"Failed to start LSF job on machine: {self.machine}.") - return - elif new_aedt_session: - installer = Path(settings.aedt_install_dir) / "ansysedt" - if student_version: # pragma: no cover - installer = Path(settings.aedt_install_dir) / "ansysedtsv" - if not is_linux: - if student_version: # pragma: no cover - installer = Path(settings.aedt_install_dir) / "ansysedtsv.exe" - else: - installer = Path(settings.aedt_install_dir) / "ansysedt.exe" - out, self.port = launch_aedt(installer, non_graphical, self.port, student_version) - oApp = self._initialize( - is_grpc=True, - non_graphical=non_graphical, - machine=self.machine, - port=self.port, - new_session=not out, - version=version_key, - ) - self.launched_by_pyaedt = True if oApp else False - if not self.launched_by_pyaedt: # pragma: no cover - self.logger.error("Failed to connect to AEDT using gRPC plugin.") - self.logger.error("Check installation, license and environment variables.") - else: - oApp = self._initialize( - is_grpc=True, - non_graphical=non_graphical, - machine=self.machine, - port=self.port, - new_session=new_aedt_session, - version=version_key, - ) - if oApp: - if new_aedt_session: - message = ( - f"{version}{' Student' if student_version else ''} version started " - f"with process ID {self.aedt_process_id}." - ) - self.logger.info(message) - - else: - self.logger.error("Failed to connect to AEDT using gRPC plugin.") - self.logger.error("Check installation, license and environment variables.") - - def _set_logger_file(self): - # Set up the log file in the AEDT project directory - if settings.logger_file_path: - self.logfile = settings.logger_file_path - else: - if settings.remote_api or settings.remote_rpc_session: - project_dir = tempfile.gettempdir() - elif self.odesktop: - project_dir = self.odesktop.GetProjectDirectory() - else: - project_dir = tempfile.gettempdir() - self.logfile = Path(project_dir) / f"pyaedt{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.log" - self._logger = AedtLogger(desktop=self) - return True - - @property - def messenger(self): - """Messenger manager for the AEDT logger.""" - return pyaedt_logger - - @property - def logger(self): - """AEDT logger.""" - return self._logger - - @pyaedt_function_handler() - def project_list(self): - """Get a list of projects. - - Returns - ------- - List - List of projects. - - """ - return list(self.odesktop.GetProjectList()) + oproject = self.active_project(project_name) + if oproject: + return oproject.GetPath() + return None @pyaedt_function_handler() - def analyze_all(self, project=None, design=None): # pragma: no cover - """Analyze all setups in a project. + def design_list(self, project=None): + """Get a list of the designs. Parameters ---------- project : str, optional - Project name. The default is ``None``, in which case the active project - is used. - design : str, optional - Design name. The default is ``None``, in which case all designs in - the project are analyzed. + Project name. The default is ``None``, in which case the active + project is used. Returns ------- - bool - ``True`` when successful, ``False`` when failed. + List + List of the designs. """ + + updateddeslist = [] if not project: oproject = self.active_project() else: oproject = self.active_project(project) if oproject: - if not design: - oproject.AnalyzeAll() - else: - odesign = self.active_design(oproject, design) - if odesign: - odesign.AnalyzeAll() - return True - - @pyaedt_function_handler() - def clear_messages(self): - """Clear all AEDT messages. - - Returns - ------- - bool - ``True`` when successful, ``False`` when failed. - """ - self.odesktop.ClearMessages("", "", 3) - return True + deslist = list(oproject.GetTopDesignList()) + for el in deslist: + m = re.search(r"[^;]+$", el) + updateddeslist.append(m.group(0)) + return updateddeslist @pyaedt_function_handler() - def save_project(self, project_name=None, project_path=None): - """Save the project. + def design_type(self, project_name=None, design_name=None): + """Get the type of design. Parameters ---------- project_name : str, optional - Project name. The default is ``None``, in which case the active project - is used. - project_path : str, optional - Full path to the project. The default is ``None``. If a path is - provided, ``save as`` is used. + Project name. The default is ``None``, in which case the active + project is used. + design_name : str, optional + Design name. The default is ``None``, in which case the active + design is used. Returns ------- - bool - ``True`` when successful, ``False`` when failed. + str + Design type. """ if not project_name: - oproject = self.odesktop.GetActiveProject() + oproject = self.active_project() else: - oproject = self.odesktop.SetActiveProject(project_name) - if project_path: - oproject.SaveAs(project_path, True) - else: - oproject.Save() - return True - - @pyaedt_function_handler() - def copy_design(self, project_name=None, design_name=None, target_project=None): - """Copy a design and paste it in an existing project or new project. - - .. deprecated:: 0.6.31 - Use :func:`copy_design_from` instead. - - Parameters - ---------- - project_name : str, optional - Project name. The default is ``None``, in which case the active project - is used. - design_name : str, optional - Design name. The default is ``None``. - target_project : str, optional - Target project. The default is ``None``. - - Returns - ------- - bool - ``True`` when successful, ``False`` when failed. - """ - if not project_name: # pragma: no cover - oproject = self.active_project() - else: # pragma: no cover - oproject = self.active_project(project_name) - if oproject: # pragma: no cover - if not design_name: - odesign = self.active_design(oproject) - else: - odesign = self.active_design(oproject, design_name) - if odesign: - oproject.CopyDesign(design_name) - if not target_project: - oproject.Paste() - return True - else: - oproject_target = self.active_project(target_project) - if not oproject_target: - oproject_target = self.odesktop.NewProject(target_project) - oproject_target.Paste() - return True - else: # pragma: no cover - return False - - @pyaedt_function_handler() - def project_path(self, project_name=None): - """Get the path to the project. - - Parameters - ---------- - project_name : str, optional - Project name. The default is ``None``, in which case the active - project is used. - - Returns - ------- - str - Path to the project. - - """ - if not project_name: - oproject = self.active_project() - else: - oproject = self.active_project(project_name) - if oproject: - return oproject.GetPath() - return None - - @pyaedt_function_handler() - def design_list(self, project=None): - """Get a list of the designs. - - Parameters - ---------- - project : str, optional - Project name. The default is ``None``, in which case the active - project is used. - - Returns - ------- - List - List of the designs. - """ - - updateddeslist = [] - if not project: - oproject = self.active_project() - else: - oproject = self.active_project(project) - if oproject: - deslist = list(oproject.GetTopDesignList()) - for el in deslist: - m = re.search(r"[^;]+$", el) - updateddeslist.append(m.group(0)) - return updateddeslist - - @pyaedt_function_handler() - def design_type(self, project_name=None, design_name=None): - """Get the type of design. - - Parameters - ---------- - project_name : str, optional - Project name. The default is ``None``, in which case the active - project is used. - design_name : str, optional - Design name. The default is ``None``, in which case the active - design is used. - - Returns - ------- - str - Design type. - """ - if not project_name: - oproject = self.active_project() - else: - oproject = self.active_project(project_name) - if not oproject: # pragma: no cover - return "" - if not design_name: - odesign = self.active_design(oproject) + oproject = self.active_project(project_name) + if not oproject: # pragma: no cover + return "" + if not design_name: + odesign = self.active_design(oproject) else: odesign = self.active_design(oproject, design_name) if odesign: @@ -1450,67 +1098,7 @@ def design_type(self, project_name=None, design_name=None): else: # pragma: no cover return "" - @property - def personallib(self): - """PersonalLib directory. - - Returns - ------- - str - Full absolute path for the ``PersonalLib`` directory. - - """ - return self.odesktop.GetPersonalLibDirectory() - - @property - def userlib(self): - """UserLib directory. - - Returns - ------- - str - Full absolute path for the ``UserLib`` directory. - - """ - return self.odesktop.GetUserLibDirectory() - - @property - def syslib(self): - """SysLib directory. - - Returns - ------- - str - Full absolute path for the ``SysLib`` directory. - - """ - return self.odesktop.GetLibraryDirectory() - - @property - def src_dir(self): - """Python source directory. - - Returns - ------- - str - Full absolute path for the ``python`` directory. - - """ - return Path(__file__) - - @property - def pyaedt_dir(self): - """PyAEDT directory. - - Returns - ------- - str - Full absolute path for the ``pyaedt`` directory. - - """ - return Path(__file__).parent - - def _exception(self, ex_value, tb_data): # pragma: no cover + def __exception(self, ex_value, tb_data): # pragma: no cover """Write the trace stack to AEDT when a Python error occurs. Parameters @@ -1570,6 +1158,97 @@ def load_project(self, project_file, design_name=None): else: # pragma: no cover return False + @pyaedt_function_handler() + def __close_aedt_application(self, close_desktop, pid, is_grpc_api): + """Release the AEDT API. + + Parameters + ---------- + desktop_class : :class:ansys.aedt.core.desktop.Desktop + Desktop class. + close_desktop : bool + Whether to close the active AEDT session. + pid : int + Process ID of the desktop app that is being closed. + is_grpc_api : bool + Whether the active AEDT session is gRPC or COM. + + Returns + ------- + bool + ``True`` when successful, ``False`` when failed. + + """ + if settings.remote_rpc_session or (settings.aedt_version >= "2022.2" and is_grpc_api): + if close_desktop and self.parent_desktop_id: # pragma: no cover + pyaedt_logger.error("A child desktop session is linked to this session.") + pyaedt_logger.error("Multiple desktop sessions must be released in reverse order.") + return False + elif close_desktop: + try: + if settings.use_multi_desktop: # pragma: no cover + os.kill(pid, 9) + else: + self.odesktop.QuitApplication() + if self.is_grpc_api: + self.grpc_plugin.Release() + timeout = 20 + while pid in active_sessions(): + time.sleep(1) + if timeout == 0: + os.kill(pid, 9) + break + timeout -= 1 + if _desktop_sessions: + for v in _desktop_sessions.values(): + if pid in v.parent_desktop_id: # pragma: no cover + del v.parent_desktop_id[v.parent_desktop_id.index(pid)] + return True + except Exception: # pragma: no cover + warnings.warn("Something went wrong closing AEDT. Exception in `_main.oDesktop.QuitApplication()`.") + else: # pragma: no cover + for k, d in _desktop_sessions.items(): + if k == pid: + d.grpc_plugin.recreate_application(True) + d.grpc_plugin.Release() + return True + elif not inside_desktop: # pragma: no cover + if close_desktop: + try: + if settings.use_multi_desktop: + self.odesktop.QuitApplication() + else: + os.kill(pid, 9) + except Exception: # pragma: no cover + warnings.warn("Something went wrong closing AEDT. Exception in `os.kill(pid, 9)`.") + return False + else: + try: + scopeID = 0 + while scopeID <= 5: + self.COMUtil.ReleaseCOMObjectScope(self.COMUtil.PInvokeProxyAPI, scopeID) + scopeID += 1 + except Exception: + pyaedt_logger.warning( + "Something went wrong releasing AEDT. Exception in `_main.COMUtil.ReleaseCOMObjectScope`." + ) + if not settings.remote_rpc_session and close_desktop: # pragma: no cover + timeout = 10 + while pid in active_sessions(): + time.sleep(1) + timeout -= 1 + if timeout == 0: + try: + os.kill(pid, 9) + return True + except Exception: # pragma: no cover + warnings.warn( + "Something went wrong closing AEDT. Exception in `os.kill(pid, 9)` after timeout." + ) + return False + + return True + @pyaedt_function_handler() def release_desktop(self, close_projects=True, close_on_exit=True): """Release AEDT. @@ -1618,7 +1297,8 @@ def release_desktop(self, close_projects=True, close_on_exit=True): self.odesktop.CloseProject(project) except Exception: # pragma: no cover self.logger.warning(f"Failed to close Project {project}") - result = _close_aedt_application(self, close_on_exit, self.aedt_process_id, self.is_grpc_api) + result = self.__close_aedt_application(close_on_exit, self.aedt_process_id, self.is_grpc_api) + if not result: # pragma: no cover self.logger.error("Error releasing desktop.") return False @@ -2084,6 +1764,11 @@ def submit_ansys_cloud_job( def get_ansyscloud_job_info(self, job_id=None, job_name=None): # pragma: no cover """Monitor a job submitted to Ansys Cloud. + .. warning:: + + Do not execute this function with untrusted environment variables. + See the :ref:`security guide` for details. + Parameters ---------- job_id : str, optional @@ -2095,7 +1780,7 @@ def get_ansyscloud_job_info(self, job_id=None, job_name=None): # pragma: no cov ------- dict - Examples + Examples -------- >>> from ansys.aedt.core import Desktop @@ -2114,15 +1799,22 @@ def get_ansyscloud_job_info(self, job_id=None, job_name=None): # pragma: no cov ... results_folder='via_gsg_results') >>> d.release_desktop(False,False) """ - command = Path(self.install_path) / "common" / "AnsysCloudCLI" / "AnsysCloudCli.exe" - + ansys_cloud_cli_path = Path(self.install_path) / "common" / "AnsysCloudCLI" / "AnsysCloudCli.exe" + if not Path(ansys_cloud_cli_path).exists(): + raise FileNotFoundError("Ansys Cloud CLI not found. Check the installation path.") + command = [ansys_cloud_cli_path] if job_name: - command = [command, "jobinfo", "-j", job_name] + command += ["jobinfo", "-j", job_name] elif job_id: - command = [command, "jobinfo", "-i", job_id] + command += ["jobinfo", "-i", job_id] cloud_info = Path(tempfile.gettempdir()) / generate_unique_name("job_info") - with open_file(cloud_info, "w") as outfile: - subprocess.Popen(" ".join(command), stdout=outfile).wait() + + try: + with open_file(cloud_info, "w") as outfile: + subprocess.run(command, stdout=outfile, check=True) # nosec + except subprocess.CalledProcessError as e: + raise AEDTRuntimeError("An error occurred while monitoring a job submitted to Ansys Cloud") from e + out = {} with open_file(cloud_info, "r") as infile: lines = infile.readlines() @@ -2192,6 +1884,11 @@ def select_scheduler( def get_available_cloud_config(self, region="westeurope"): # pragma: no cover """Get available Ansys Cloud machines configuration. + .. warning:: + + Do not execute this function with untrusted environment variables. + See the :ref:`security guide` for details. + Parameters ---------- region : str @@ -2223,12 +1920,17 @@ def get_available_cloud_config(self, region="westeurope"): # pragma: no cover ... results_folder='via_gsg_results') >>> d.release_desktop(False,False) """ - command = Path(self.install_path) / "common" / "AnsysCloudCLI" / "AnsysCloudCli.exe" + ansys_cloud_cli_path = Path(self.install_path) / "common" / "AnsysCloudCLI" / "AnsysCloudCli.exe" + if not Path(ansys_cloud_cli_path).exists(): + raise FileNotFoundError("Ansys Cloud CLI not found. Check the installation path.") ver = self.aedt_version_id.replace(".", "R") - command = [command, "getQueues", "-p", "AEDT", "-v", ver, "--details"] + command = [ansys_cloud_cli_path, "getQueues", "-p", "AEDT", "-v", ver, "--details"] cloud_info = Path(tempfile.gettempdir()) / generate_unique_name("cloud_info") - with open_file(cloud_info, "w") as outfile: - subprocess.Popen(" ".join(command), stdout=outfile).wait() + try: + with open_file(cloud_info, "w") as outfile: + subprocess.run(command, stdout=outfile, check=True) # nosec + except subprocess.CalledProcessError as e: + raise AEDTRuntimeError("An error occurred while monitoring a job submitted to Ansys Cloud") from e dict_out = {} with open_file(cloud_info, "r") as infile: @@ -2272,29 +1974,11 @@ def download_job_results(self, job_id, project_path, results_folder, filter="*") download_status = self.odesktop.DownloadJobResults(job_id, project_path, results_folder, filter) return True if download_status == 1 else False - @property - def are_there_simulations_running(self): - """Check if there are simulation running. - - .. note:: - It works only for AEDT >= ``"2023.2"``. - - Returns - ------- - float - - """ - if self.aedt_version_id > "2023.1": - return self.odesktop.AreThereSimulationsRunning() - return False - @pyaedt_function_handler() + @min_aedt_version("2023.2") def get_monitor_data(self): # pragma: no cover """Check and get monitor data of an existing analysis. - .. note:: - It works only for AEDT >= ``"2023.2"``. - Returns ------- dict @@ -2322,19 +2006,343 @@ def get_monitor_data(self): # pragma: no cover return counts @pyaedt_function_handler() + @min_aedt_version("2023.2") def stop_simulations(self, clean_stop=True): """Check if there are simulation running and stops them. - .. note:: - It works only for AEDT >= ``"2023.2"``. - Returns ------- str """ - if self.aedt_version_id > "2023.1": - return self.odesktop.StopSimulations(clean_stop) - else: - self.logger.error("It works only for AEDT >= `2023.2`.") - return False + return self.odesktop.StopSimulations(clean_stop) + + # ############################################# # + # Private methods # + # ############################################# # + + def __init_desktop(self): + # run it after the settings.non_graphical is set + self.pyaedt_version = __version__ + settings.aedt_version = self.odesktop.GetVersion()[0:6] + self.odesktop.RestoreWindow() + settings.aedt_install_dir = self.odesktop.GetExeDir() + + def __check_version(self, specified_version, student_version): + if self.current_version == "" and aedt_versions.latest_version == "": + raise AEDTRuntimeError("AEDT is not installed on your system. Install AEDT version 2022 R2 or higher.") + if not specified_version: + if student_version and self.current_student_version: + specified_version = self.current_student_version + elif student_version and self.current_version: + specified_version = self.current_version + student_version = False + self.logger.warning("AEDT Student Version not found on the system. Using regular version.") + else: + if self.current_version != "": + specified_version = self.current_version + else: + specified_version = aedt_versions.latest_version + if "SV" in specified_version: + student_version = True + self.logger.warning("Only AEDT Student Version found on the system. Using Student Version.") + elif student_version: + specified_version += "SV" + specified_version = get_string_version(specified_version) + + if float(specified_version[0:6]) < 2019: + raise ValueError("PyAEDT supports AEDT version 2021 R1 and later. Recommended version is 2022 R2 or later.") + elif float(specified_version[0:6]) < 2022.2: + warnings.warn( + """PyAEDT has limited capabilities when used with an AEDT version earlier than 2022 R2. + Update your AEDT installation to 2022 R2 or later.""" + ) + if not (specified_version in self.installed_versions) and not ( + specified_version + "CL" in self.installed_versions + ): + raise ValueError( + f"Specified version {specified_version[0:6]}{' Student Version' if student_version else ''} is not " + f"installed on your system" + ) + + version = "Ansoft.ElectronicsDesktop." + specified_version[0:6] + settings.aedt_install_dir = None + if specified_version in self.installed_versions: + settings.aedt_install_dir = self.installed_versions[specified_version] + if settings.remote_rpc_session: + try: + version = "Ansoft.ElectronicsDesktop." + settings.remote_rpc_session.aedt_version[0:6] + return settings.remote_rpc_session.student_version, settings.remote_rpc_session.aedt_version, version + except Exception: + return False, "", "" + + return student_version, specified_version, version + + def __run_student(self): # pragma: no cover + executable = Path(Path(settings.aedt_install_dir) / "ansysedtsv.exe").resolve(strict=True) + if not executable.exists(): + raise FileNotFoundError(f"Student version executable {executable} not found") + pid = subprocess.Popen([executable], creationflags=subprocess.DETACHED_PROCESS) # nosec + self.logger.debug(f"Running Electronic Desktop Student Version with PID {pid}.") + time.sleep(5) + + def __dispatch_win32(self, version): # pragma: no cover + from ansys.aedt.core.internal.clr_module import win32_client + + o_ansoft_app = win32_client.Dispatch(version) + self.odesktop = o_ansoft_app.GetAppDesktop() + self.isoutsideDesktop = True + + def __init_dotnet( + self, + non_graphical, + new_aedt_session, + version, + student_version, + version_key, + aedt_process_id=None, + ): # pragma: no cover + import pythoncom + + pythoncom.CoInitialize() + + if is_linux: + raise Exception( + "PyAEDT supports COM initialization in Windows only. To use in Linux, upgrade to AEDT 2022 R2 or later." + ) + base_path = settings.aedt_install_dir + sys.path.insert(0, base_path) + sys.path.insert(0, str(Path(base_path) / "PythonFiles" / "DesktopPlugin")) + launch_msg = f"AEDT installation Path {base_path}." + self.logger.info(launch_msg) + processID = [] + if is_windows: + processID = com_active_sessions(version, student_version, non_graphical) + if student_version and not processID: # Opens an instance if processID is an empty list + self.__run_student() + elif non_graphical or new_aedt_session or not processID: + # Force new object if no non-graphical instance is running or if there is not an already existing process. + self.__initialize(non_graphical=non_graphical, new_session=True, is_grpc=False, version=version_key) + else: + self.__initialize(new_session=False, is_grpc=False, version=version_key) + processID2 = [] + if is_windows: + processID2 = com_active_sessions(version, student_version, non_graphical) + proc = [i for i in processID2 if i not in processID] # Looking for the "new" process + if ( + not proc and (not new_aedt_session) and aedt_process_id + ): # if it isn't a new aedt session and a process ID is given + proc = [aedt_process_id] + elif not proc: + proc = processID2 + if proc == processID2 and len(processID2) > 1: + self.__dispatch_win32(version) + elif version_key >= "2021.2": + + context = pythoncom.CreateBindCtx(0) + running_coms = pythoncom.GetRunningObjectTable() + monikiers = running_coms.EnumRunning() + for monikier in monikiers: + m = re.search(version[10:] + r"\.\d:" + str(proc[0]), monikier.GetDisplayName(context, monikier)) + if m: + obj = running_coms.GetObject(monikier) + self.isoutsideDesktop = True + from ansys.aedt.core.internal.clr_module import win32_client + + self.odesktop = win32_client.Dispatch(obj.QueryInterface(pythoncom.IID_IDispatch)) + if student_version: + self.logger.info(f"New AEDT {version_key} Student version process ID {proc[0]}.") + elif aedt_process_id: + self.logger.info(f"Existing AEDT session process ID {proc[0]}.") + else: + self.logger.info(f"New AEDT {version_key} Started process ID {proc[0]}.") + break + else: + self.logger.warning( + "PyAEDT is not supported in AEDT versions earlier than 2021 R2. Trying to launch PyAEDT with PyWin32." + ) + self.__dispatch_win32(version) + # we should have a check here to see if AEDT is really started + self.is_grpc_api = False + + def __initialize( + self, + machine="", + port=0, + non_graphical=False, + new_session=False, + version=None, + is_grpc=True, + ): + if not is_grpc: # pragma: no cover + from ansys.aedt.core.internal.clr_module import _clr + + _clr.AddReference("Ansys.Ansoft.CoreCOMScripting") + AnsoftCOMUtil = __import__("Ansys.Ansoft.CoreCOMScripting") + self.COMUtil = AnsoftCOMUtil.Ansoft.CoreCOMScripting.Util.COMUtil + StandalonePyScriptWrapper = AnsoftCOMUtil.Ansoft.CoreCOMScripting.COM.StandalonePyScriptWrapper + if non_graphical or new_session: + self.launched_by_pyaedt = True + return StandalonePyScriptWrapper.CreateObjectNew(non_graphical) + else: + return StandalonePyScriptWrapper.CreateObject(version) + else: + settings.use_grpc_api = True + self.is_grpc_api = True + base_path = settings.aedt_install_dir + sys.path.insert(0, base_path) + sys.path.insert(0, str(Path(base_path) / "PythonFiles" / "DesktopPlugin")) + if is_linux: + pyaedt_path = Path(__file__).parent + os.environ["PATH"] = str(pyaedt_path) + os.pathsep + os.environ["PATH"] + os.environ["DesktopPluginPyAEDT"] = str(Path(settings.aedt_install_dir) / "PythonFiles" / "DesktopPlugin") + launch_msg = f"AEDT installation Path {base_path}" + self.logger.info(launch_msg) + from ansys.aedt.core.internal.grpc_plugin_dll_class import AEDT + + if settings.use_multi_desktop: + os.environ["DesktopPluginPyAEDT"] = str( + Path(list(self.installed_versions.values())[0]) / "PythonFiles" / "DesktopPlugin" + ) + self.grpc_plugin = AEDT(os.environ["DesktopPluginPyAEDT"]) + oapp = self.grpc_plugin.CreateAedtApplication(machine, port, non_graphical, new_session) + if oapp: + self.isoutsideDesktop = True + self.aedt_process_id = self.odesktop.GetProcessID() + return True + + def __init_grpc(self, non_graphical, new_aedt_session, version, student_version, version_key): + if settings.remote_rpc_session: # pragma: no cover + settings.remote_api = True + if not self.machine: + try: + self.machine = settings.remote_rpc_session.server_name + except Exception: + self.logger.debug("Failed to retrieve server name from RPyC connection") + if not self.port: + try: + self.port = settings.remote_rpc_session.port + except Exception: + self.logger.debug("Failed to retrieve port from RPyC connection") + if not self.machine or self.machine in [ + "localhost", + "127.0.0.1", + socket.getfqdn(), + socket.getfqdn().split(".")[0], + ]: + self.machine = "127.0.0.1" + else: # pragma: no cover + settings.remote_api = True + if not self.port: + if self.machine and self.machine != "127.0.0.1": # pragma: no cover + self.logger.error("New session of AEDT cannot be started on remote machine from Desktop Class.") + self.logger.error("Either use port argument or start an rpc session to start AEDT on remote machine.") + self.logger.error( + "Use client = ansys.aedt.core.common_rpc.client(machinename) to start a remote session." + ) + self.logger.error("Use client.aedt(port) to start aedt on remote machine before connecting.") + elif new_aedt_session: + self.port = _find_free_port() + self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") + else: + sessions = grpc_active_sessions( + version=version, student_version=student_version, non_graphical=non_graphical + ) + if sessions: + self.port = sessions[0] + if len(sessions) == 1: + self.logger.info(f"Found active AEDT gRPC session on port {self.port}.") + else: + self.logger.warning( + f"Multiple AEDT gRPC sessions are found. Setting the active session on port {self.port}." + ) + else: + if is_windows: # pragma: no cover + if com_active_sessions( + version=version, student_version=student_version, non_graphical=non_graphical + ): + # settings.use_grpc_api = False + self.logger.info("No AEDT gRPC found. Found active COM Sessions.") + return self.__init_dotnet( + non_graphical, new_aedt_session, version, student_version, version_key + ) + self.port = _find_free_port() + self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") + new_aedt_session = True + elif new_aedt_session and not _is_port_occupied(self.port, self.machine): + self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") + elif new_aedt_session: + self.logger.warning("New Session of AEDT cannot be started on specified port because occupied.") + self.port = _find_free_port() + self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") + elif _is_port_occupied(self.port, self.machine): + self.logger.info(f"Connecting to AEDT session on gRPC port {self.port}.") + else: + self.logger.info(f"AEDT session is starting on gRPC port {self.port}.") + new_aedt_session = True + + if new_aedt_session and settings.use_lsf_scheduler and is_linux: # pragma: no cover + out, self.machine = launch_aedt_in_lsf(non_graphical, self.port) + if out: + self.launched_by_pyaedt = True + oApp = self.__initialize( + is_grpc=True, machine=self.machine, port=self.port, new_session=False, version=version_key + ) + else: + self.logger.error(f"Failed to start LSF job on machine: {self.machine}.") + return + elif new_aedt_session: + installer = Path(settings.aedt_install_dir) / "ansysedt" + if student_version: # pragma: no cover + installer = Path(settings.aedt_install_dir) / "ansysedtsv" + if not is_linux: + if student_version: # pragma: no cover + installer = Path(settings.aedt_install_dir) / "ansysedtsv.exe" + else: + installer = Path(settings.aedt_install_dir) / "ansysedt.exe" + + out, self.port = launch_aedt(installer, non_graphical, self.port, student_version) + self.launched_by_pyaedt = True + oApp = self.__initialize( + is_grpc=True, + non_graphical=non_graphical, + machine=self.machine, + port=self.port, + new_session=not out, + version=version_key, + ) + else: + oApp = self.__initialize( + is_grpc=True, + non_graphical=non_graphical, + machine=self.machine, + port=self.port, + new_session=new_aedt_session, + version=version_key, + ) + if oApp: + if new_aedt_session: + message = ( + f"{version}{' Student' if student_version else ''} version started " + f"with process ID {self.aedt_process_id}." + ) + self.logger.info(message) + + else: + self.logger.error("Failed to connect to AEDT using gRPC plugin.") + self.logger.error("Check installation, license and environment variables.") + + def __set_logger_file(self): + # Set up the log file in the AEDT project directory + if settings.logger_file_path: + self.logfile = settings.logger_file_path + else: + if settings.remote_api or settings.remote_rpc_session: + project_dir = tempfile.gettempdir() + elif self.odesktop: + project_dir = self.odesktop.GetProjectDirectory() + else: + project_dir = tempfile.gettempdir() + self.logfile = Path(project_dir) / f"pyaedt{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.log" + self._logger = AedtLogger(desktop=self) + return True diff --git a/src/ansys/aedt/core/icepak.py b/src/ansys/aedt/core/icepak.py index 42b5407ddec..a0d02c8ac43 100644 --- a/src/ansys/aedt/core/icepak.py +++ b/src/ansys/aedt/core/icepak.py @@ -28,9 +28,9 @@ import os from pathlib import Path import re +import subprocess # nosec import warnings -import ansys.aedt.core from ansys.aedt.core.application.analysis_icepak import FieldAnalysisIcepak from ansys.aedt.core.generic.constants import SOLUTIONS from ansys.aedt.core.generic.data_handlers import _arg2dict @@ -3013,12 +3013,16 @@ def generate_fluent_mesh( else: fl_ucommand = ["bash"] + fl_ucommand + ['"' + fl_uscript_file_pointer + '"'] self.logger.info(" ".join(fl_ucommand)) - ansys.aedt.core.desktop.run_process(fl_ucommand) + try: + subprocess.run(fl_ucommand, check=True) # nosec + except subprocess.CalledProcessError as e: + raise AEDTRuntimeError("An error occurred while creating Fluent mesh") from e + if Path(mesh_file_pointer).exists(): self.logger.info("'" + mesh_file_pointer + "' has been created.") return self.mesh.assign_mesh_from_file(object_lists, mesh_file_pointer) - raise AEDTRuntimeError("Failed to create mesh file") + raise AEDTRuntimeError("Failed to create Fluent mesh file") @pyaedt_function_handler() def apply_icepak_settings( diff --git a/tests/system/general/conftest.py b/tests/system/general/conftest.py index 3be9d07b553..496d9b6ebfe 100644 --- a/tests/system/general/conftest.py +++ b/tests/system/general/conftest.py @@ -42,6 +42,7 @@ """ +import gc import json import os import random @@ -71,7 +72,6 @@ from ansys.aedt.core import Hfss from ansys.aedt.core.aedt_logger import pyaedt_logger from ansys.aedt.core.desktop import Desktop -from ansys.aedt.core.desktop import _delete_objects from ansys.aedt.core.generic.file_utils import generate_unique_name from ansys.aedt.core.internal.desktop_sessions import _desktop_sessions from ansys.aedt.core.internal.filesystem import Scratch @@ -127,6 +127,16 @@ def generate_random_ident(): return ident +def _delete_objects(): + settings.remote_api = False + pyaedt_logger.remove_all_project_file_logger() + try: + del sys.modules["glob"] + except Exception: + logger.debug("Failed to delete glob module") + gc.collect() + + @pytest.fixture(scope="session", autouse=True) def init_scratch(): test_folder_name = "unit_test" + generate_random_ident() diff --git a/tests/unit/test_desktop.py b/tests/unit/test_desktop.py index fbf62c923b5..f5984d969d2 100644 --- a/tests/unit/test_desktop.py +++ b/tests/unit/test_desktop.py @@ -23,19 +23,32 @@ # SOFTWARE. import socket +from unittest.mock import MagicMock +from unittest.mock import PropertyMock from unittest.mock import patch from ansys.aedt.core.desktop import Desktop +from ansys.aedt.core.desktop import _check_port +from ansys.aedt.core.desktop import _check_settings from ansys.aedt.core.desktop import _find_free_port from ansys.aedt.core.desktop import _is_port_occupied -from ansys.aedt.core.desktop import run_process +from ansys.aedt.core.generic.settings import Settings +from ansys.aedt.core.internal.errors import AEDTRuntimeError import pytest @pytest.fixture(scope="module", autouse=True) -def desktop(): - """Override the desktop fixture to DO NOT open the Desktop when running this test class""" - return +def mock_desktop(): + """Fixture used to mock the creation of a Desktop instance.""" + with patch("ansys.aedt.core.desktop.Desktop.__init__", lambda x: None): + yield + + +@pytest.fixture(scope="module", autouse=True) +def mock_settings(): + """Fixture used to mock the creation of a Settings instance.""" + with patch("ansys.aedt.core.generic.settings.Settings.__init__", lambda x: None): + yield # Test _is_port_occupied @@ -60,16 +73,107 @@ def test_find_free_port(mock_socket, mock_active_sessions): assert port == 12345 -# Test run_process -@patch("subprocess.call") -def test_run_process(mock_call): - command = "dummy_command" - run_process(command) - mock_call.assert_called_once_with(command) - - # Test Desktop.get_available_toolkits() static method def test_get_available_toolkits(): toolkits = Desktop.get_available_toolkits() result = ["Circuit", "HFSS", "HFSS3DLayout", "Icepak", "Maxwell3D", "Project", "TwinBuilder"] all(elem in toolkits for elem in result) + + +@patch.object(Settings, "use_grpc_api", new_callable=lambda: True) +@patch("time.sleep", return_value=None) +def test_desktop_odesktop_retries(mock_settings, mock_sleep): + """Test Desktop.odesktop property retries to get the odesktop object.""" + desktop = Desktop() + desktop.grpc_plugin = MagicMock() + aedt_app = MagicMock() + mock_odesktop = PropertyMock(name="oui", side_effect=[Exception("Failure"), aedt_app]) + # NOTE: Use of type(...) is required for odesktop to be defined as a property and + # not an attribute. Without it, the side effect does not work. + type(desktop.grpc_plugin).odesktop = mock_odesktop + + assert aedt_app == desktop.odesktop + assert mock_odesktop.call_count == 2 + + +def test_desktop_odesktop_setter(): + """Test Desktop.odesktop property retries to get the odesktop object.""" + desktop = Desktop() + aedt_app = MagicMock() + + desktop.odesktop = aedt_app + + assert desktop._odesktop == aedt_app + + +def test_desktop_check_setttings_failure_with_lsf_num_cores(mock_settings): + """Test _check_setttings failure due to lsf_num_cores value.""" + settings = Settings() + settings.lsf_num_cores = -1 + + with pytest.raises(ValueError): + _check_settings(settings) + + +def test_desktop_check_setttings_failure_with_lsf_ram(mock_settings): + """Test _check_setttings failure due to lsf_ram value.""" + settings = Settings() + settings.lsf_num_cores = 1 + settings.lsf_ram = -1 + + with pytest.raises(ValueError): + _check_settings(settings) + + +def test_desktop_check_setttings_failure_with_lsf_aedt_command(mock_settings): + """Test _check_setttings failure due to lsf_aedt_command value.""" + settings = Settings() + settings.lsf_num_cores = 1 + settings.lsf_ram = 1 + settings.lsf_aedt_command = None + + with pytest.raises(ValueError, match="Invalid LSF AEDT command."): + _check_settings(settings) + + +def test_desktop_check_port_failure(): + """Test _check_port failure.""" + port = "twelve" + + with pytest.raises(ValueError): + _check_port(port) + + +@patch("ansys.aedt.core.desktop.aedt_versions") +def test_desktop_check_version_failure(mock_aedt_versions, mock_desktop): + mock_specified_version = MagicMock() + mock_student_version = MagicMock() + mock_aedt_versions.latest_version = "" + mock_aedt_versions.current_version = "" + desktop = Desktop() + + with pytest.raises( + AEDTRuntimeError, match="AEDT is not installed on your system. Install AEDT version 2022 R2 or higher." + ): + desktop._Desktop__check_version(mock_specified_version, mock_student_version) + + +@patch("ansys.aedt.core.desktop.aedt_versions") +def test_desktop_check_version_failure_with_old_specified_version(mock_aedt_versions, mock_desktop): + mock_student_version = MagicMock() + desktop = Desktop() + specified_version = "1989.6" + + with pytest.raises( + ValueError, match="PyAEDT supports AEDT version 2021 R1 and later. Recommended version is 2022 R2 or later." + ): + desktop._Desktop__check_version(specified_version, mock_student_version) + + +@patch("ansys.aedt.core.desktop.aedt_versions") +def test_desktop_check_version_failure_with_unknown_specified_version(mock_aedt_versions, mock_desktop): + desktop = Desktop() + specified_version = "2022.6" + + with pytest.raises(ValueError, match=f"Specified version {specified_version} is not installed on your system"): + desktop._Desktop__check_version(specified_version, False)