CHORE: Rpc test updates

.github/workflows/ci_cd.yml

@@ -284,79 +284,78 @@ jobs:
           path: mechanical_tests_log-${{ matrix.mechanical-version }}.txt
           retention-days: 7
 
-  # embedding-rpc-tests:
-  #   name: Embedding rpc testing and coverage
-  #   runs-on: public-ubuntu-latest-8-cores
-  #   timeout-minutes: 10
-  #   needs: [smoke-tests, revn-variations, container-stability-check]
-  #   container:
-  #     image: ${{ needs.revn-variations.outputs.test_container }}
-  #     options: --entrypoint /bin/bash
-  #   strategy:
-  #     fail-fast: false
-  #     matrix:
-  #       python-version: ['3.10', '3.11', '3.12', '3.13']
-
-  #   steps:
-  #     - uses: actions/checkout@v4
-  #     - name: Set up python and pip
-  #       run: |
-  #         apt update
-  #         apt install --reinstall ca-certificates
-  #         apt install lsb-release xvfb software-properties-common -y
-  #         add-apt-repository ppa:deadsnakes/ppa -y
-  #         apt install -y python${{ matrix.python-version }} python${{ matrix.python-version }}-venv
-  #         python${{ matrix.python-version }} -m venv /env
-
-  #     - name: Install packages for testing
-  #       run: |
-  #         . /env/bin/activate
-  #         pip install --upgrade pip
-  #         pip install uv
-  #         uv pip install -e .[tests,rpc]
-
-  #     - name: Unit Testing and coverage
-  #       env:
-  #         ANSYS_WORKBENCH_LOGGING_CONSOLE: 0
-  #         ANSYS_WORKBENCH_LOGGING: 0
-  #         ANSYS_WORKBENCH_LOGGING_FILTER_LEVEL: 2
-  #         PYTHONUNBUFFERED: 1
-  #       run: |
-  #         . /env/bin/activate
-  #         # if [ "${{ needs.container-stability-check.outputs.container_stable_exit }}" = "true" ]; then
-  #         #   xvfb-run mechanical-env pytest -m remote_session_connect --remote-server-type=rpyc -s --junitxml test_results${{ matrix.python-version }}.xml
-  #         # else
-  #         #   xvfb-run mechanical-env pytest -m remote_session_connect --remote-server-type=rpyc -s --junitxml test_results${{ matrix.python-version }}.xml || true
-  #         # fi
-  #         xvfb-run mechanical-env pytest -m remote_session_connect --remote-server-type=rpyc || true
-
-      # - name: Upload coverage results
-      #   uses: actions/upload-artifact@v4
-      #   if: env.MAIN_PYTHON_VERSION == matrix.python-version
-      #   with:
-      #     include-hidden-files: true
-      #     name: coverage-tests-embedding-rpc
-      #     path: .cov
-      #     retention-days: 7
-
-      # - name: Upload coverage results (as .coverage)
-      #   uses: actions/upload-artifact@v4
-      #   if: env.MAIN_PYTHON_VERSION == matrix.python-version
-      #   with:
-      #     include-hidden-files: true
-      #     name: coverage-file-tests-embedding-rpc
-      #     path: .coverage
-      #     retention-days: 7
-
-      # - name: Publish Test Report
-      #   uses: mikepenz/action-junit-report@v5
-      #   if: always()
-      #   with:
-      #     report_paths: '**/test_results*.xml'
-      #     check_name: Test Report ${{ matrix.python-version }}
-      #     detailed_summary: true
-      #     include_passed: true
-      #     fail_on_failure: false
+  embedding-rpc-tests:
+    name: Embedding rpc testing and coverage
+    runs-on: public-ubuntu-latest-8-cores
+    timeout-minutes: 60
+    needs: [smoke-tests, revn-variations, container-stability-check]
+    container:
+      image: ${{ needs.revn-variations.outputs.test_container }}
+      options: --entrypoint /bin/bash
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.10', '3.11', '3.12', '3.13']
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up python and pip
+        run: |
+          apt update
+          apt install --reinstall ca-certificates
+          apt install lsb-release xvfb software-properties-common -y
+          add-apt-repository ppa:deadsnakes/ppa -y
+          apt install -y python${{ matrix.python-version }} python${{ matrix.python-version }}-venv
+          python${{ matrix.python-version }} -m venv /env
+
+      - name: Install packages for testing
+        run: |
+          . /env/bin/activate
+          pip install --upgrade pip
+          pip install uv
+          uv pip install -e .[tests,rpc]
+
+      - name: Unit Testing and coverage
+        env:
+          ANSYS_WORKBENCH_LOGGING_CONSOLE: 0
+          ANSYS_WORKBENCH_LOGGING: 0
+          ANSYS_WORKBENCH_LOGGING_FILTER_LEVEL: 2
+          PYTHONUNBUFFERED: 1
+        run: |
+          . /env/bin/activate
+          if [ "${{ needs.container-stability-check.outputs.container_stable_exit }}" = "true" ]; then
+            xvfb-run mechanical-env pytest -m remote_session_connect --remote-server-type=rpyc -s --junitxml test_results${{ matrix.python-version }}.xml
+          else
+            xvfb-run mechanical-env pytest -m remote_session_connect --remote-server-type=rpyc -s --junitxml test_results${{ matrix.python-version }}.xml || true
+          fi
+
+      - name: Upload coverage results
+        uses: actions/upload-artifact@v4
+        if: env.MAIN_PYTHON_VERSION == matrix.python-version
+        with:
+          include-hidden-files: true
+          name: coverage-tests-embedding-rpc
+          path: .cov
+          retention-days: 7
+
+      - name: Upload coverage results (as .coverage)
+        uses: actions/upload-artifact@v4
+        if: env.MAIN_PYTHON_VERSION == matrix.python-version
+        with:
+          include-hidden-files: true
+          name: coverage-file-tests-embedding-rpc
+          path: .coverage
+          retention-days: 7
+
+      - name: Publish Test Report
+        uses: mikepenz/action-junit-report@v5
+        if: always()
+        with:
+          report_paths: '**/test_results*.xml'
+          check_name: Test Report ${{ matrix.python-version }}
+          detailed_summary: true
+          include_passed: true
+          fail_on_failure: false
 
 
   embedding-tests:
