Skip to content

Commit

Permalink
test: add connection alive
Browse files Browse the repository at this point in the history
  • Loading branch information
dipinknair committed Jan 22, 2025
1 parent cf8b562 commit 14254a7
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 0 deletions.
10 changes: 10 additions & 0 deletions src/ansys/mechanical/core/embedding/rpc/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,16 @@ def download_project(self, extensions=None, target_dir=None, progress_bar=False)
list_of_files.extend(temp_files)
return list_of_files

@property
def is_alive(self):
"""Check if the Mechanical instance is alive."""
try:
conn = rpyc.connect(self.host, self.port)
conn.ping()
return True
except:
return False

Check warning on line 231 in src/ansys/mechanical/core/embedding/rpc/client.py

View check run for this annotation

Codecov / codecov/patch

src/ansys/mechanical/core/embedding/rpc/client.py#L230-L231

Added lines #L230 - L231 were not covered by tests

def exit(self):
"""Shuts down the Mechanical instance."""
# TODO: implement
Expand Down
8 changes: 8 additions & 0 deletions tests/test_mechanical.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,8 @@ def test_run_python_script_from_file_error(mechanical):
@pytest.mark.remote_session_connect
@pytest.mark.parametrize("file_name", [r"hsec.x_t"])
def test_upload(mechanical, file_name, assets):
if not mechanical.is_alive:
pytest.fail("Mechanical instance is not alive")
mechanical.run_python_script("ExtAPI.DataModel.Project.New()")
directory = mechanical.run_python_script("ExtAPI.DataModel.Project.ProjectDirectory")
print(directory)
Expand Down Expand Up @@ -120,6 +122,8 @@ def test_upload(mechanical, file_name, assets):
# ideally this will be 64*1024, 1024*1024, etc.
@pytest.mark.parametrize("chunk_size", [10, 50, 100])
def test_upload_with_different_chunk_size(mechanical, chunk_size, assets):
if not mechanical.is_alive:
pytest.fail("Mechanical instance is not alive")

Check failure on line 126 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.10

test_mechanical.test_upload_with_different_chunk_size[50]

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f33ccc50b50>
chunk_size = 50
assets = PosixPath('/__w/pymechanical/pymechanical/tests/assets')

    @pytest.mark.remote_session_connect
    # we are using only a small test file
    # change the chunk_size for that
    # ideally this will be 64*1024, 1024*1024, etc.
    @pytest.mark.parametrize("chunk_size", [10, 50, 100])
    def test_upload_with_different_chunk_size(mechanical, chunk_size, assets):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:126: Failed

Check failure on line 126 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.10

test_mechanical.test_upload_with_different_chunk_size[100]

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f33ccc50b50>
chunk_size = 100
assets = PosixPath('/__w/pymechanical/pymechanical/tests/assets')

    @pytest.mark.remote_session_connect
    # we are using only a small test file
    # change the chunk_size for that
    # ideally this will be 64*1024, 1024*1024, etc.
    @pytest.mark.parametrize("chunk_size", [10, 50, 100])
    def test_upload_with_different_chunk_size(mechanical, chunk_size, assets):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:126: Failed

Check failure on line 126 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.11

test_mechanical.test_upload_with_different_chunk_size[100]

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f6ff561f8d0>
chunk_size = 100
assets = PosixPath('/__w/pymechanical/pymechanical/tests/assets')

    @pytest.mark.remote_session_connect
    # we are using only a small test file
    # change the chunk_size for that
    # ideally this will be 64*1024, 1024*1024, etc.
    @pytest.mark.parametrize("chunk_size", [10, 50, 100])
    def test_upload_with_different_chunk_size(mechanical, chunk_size, assets):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:126: Failed
file_path = os.path.join(assets, "hsec.x_t")
mechanical.run_python_script("ExtAPI.DataModel.Project.New()")

Check failure on line 128 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.10

test_mechanical.test_upload_with_different_chunk_size[10]

TimeoutError: result expired
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f33ccc50b50>
chunk_size = 10
assets = PosixPath('/__w/pymechanical/pymechanical/tests/assets')

    @pytest.mark.remote_session_connect
    # we are using only a small test file
    # change the chunk_size for that
    # ideally this will be 64*1024, 1024*1024, etc.
    @pytest.mark.parametrize("chunk_size", [10, 50, 100])
    def test_upload_with_different_chunk_size(mechanical, chunk_size, assets):
        if not mechanical.is_alive:
            pytest.fail("Mechanical instance is not alive")
        file_path = os.path.join(assets, "hsec.x_t")
>       mechanical.run_python_script("ExtAPI.DataModel.Project.New()")

tests/test_mechanical.py:128: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/env/lib/python3.10/site-packages/rpyc/core/netref.py:239: in __call__
    return syncreq(_self, consts.HANDLE_CALL, args, kwargs)
