diff --git a/doc/changelog.d/7918.fixed.md b/doc/changelog.d/7918.fixed.md new file mode 100644 index 000000000000..a26b1fc05550 --- /dev/null +++ b/doc/changelog.d/7918.fixed.md @@ -0,0 +1 @@ +Connect to existing student version diff --git a/doc/source/User_guide/desktop_sessions.rst b/doc/source/User_guide/desktop_sessions.rst index 744d1828fc59..ff9d5772826f 100644 --- a/doc/source/User_guide/desktop_sessions.rst +++ b/doc/source/User_guide/desktop_sessions.rst @@ -82,7 +82,7 @@ If needed, you can still override the default behavior explicitly: # The AEDT session remains open here. Use ``Desktop`` class directly --------------------- +------------------------------ When ``Desktop`` is used directly, the default behavior depends on whether PyAEDT starts or attaches to AEDT. You can also release the desktop explicitly for finer control: @@ -108,3 +108,35 @@ Recommendations - Use ``with Desktop(...)`` when you want predictable cleanup. - Use direct ``Desktop(...)`` construction when you need more manual control over the AEDT session lifecycle. - When attaching to an existing AEDT session, consider leaving ``close_on_exit`` unset or setting it explicitly to ``True`` if you do want PyAEDT to close that session. + + +Session selection precedence +---------------------------- + +When ``Desktop`` decides whether to connect to an existing AEDT session or to start a new one, +PyAEDT evaluates several inputs in a specific order. The following list describes the exact +checks performed by the library (this order matches the implementation in +``ansys.aedt.core.desktop._validate_port`` and related initialization logic): + +- If ``port`` is ``0``: a concrete port is assigned (``_assign_port``) and used. +- If ``new_desktop`` is ``True``: PyAEDT prefers to start a new AEDT instance. If the + requested port is already in use by any active session, PyAEDT chooses a different + free port (``_find_free_port``) so the new session can be started. +- If a remote RPyC RPC connection is configured (``settings.remote_rpc_session``): PyAEDT + uses the remote session and short-circuits further local port checks (``new_desktop`` is + set to ``False`` and the requested port is used). +- If there is an active session for the same AEDT version and the requested port is used by + that session, PyAEDT connects to it (reuse). +- If there is an active session for the same AEDT version but the opposite display mode + (graphical vs non-graphical) using the requested port, PyAEDT flips the ``non_graphical`` + flag and connect to that session. +- If the requested port is in use by a different AEDT version, PyAEDT treats this as a + conflict and (to avoid attaching to the wrong version) select a new free port and start + a new session (``new_desktop`` becomes ``True``). +- If none of the above conditions apply, PyAEDT uses the requested port and start a + new AEDT session. + +This precedence ensures predictable behavior: version and the desire to force a new +session (``new_desktop``) govern whether PyAEDT attaches or starts, while port and display +mode determine whether an existing session can be reused or whether a new one must be +created. diff --git a/src/ansys/aedt/core/desktop.py b/src/ansys/aedt/core/desktop.py index 54455ae4ac0a..fe509e7468e7 100644 --- a/src/ansys/aedt/core/desktop.py +++ b/src/ansys/aedt/core/desktop.py @@ -65,6 +65,7 @@ from ansys.aedt.core.generic.general_methods import _is_version_format_valid from ansys.aedt.core.generic.general_methods import _normalize_version_to_string from ansys.aedt.core.generic.general_methods import active_sessions +from ansys.aedt.core.generic.general_methods import all_active_sessions from ansys.aedt.core.generic.general_methods import com_active_sessions from ansys.aedt.core.generic.general_methods import grpc_active_sessions from ansys.aedt.core.generic.general_methods import inside_desktop_ironpython_console @@ -437,10 +438,7 @@ def launch_aedt( timeout = settings.desktop_launch_timeout start = time.time() while timeout > 0: - if is_grpc_session_active( - port, - host, - ): + if is_grpc_session_active(port, host, student_version): break timeout -= 1 time.sleep(1) @@ -799,7 +797,7 @@ def __init__( ) self.__close_on_exit_arg = close_on_exit self.__machine = machine if machine else None - self.__port = port + self.__port = port if port is not None else 0 self.__is_grpc_api = True self.__student_version = False self.__aedt_version_string = "" @@ -1030,8 +1028,6 @@ def port(self) -> int: >>> d.port """ - if not self.__port: - self._assign_port() return self.__port @port.setter @@ -3085,39 +3081,75 @@ def _check_machine(self) -> None: self.machine = "127.0.0.1" @pyaedt_function_handler() - def _validate_port(self, port, machine=None): + def _validate_port( + self, + ): """Validate the specified gRPC port. On top of checking the port, this method also determines if a new AEDT session needs to be launched. """ - self.logger.debug(f"Validating specified gRPC port: {port}") - if port == 0: - return port - active_ports = is_grpc_session_active(port, machine) - if self.new_desktop and active_ports: - self.logger.warning(f"Port {port} is already in use. Finding a new free port.") - return _find_free_port() - elif not settings.remote_rpc_session and not self.new_desktop and not active_ports: - self.logger.warning(f"No active AEDT gRPC session found on port {port}. Opening a new AEDT session.") + self.logger.debug(f"Validating specified gRPC port: {self.port}") + + if self.port == 0: # Checking if available session is there or eventually assign new port + self._assign_port() + return self.port + all_sessions = all_active_sessions() + version = self.aedt_version_id[2:4] + self.aedt_version_id[5] + version += "_nongraphical" if self.non_graphical else "_graphical" + version += "_student" if self.student_version else "" + + version_neg = self.aedt_version_id[2:4] + self.aedt_version_id[5] + version_neg += "_nongraphical" if not self.non_graphical else "_graphical" + version_neg += "_student" if self.student_version else "" + + if self.new_desktop: + for el in all_sessions.values(): + if self.port in el.values(): + self.logger.warning(f"Port {self.port} is already in use. Finding a new free port.") + self.port = _find_free_port() + break + return self.port + elif settings.remote_rpc_session: # remote session -> no port check + self.logger.warning(f"Remote session found on port {self.port}. Using it.") + self.new_desktop = False + return self.port + elif version in all_sessions and self.port in all_sessions[version].values(): + self.logger.info(f"Port {self.port} session has been found.") + return self.port + elif version_neg in all_sessions and self.port in all_sessions[version_neg].values(): + mode = "graphical" if self.non_graphical else "non_graphical" + self.logger.warning(f"Port {self.port} is already in use in {mode} mode. Using it.") + self.non_graphical = not self.non_graphical + return self.port + else: + for el in all_sessions.values(): + if self.port in el.values(): + self.logger.warning( + f"Port {self.port} is already in use by another AEDT version. Finding a new free port." + ) + self.new_desktop = True + self.port = _find_free_port() + return self.port + # No active sessions found, open a new AEDT session self.new_desktop = True - return port + return self.port @pyaedt_function_handler() def _assign_port(self): - self.__port = 0 + self.port = 0 if settings.remote_rpc_session: self.logger.warning( "Remote AEDT connection without specified port. Trying to use the port from the RPyC connection." ) try: - self.__port = settings.remote_rpc_session.port - except Exception: + self.port = settings.remote_rpc_session.port + except Exception: # pragma: no cover self.logger.debug("Failed to retrieve port from RPyC connection") raise Exception("Failed to retrieve port from RPyC connection") - if settings.use_multi_desktop or self.new_desktop: - self.__port = _find_free_port() + elif settings.use_multi_desktop or self.new_desktop: + self.port = _find_free_port() self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") else: @@ -3127,7 +3159,7 @@ def _assign_port(self): non_graphical=self.non_graphical, ) if sessions: - self.__port = sessions[0] + self.port = sessions[0] if len(sessions) == 1: self.logger.info(f"Found active AEDT gRPC session on port {self.port}.") else: @@ -3135,7 +3167,7 @@ def _assign_port(self): f"Multiple AEDT gRPC sessions are found. Setting the active session on port {self.port}." ) else: - self.__port = _find_free_port() + self.port = _find_free_port() self.logger.info(f"New AEDT session is starting on gRPC port {self.port}.") self.new_desktop = True @@ -3285,17 +3317,8 @@ def __init_grpc(self): lock_file = self._on_ci_generate_lock_file() # Validate port availability/compatibility - try: - self.__port = self._validate_port(self.port) - except Exception: - # NOTE: When we can't validate the port and are not in a - # remote RPC session, we try to launch a new instance by default. - self.logger.warning(f"Could not validate port {self.port}") - if not settings.remote_rpc_session: - self.logger.info("Opening a new AEDT session.") - self.new_desktop = True + self._validate_port() - self.__port = self._validate_port(self.port, self.machine) is_launched = True # Launch new AEDT instance if needed if self.new_desktop: diff --git a/src/ansys/aedt/core/generic/general_methods.py b/src/ansys/aedt/core/generic/general_methods.py index a416c2b43731..5b2c665abe6d 100644 --- a/src/ansys/aedt/core/generic/general_methods.py +++ b/src/ansys/aedt/core/generic/general_methods.py @@ -32,6 +32,7 @@ import itertools import logging import os +import pathlib import platform import re import shutil @@ -54,6 +55,7 @@ from ansys.aedt.core.aedt_logger import pyaedt_logger from ansys.aedt.core.base import PyAedtBase from ansys.aedt.core.generic.numbers_utils import _units_assignment +from ansys.aedt.core.generic.numbers_utils import is_number from ansys.aedt.core.generic.settings import settings from ansys.aedt.core.internal.errors import AEDTRuntimeError from ansys.aedt.core.internal.errors import GrpcApiError @@ -1292,7 +1294,9 @@ def _is_port_occupied(port, host=None): @pyaedt_function_handler() -def is_grpc_session_active(port: int, machine: str | None = None) -> bool: +def is_grpc_session_active( + port: int, machine: str | None = None, student_version: bool = False, version=None, non_graphical=None +) -> bool: """Check if a gRPC session is active on the specified port. This function verifies whether an AEDT session is actively listening on @@ -1310,6 +1314,13 @@ def is_grpc_session_active(port: int, machine: str | None = None) -> bool: The gRPC port number to check. machine : str, optional Specific machine IP address. + student_version : bool, optional + Whether to search for student version sessions (ansysedtsv). The default is ``False``. + When ``True``, searches for ``ansysedtsv.exe`` or ``ansysedtsv`` processes. + version : str, optional + Specific AEDT version. + non_graphical : bool, optional + Whether to search for graphical or non-graphical version. The default is ``None``. Returns ------- @@ -1338,7 +1349,12 @@ def is_grpc_session_active(port: int, machine: str | None = None) -> bool: if machine and machine not in ["localhost", "127.0.0.1", "::ffff:127.0.0.1", socket.gethostname()]: return _is_port_occupied(port, machine) - return True if port in active_sessions().values() else False + return ( + True + if port + in active_sessions(version=version, student_version=student_version, non_graphical=non_graphical).values() + else False + ) @pyaedt_function_handler() @@ -1492,6 +1508,109 @@ def active_sessions( return return_dict_filtered +@pyaedt_function_handler() +def all_active_sessions() -> dict[str, dict]: + """Get information for active AEDT sessions. + + This function detects running AEDT processes and identifies their gRPC ports or + marks them as COM sessions. It works on both Windows and Linux platforms by using + multiple detection strategies to ensure reliable session discovery. + + Detection Strategy (in order of execution): + 1. **Process Discovery**: Searches for AEDT processes (ansysedt.exe or ansysedtsv.exe). + 2. **Command-Line Parsing**: Extracts gRPC port from ``-grpcsrv`` command-line argument. + 3. **Unix gRPC Analysis** (Linux only): Uses ``ss -Hnlp`` to find ports from socket files. + 4. **TCP Connection Analysis**: Falls back to checking active TCP connections via psutil. + + Port Detection Results: + - Positive integer, gRPC session on that port. + - ``-1``: COM session (no gRPC server running). + + + Returns + ------- + dict[str, dict] + Dictionary mapping AEDT process IDs to their corresponding ports. + Port is set to ``-1`` if the session is using COM instead of gRPC. + + Examples + -------- + Get all active AEDT sessions (any version, any mode): + + + """ + # Step 1: Determine target process names based on version type and operating system + # Student version uses different executable names (ansysedtsv vs ansysedt) + + targets = ["ansysedtsv.exe", "ansysedt.exe"] + + # Step 3: Get all matching AEDT processes from the system + # Returns list of tuples: [(pid, command_line_args), ...] + + target_processes = _get_target_processes(targets) + + # Step 4: AEDT processes launched + return_dict = {pid: -1 for pid, _ in target_processes} + + # Step 5: On Linux, try to resolve unknown ports using Unix socket analysis + # In Linux, running AEDT locally uses Unix domain sockets with filenames containing port numbers + # Example socket: AnsysEMUDS-50051.sock + if is_linux and any(port == -1 for port in return_dict.values()): + try: + # Run 'ss -Hnlp' command to get Unix socket information + sockets = _run_ss() # Returns {pid: port} mapping from socket filenames + + # Update return_dict with discovered ports + for pid, port in sockets.items(): + # Only update if PID is in our results and port is still unknown (-1) + if pid in return_dict and return_dict[pid] == -1: + return_dict[pid] = port + except Exception as e: + # Log but don't fail - we have other detection methods + pyaedt_logger.debug(f"Failed to analyze Unix sockets for port detection: {str(e)}") + + # Get all TCP connections for our AEDT processes + connections = _check_psutil_connections(list(return_dict.keys())) + return_dict_filtered = {} + for pid, port in return_dict.items(): + cmdline = "" + if pid in connections and len(connections[pid]) > 0 and "cmdline" in connections[pid][0]: + cmdline = connections[pid][0]["cmdline"] + + version = [i[1:] for i in pathlib.Path(cmdline).parts if i.startswith("v") and is_number(i[1:])] + + if version: + version = version[0] + flag_present = "nongraphical" if "-ng" in cmdline else "graphical" + version += f"_{flag_present}" + + if "ansysedtsv" in cmdline: + version += "_student" + + if version not in return_dict_filtered: + return_dict_filtered[version] = {pid: port} + else: + return_dict_filtered[version][pid] = port + else: + pyaedt_logger.debug( + f"Failed to retrieve AEDT version, the version should be included in the command line: {cmdline}." + ) + + # Step 6: Fallback method - Try to find ports by checking TCP network connections + for version, sessions in return_dict_filtered.items(): + if any(port == -1 for port in sessions.values()): + for pid in [i for i, v in sessions.items() if v == -1]: + version_number = version.replace("_student", "").replace("_nongraphical", "").replace("_graphical", "") + sessions[pid] = _check_connection_grpc_port( + connections, + pid, + version_number, + True if "nongraphical" in version else False, + ) + + return return_dict_filtered + + @pyaedt_function_handler() def com_active_sessions( version: str | None = None, student_version: bool | None = False, non_graphical: bool | None = False diff --git a/tests/unit/test_validate_port_sequence.py b/tests/unit/test_validate_port_sequence.py new file mode 100644 index 000000000000..d0ca11e327b8 --- /dev/null +++ b/tests/unit/test_validate_port_sequence.py @@ -0,0 +1,294 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2021 - 2026 Synopsys, Inc. and ANSYS, Inc. All rights reserved. +# 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. + +from unittest.mock import MagicMock +from unittest.mock import patch + +import pytest + +from ansys.aedt.core.desktop import Desktop +from ansys.aedt.core.generic.settings import Settings + +ACTIVE_SESSIONS = { + "261_graphical": {"pid_1": 50051, "pid_2": 50052}, + "261_nongraphical": {"pid_3": 50053, "pid_4": 50054}, + "252_graphical": {"pid_5": 50055, "pid_6": 50056}, + "252_nongraphical": {"pid_7": 50057, "pid_8": 50058}, + "252_graphical_student": {"pid_9": 50059}, + "252_nongraphical_student": {"pid_10": 50060}, +} + + +class _SmallLogger: + def debug(self, *a, **k): + pass + + def info(self, *a, **k): + pass + + def warning(self, *a, **k): + pass + + +def _make_desktop(port=0, version="2026.1", student_version=False, non_graphical=True, new_desktop=False): + d = Desktop.__new__(Desktop) + d._Desktop__port = port + d._Desktop__machine = "127.0.0.1" + d._Desktop__aedt_version_id = version + d._Desktop__student_version = student_version + d._Desktop__non_graphical = non_graphical + d._Desktop__new_desktop = new_desktop + # Use MagicMock for logger so tests can assert logging calls + d._Desktop__logger = MagicMock() + d._Desktop__close_on_exit = False + # Minimal attributes to prevent __del__ from failing in tests/debugger + d._Desktop__closed = False + d._Desktop__aedt_process_id = None + # Whether this Desktop instance uses gRPC API (avoid missing attribute in __del__) + d._Desktop__is_grpc_api = True + d.odesktop = None + d.grpc_plugin = MagicMock() + d.grpc_plugin.recreate_application = MagicMock() + return d + + +@pytest.fixture +def mock_settings(monkeypatch): + m = MagicMock(spec=Settings) + m.remote_rpc_session = None + m.aedt_version = "2026.1" + m.enable_desktop_logs = False + m.enable_file_logs = False + m.enable_screen_logs = False + m.use_multi_desktop = False + monkeypatch.setattr("ansys.aedt.core.desktop.settings", m, raising=False) + return m + + +def test_new_session_no_active_sessions(mock_settings): + """New AEDT session on port 50051, no active sessions.""" + base_port = 50051 + + # Start 2026.1 graphical session (no active sessions yet) + d1 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=True) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + assert d1._validate_port() == base_port + + # Start 2026.1 non-graphical session (no active sessions yet), new_desktop is False, and PyAEDT will flip it + d2 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + assert d2._validate_port() == base_port + assert d2.new_desktop + + +def test_new_session_port_0(mock_settings): + """New AEDT session on port 0.""" + base_port = 0 + random_port = 12345 + + # No sessions + + # Start 2026.1 + d1 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=True) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + assert d1._validate_port() == 12345 + d1._Desktop__logger.info.assert_called_with("New AEDT session is starting on gRPC port 12345.") + + # Start 2026.1, new_desktop False + d2 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + assert d2._validate_port() == random_port + assert d2.new_desktop + + # Start 2026.1, remote_rpc_session + mock_settings.remote_rpc_session = MagicMock() + mock_settings.remote_rpc_session.port = random_port + d3 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + assert d3._validate_port() == random_port + d3._Desktop__logger.warning.assert_called_with( + "Remote AEDT connection without specified port. Trying to use the port from the RPyC connection." + ) + mock_settings.remote_rpc_session = None + + # Two sessions in different versions + + # Start 2026.1 + d4 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=True) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value=ACTIVE_SESSIONS): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + assert d4._validate_port() == random_port + d4._Desktop__logger.info.assert_called_with(f"New AEDT session is starting on gRPC port {random_port}.") + + # Start 2026.1, new_desktop False + d5 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value=ACTIVE_SESSIONS): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + assert d5._validate_port() == random_port + assert d5.new_desktop + + # Ensure mock settings enables multi desktop + mock_settings.use_multi_desktop = True + + d = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + res = d._validate_port() + assert res == random_port + d._Desktop__logger.info.assert_called_with(f"New AEDT session is starting on gRPC port {random_port}.") + mock_settings.use_multi_desktop = False + + +def test_new_session(mock_settings): + """New AEDT session.""" + base_port = 50051 + random_port = 12345 + + # No sessions + + # Start 2026.1 + d1 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=True) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + assert d1._validate_port() == base_port + + # Start 2026.1, new_desktop False + d2 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + assert d2._validate_port() == base_port + assert d2.new_desktop + + # Start 2026.1, remote_rpc_session + mock_settings.remote_rpc_session = MagicMock() + d3 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value={}): + assert d3._validate_port() == base_port + mock_settings.remote_rpc_session = None + + # Two sessions in different versions + + # Start 2026.1, trying base_port with new_desktop True, but it is occupied + d4 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=True) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value=ACTIVE_SESSIONS): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + assert d4._validate_port() == random_port + d4._Desktop__logger.warning.assert_called_with( + f"Port {base_port} is already in use. Finding a new free port." + ) + + # Start student version, but port is occupied by another version + d5 = _make_desktop(port=base_port, version="2025.2", student_version=True, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value=ACTIVE_SESSIONS): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + assert d5._validate_port() == random_port + assert d5.new_desktop + + # Start 2026.1 in a port not occupied by another version, but new_desktop is False, so it will flip to True + d6 = _make_desktop(port=1, version="2025.2", student_version=True, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value=ACTIVE_SESSIONS): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + assert d6._validate_port() == 1 + assert d6.new_desktop + + +def test_connect_session(mock_settings): + """New connect AEDT session.""" + base_port = 50051 + random_port = 12345 + + # Sessions in different versions + + # Connect 2026.1, trying base_port with new_desktop False + d1 = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False) + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value=ACTIVE_SESSIONS): + with patch("ansys.aedt.core.desktop._find_free_port", return_value=random_port): + assert d1._validate_port() == base_port + d1._Desktop__logger.info.assert_called_with(f"Port {base_port} session has been found.") + + # Connect 2026.1 not passing port, this is using grpc_active_sessions, with multiple sessions + d2 = _make_desktop(version="2025.2", student_version=True, non_graphical=True, new_desktop=False) + with patch("ansys.aedt.core.desktop.grpc_active_sessions", return_value=[base_port, 50052]): + assert d2._validate_port() == base_port + + # Connect 2026.1 not passing port, this is using grpc_active_sessions, with multiple sessions + d3 = _make_desktop(version="2025.2", student_version=True, non_graphical=True, new_desktop=False) + with patch("ansys.aedt.core.desktop.grpc_active_sessions", return_value=[base_port]): + assert d3._validate_port() == base_port + + # Connect 2026.1, not found ports with get_target_processes, and then using _check_grpc_connection + connections = { + 11111: [ + { + "cmdline": "v261/ansysedt.exe -grpcsrv 50700 -ng", + "ip": "127.0.0.1", + "port": 49236, + "status": "ESTABLISHED", + }, + { + "cmdline": "v261/ansysedt.exe -grpcsrv 50700 -ng", + "ip": "127.0.0.1", + "port": random_port, + "status": "LISTEN", + }, + {"cmdline": "v261/ansysedt.exe -grpcsrv 50700 -ng", "ip": "0.0.0.0", "port": 2002, "status": "LISTEN"}, + { + "cmdline": "v261/ansysedt.exe -grpcsrv 50700 -ng", + "ip": "127.0.0.1", + "port": 49229, + "status": "ESTABLISHED", + }, + {"cmdline": "v261/ansysedt.exe -grpcsrv 50700 -ng", "ip": "0.0.0.0", "port": 56621, "status": "LISTEN"}, + ] + } + + target_process = [(11111, ["v261/ansysedt.exe", "-grpcsrv", f"127.0.0.1:{random_port}", "-ng"])] + d4 = _make_desktop( + port=random_port, version="2026.1", student_version=False, non_graphical=False, new_desktop=False + ) + with patch("ansys.aedt.core.generic.general_methods._check_psutil_connections", return_value=connections): + with patch("ansys.aedt.core.generic.general_methods._get_target_processes", return_value=target_process): + assert d4._validate_port() == random_port + d4._Desktop__logger.warning.assert_called_with( + f"Port {random_port} is already in use in non_graphical mode. Using it." + ) + + +def test_version_mode_flip_logs_and_changes(mock_settings): + """Port is in use by the opposite mode (graphical vs nongraphical).""" + base_port = 50051 + + # Create a desktop that is non-graphical but the session is graphical + d = _make_desktop(port=base_port, version="2026.1", student_version=False, non_graphical=True, new_desktop=False) + # all_active_sessions contains the opposite mode (graphical) + sessions = {"261_graphical": {"p": base_port}} + with patch("ansys.aedt.core.desktop.all_active_sessions", return_value=sessions): + res = d._validate_port() + assert res == base_port + # Should have flipped non_graphical to False + assert not d.non_graphical + # Logger warning about mode usage + d._Desktop__logger.warning.assert_called_with( + f"Port {base_port} is already in use in graphical mode. Using it." + )