@@ -702,8 +701,7 @@ jobs:
 
   coverage:
     name: Merging coverage
-    needs: [remote-connect, embedding-tests, embedding-scripts-tests, launch-tests]
-    # needs: [remote-connect, embedding-tests, embedding-scripts-tests, embedding-rpc-tests, launch-tests]
+    needs: [remote-connect, embedding-tests, embedding-scripts-tests, embedding-rpc-tests, launch-tests]
     runs-on: ubuntu-latest
     steps:
       - name: Checkout repository
@@ -729,10 +727,10 @@ jobs:
           name: coverage-file-tests-embedding
           path: cov-dir/embedding
 
-      # - uses: actions/download-artifact@v4
-      #   with:
-      #     name: coverage-file-tests-embedding-rpc
-      #     path: cov-dir/embedding-rpc
+      - uses: actions/download-artifact@v4
+        with:
+          name: coverage-file-tests-embedding-rpc
+          path: cov-dir/embedding-rpc
 
       - uses: actions/download-artifact@v4
         with:

doc/changelog.d/1120.maintenance.md

@@ -0,0 +1 @@
+Rpc test updates

src/ansys/mechanical/core/embedding/rpc/client.py

@@ -61,6 +61,7 @@ def __init__(self, host: str, port: int, timeout: float = 120.0, cleanup_on_exit
         self.root = None
         self._connect()
         self._cleanup_on_exit = cleanup_on_exit
+        self._error_type = Exception
 
     def __getattr__(self, attr):
         """Get attribute from the root object."""
@@ -234,6 +235,11 @@ def download_project(self, extensions=None, target_dir=None, progress_bar=False)
             list_of_files.extend(temp_files)
         return list_of_files
 
+    @property
+    def backend(self) -> str:
+        """Get the backend type."""
+        return "python"
+
     @property
     def is_alive(self):
         """Check if the Mechanical instance is alive."""

src/ansys/mechanical/core/mechanical.py

@@ -460,10 +460,11 @@ def __init__(
         else:
             self.log_info("Mechanical connection is treated as remote.")
 
+        self._error_type = grpc.RpcError
+
         # connect and validate to the channel
         self._multi_connect(timeout=timeout)
         self.log_info("Mechanical is ready to accept grpc calls.")
-        self._rpc_type = "grpc"
 
     def __del__(self):  # pragma: no cover
         """Clean up on exit."""
@@ -482,6 +483,11 @@ def log(self):
         """Log associated with the current Mechanical instance."""
         return self._log
 
+    @property
+    def backend(self) -> str:
+        """Return the backend type."""
+        return "mechanical"
+
     @property
     def version(self) -> str:
         """Get the Mechanical version based on the instance.

tests/conftest.py

@@ -30,7 +30,6 @@
 import time
 
 import ansys.tools.path as atp
-import grpc
 import pytest
 
 import ansys.mechanical.core as pymechanical
@@ -292,11 +291,9 @@ def connect_to_mechanical_instance(port=None, clear_on_connect=False):
 
 def launch_rpc_embedded_server(port: int, version: int, server_script: str):
     """Start the server as a subprocess using `port`."""
-    global embedded_server
     env_copy = os.environ.copy()
-    embedded_server = subprocess.Popen(
-        [sys.executable, server_script, str(port), str(version)], env=env_copy
-    )
+    p = subprocess.Popen([sys.executable, server_script, str(port), str(version)], env=env_copy)
+    return p
 
 
 def connect_rpc_embedded_server(port: int):
@@ -306,53 +303,94 @@ def connect_rpc_embedded_server(port: int):
     return client
 
 
+def _launch_mechanical_rpyc_server(rootdir: str, version: int):
+    """Start rpyc server process, return the process object."""
+    from ansys.mechanical.core.embedding.rpc import MechanicalEmbeddedServer
+
+    server_py = os.path.join(rootdir, "tests", "scripts", "rpc_server_embedded.py")
+    port = MechanicalEmbeddedServer.get_free_port()
+    embedded_server = launch_rpc_embedded_server(
+        port=port, version=version, server_script=server_py
+    )
+    return embedded_server, port
+
+
+def _get_mechanical_server():
+    if not pymechanical.mechanical.get_start_instance():
+        mechanical = connect_to_mechanical_instance()
+    else:
+        mechanical = launch_mechanical_instance()
+    return mechanical
+
+
+def _stop_python_server(mechanical, server_process):
+    mechanical.exit()
+    start_time = time.time()
+    while server_process.poll() is None:
+        if time.time() - start_time > 10:
+            try:
+                server_process.terminate()
+                server_process.wait()
+            except subprocess.TimeoutExpired:
+                server_process.kill()
+            break
+        time.sleep(0.5)
+
+
+def _stop_mechanical_server(mechanical):
+    assert "Ansys Mechanical" in str(mechanical)
+    if pymechanical.mechanical.get_start_instance():
+        print(f"get_start_instance() returned True. exiting mechanical.")
+        mechanical.exit(force=True)
+        assert mechanical.exited
+        assert "Mechanical exited" in str(mechanical)
+        with pytest.raises(MechanicalExitedError):
+            mechanical.run_python_script("3+4")
+
+
 @pytest.fixture(scope="session")
-def mechanical(pytestconfig, rootdir):
-    print("current working directory: ", os.getcwd())
-    is_embedded_server = pytestconfig.getoption("remote_server_type") == "rpyc"
-    if is_embedded_server:
-        from ansys.mechanical.core.embedding.rpc import MechanicalEmbeddedServer
-
-        _version = int(pytestconfig.getoption("ansys_version"))
-        server_py = os.path.join(rootdir, "tests", "scripts", "rpc_server_embedded.py")
-        _port = MechanicalEmbeddedServer.get_free_port()
-        launch_rpc_embedded_server(port=_port, version=_version, server_script=server_py)
-        mechanical = connect_rpc_embedded_server(port=_port)
-        setattr(mechanical, "_rpc_error_type", Exception)
-        setattr(mechanical, "_rpc_type", "rpyc")
+def mechanical_session(pytestconfig, rootdir):
+    print("Mechanical session fixture")
+    is_python_server = pytestconfig.getoption("remote_server_type") == "rpyc"
+    version = int(pytestconfig.getoption("ansys_version"))
+    if is_python_server:
+        print("Mechanical session fixture - starting subprocess")
+        server_process, port = _launch_mechanical_rpyc_server(rootdir, version)
+        print(f"connecting to {port}")
+        mechanical = connect_rpc_embedded_server(port=port)
+    else:
+        server_process = None
+        mechanical = _get_mechanical_server()
+
+    print("Yielding server")
+    yield (mechanical, server_process)
+    print("Stopping server")
+    if is_python_server:
+        _stop_python_server(mechanical, server_process)
     else:
-        if not pymechanical.mechanical.get_start_instance():
-            mechanical = connect_to_mechanical_instance()
-        else:
-            mechanical = launch_mechanical_instance()
-        setattr(mechanical, "_rpc_error_type", grpc.RpcError)
+        _stop_mechanical_server(mechanical)
+    print("mechanical rpc session fixture exited cleanly")
 
-    print(mechanical)
+
+@pytest.fixture(autouse=True)
+def mke_app_reset(request, printer):
+    global EMBEDDED_APP
+    if EMBEDDED_APP is None:
+        # embedded app was not started - no need to do anything
+        return
+    printer(f"starting test {request.function.__name__} - file new")
+    EMBEDDED_APP.new()
+
+
+@pytest.fixture()
+def mechanical(request, printer, mechanical_session):
+    mechanical, server_process = mechanical_session
+    if server_process is not None:
+        ret = server_process.poll()
+        if ret is not None:
+            raise Exception(f"The server process has terminated with error code {ret}")
+    assert mechanical.is_alive, "The server process has not terminated but connection has been lost"
     yield mechanical
-    if is_embedded_server:
-        print("\n Stopping embedded server")
-        global embedded_server
-        mechanical.exit()
-        start_time = time.time()
-        while embedded_server.poll() is None:
-            if time.time() - start_time > 10:
-                try:
-                    embedded_server.terminate()
-                    embedded_server.wait()
-                except subprocess.TimeoutExpired:
-                    embedded_server.kill()
-                break
-            time.sleep(0.5)
-    else:
-        assert "Ansys Mechanical" in str(mechanical)
-
-        if pymechanical.mechanical.get_start_instance():
-            print(f"get_start_instance() returned True. exiting mechanical.")
-            mechanical.exit(force=True)
-            assert mechanical.exited
-            assert "Mechanical exited" in str(mechanical)
-            with pytest.raises(MechanicalExitedError):
-                mechanical.run_python_script("3+4")
 
 
 # used only once