/env/lib/python3.10/site-packages/rpyc/core/netref.py:63: in syncreq
    return conn.sync_request(handler, proxy, *args)
/env/lib/python3.10/site-packages/rpyc/core/protocol.py:744: in sync_request
    return _async_res.value
/env/lib/python3.10/site-packages/rpyc/core/async_.py:109: in value
    self.wait()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <AsyncResult object (expired) at 0x7f33ccb9ab60>

    def wait(self):
        """Waits for the result to arrive. If the AsyncResult object has an
        expiry set, and the result did not arrive within that timeout,
        an :class:`AsyncResultTimeout` exception is raised"""
        while self._waiting():
            # Serve the connection since we are not ready. Suppose
            # the reply for our seq is served. The callback is this class
            # so __call__ sets our obj and _is_ready to true.
            self._conn.serve(self._ttl, waiting=self._waiting)
    
        # Check if we timed out before result was ready
        if not self._is_ready:
>           raise AsyncResultTimeout("result expired")
E           TimeoutError: result expired

/env/lib/python3.10/site-packages/rpyc/core/async_.py:55: TimeoutError

Check failure on line 128 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.12

test_mechanical.test_upload_with_different_chunk_size[100]

TimeoutError: result expired
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f8a4f0f5550>
chunk_size = 100
assets = PosixPath('/__w/pymechanical/pymechanical/tests/assets')

    @pytest.mark.remote_session_connect
    # we are using only a small test file
    # change the chunk_size for that
    # ideally this will be 64*1024, 1024*1024, etc.
    @pytest.mark.parametrize("chunk_size", [10, 50, 100])
    def test_upload_with_different_chunk_size(mechanical, chunk_size, assets):
        if not mechanical.is_alive:
            pytest.fail("Mechanical instance is not alive")
        file_path = os.path.join(assets, "hsec.x_t")
>       mechanical.run_python_script("ExtAPI.DataModel.Project.New()")

tests/test_mechanical.py:128: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/env/lib/python3.12/site-packages/rpyc/core/netref.py:239: in __call__
    return syncreq(_self, consts.HANDLE_CALL, args, kwargs)
/env/lib/python3.12/site-packages/rpyc/core/netref.py:63: in syncreq
    return conn.sync_request(handler, proxy, *args)
/env/lib/python3.12/site-packages/rpyc/core/protocol.py:744: in sync_request
    return _async_res.value
/env/lib/python3.12/site-packages/rpyc/core/async_.py:109: in value
    self.wait()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <AsyncResult object (expired) at 0x7f8a4eb3bab0>

    def wait(self):
        """Waits for the result to arrive. If the AsyncResult object has an
        expiry set, and the result did not arrive within that timeout,
        an :class:`AsyncResultTimeout` exception is raised"""
        while self._waiting():
            # Serve the connection since we are not ready. Suppose
            # the reply for our seq is served. The callback is this class
            # so __call__ sets our obj and _is_ready to true.
            self._conn.serve(self._ttl, waiting=self._waiting)
    
        # Check if we timed out before result was ready
        if not self._is_ready:
>           raise AsyncResultTimeout("result expired")
E           TimeoutError: result expired

/env/lib/python3.12/site-packages/rpyc/core/async_.py:55: TimeoutError

Check failure on line 128 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.11

test_mechanical.test_upload_with_different_chunk_size[50]

TimeoutError: result expired
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f6ff561f8d0>
chunk_size = 50
assets = PosixPath('/__w/pymechanical/pymechanical/tests/assets')

    @pytest.mark.remote_session_connect
    # we are using only a small test file
    # change the chunk_size for that
    # ideally this will be 64*1024, 1024*1024, etc.
    @pytest.mark.parametrize("chunk_size", [10, 50, 100])
    def test_upload_with_different_chunk_size(mechanical, chunk_size, assets):
        if not mechanical.is_alive:
            pytest.fail("Mechanical instance is not alive")
        file_path = os.path.join(assets, "hsec.x_t")
>       mechanical.run_python_script("ExtAPI.DataModel.Project.New()")

tests/test_mechanical.py:128: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/env/lib/python3.11/site-packages/rpyc/core/netref.py:239: in __call__
    return syncreq(_self, consts.HANDLE_CALL, args, kwargs)
/env/lib/python3.11/site-packages/rpyc/core/netref.py:63: in syncreq
    return conn.sync_request(handler, proxy, *args)
/env/lib/python3.11/site-packages/rpyc/core/protocol.py:744: in sync_request
    return _async_res.value
/env/lib/python3.11/site-packages/rpyc/core/async_.py:109: in value
    self.wait()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <AsyncResult object (expired) at 0x7f6ff52747c0>

    def wait(self):
        """Waits for the result to arrive. If the AsyncResult object has an
        expiry set, and the result did not arrive within that timeout,
        an :class:`AsyncResultTimeout` exception is raised"""
        while self._waiting():
            # Serve the connection since we are not ready. Suppose
            # the reply for our seq is served. The callback is this class
            # so __call__ sets our obj and _is_ready to true.
            self._conn.serve(self._ttl, waiting=self._waiting)
    
        # Check if we timed out before result was ready
        if not self._is_ready:
>           raise AsyncResultTimeout("result expired")
E           TimeoutError: result expired

/env/lib/python3.11/site-packages/rpyc/core/async_.py:55: TimeoutError
directory = mechanical.run_python_script("ExtAPI.DataModel.Project.ProjectDirectory")
Expand Down Expand Up @@ -274,6 +278,8 @@ def verify_project_download(mechanical, tmpdir):
# @pytest.mark.wip
# @pytest.mark.skip(reason="avoid long running")
def test_upload_attach_mesh_solve_use_api_non_distributed_solve(mechanical, tmpdir):
if not mechanical.is_alive:
pytest.fail("Mechanical instance is not alive")

Check failure on line 282 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.10

test_mechanical.test_upload_attach_mesh_solve_use_api_non_distributed_solve

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f33ccc50b50>
tmpdir = local('/tmp/pytest-of-root/pytest-0/test_upload_attach_mesh_solve_0')

    @pytest.mark.remote_session_connect
    # @pytest.mark.wip
    # @pytest.mark.skip(reason="avoid long running")
    def test_upload_attach_mesh_solve_use_api_non_distributed_solve(mechanical, tmpdir):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:282: Failed

Check failure on line 282 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.12

test_mechanical.test_upload_attach_mesh_solve_use_api_non_distributed_solve

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f8a4f0f5550>
tmpdir = local('/tmp/pytest-of-root/pytest-0/test_upload_attach_mesh_solve_0')

    @pytest.mark.remote_session_connect
    # @pytest.mark.wip
    # @pytest.mark.skip(reason="avoid long running")
    def test_upload_attach_mesh_solve_use_api_non_distributed_solve(mechanical, tmpdir):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:282: Failed

Check failure on line 282 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.11

test_mechanical.test_upload_attach_mesh_solve_use_api_non_distributed_solve

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f6ff561f8d0>
tmpdir = local('/tmp/pytest-of-root/pytest-0/test_upload_attach_mesh_solve_0')

    @pytest.mark.remote_session_connect
    # @pytest.mark.wip
    # @pytest.mark.skip(reason="avoid long running")
    def test_upload_attach_mesh_solve_use_api_non_distributed_solve(mechanical, tmpdir):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:282: Failed
# default is distributed solve
# let's disable the distributed solve and then solve
# enable the distributed solve back
Expand Down Expand Up @@ -307,6 +313,8 @@ def test_upload_attach_mesh_solve_use_api_non_distributed_solve(mechanical, tmpd

@pytest.mark.remote_session_connect
def test_upload_attach_mesh_solve_use_api_distributed_solve(mechanical, tmpdir):
if not mechanical.is_alive:
pytest.fail("Mechanical instance is not alive")

Check failure on line 317 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.10

test_mechanical.test_upload_attach_mesh_solve_use_api_distributed_solve

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f33ccc50b50>
tmpdir = local('/tmp/pytest-of-root/pytest-0/test_upload_attach_mesh_solve_1')

    @pytest.mark.remote_session_connect
    def test_upload_attach_mesh_solve_use_api_distributed_solve(mechanical, tmpdir):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:317: Failed

Check failure on line 317 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.12

test_mechanical.test_upload_attach_mesh_solve_use_api_distributed_solve

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f8a4f0f5550>
tmpdir = local('/tmp/pytest-of-root/pytest-0/test_upload_attach_mesh_solve_1')

    @pytest.mark.remote_session_connect
    def test_upload_attach_mesh_solve_use_api_distributed_solve(mechanical, tmpdir):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:317: Failed

Check failure on line 317 in tests/test_mechanical.py

View workflow job for this annotation

GitHub Actions / Test Report 3.11

test_mechanical.test_upload_attach_mesh_solve_use_api_distributed_solve

Failed: Mechanical instance is not alive
Raw output
mechanical = <ansys.mechanical.core.embedding.rpc.client.Client object at 0x7f6ff561f8d0>
tmpdir = local('/tmp/pytest-of-root/pytest-0/test_upload_attach_mesh_solve_1')

    @pytest.mark.remote_session_connect
    def test_upload_attach_mesh_solve_use_api_distributed_solve(mechanical, tmpdir):
        if not mechanical.is_alive:
>           pytest.fail("Mechanical instance is not alive")
E           Failed: Mechanical instance is not alive

tests/test_mechanical.py:317: Failed
# default is distributed solve

result = solve_and_return_results(mechanical)
Expand Down

0 comments on commit 14254a7

Please sign in to comment.