Transport: Three bug fixed in test_execmanager and AsyncSshTransport (#6855)

This commit fixes, three bugs! 
Everything is squashed to one commit as they are co-dependent and separating them will not do much benefit. 

First bug fix:
- Adopt `AsyncSshTransport` with the *COUNTER-INTUITIVE* and expected behavior of engine: 
`transport.copy()` should raise `FileNotFoundError` if remote source, or remote destination are not found. 
This later, will result in silent failure of a process, handled here: `execmanager.py`::_copy_remote_files

Second bug fix:
- `AsyncTransport::run_command_blocking` had to use the current event loop of the manager instead of creating a new one. This likely increases efficiency quite a lot, by removing unnecessarily dead locks. 
- `test_execmanager.py::node_and_calc_info` was not being run for `core.ssh_async` this was the reason that let go of these bugs un-notice. Fixed via this commit.

Third bug fix:
- Converts all sync tests in `test_execmanager.py` that where `runner.loop.run_until_complete()` into async tests. This prevents attaching a `Future` to a wrong event loop via `asyncssh`
- Improves `event_loop` fixture:
Formally always required a profile to be loaded, therefore the event loop was not cleaned before passing to a new test. Now passes `aiida_profile_clean` to `event_loop` fixture, so that a profile is loaded for all async tests, and also the event loop is cleaned for each test. This is done because `aiida_profile_clean` also triggers `manager.reset_profile()`.

---------

Co-authored-by: Alexander Goscinski <[email protected]>

Author: khsrali

src/aiida/transports/plugins/ssh_async.py

@@ -618,6 +618,8 @@ async def copy_async(
         :type preserve: bool
 
         :raises: OSError, src does not exist or if the copy execution failed.
+        :raises: FileNotFoundError, if either remotesource does not exists
+            or remotedestination's parent path does not exists
         """
 
         remotesource = str(remotesource)
@@ -648,7 +650,8 @@ async def copy_async(
                     )
                 else:
                     if not await self.path_exists_async(remotesource):
-                        raise OSError(f'The remote path {remotesource} does not exist')
+                        raise FileNotFoundError(f'The remote path {remotesource} does not exist')
+
                     await self._sftp.copy(
                         remotesource,
                         remotedestination,
@@ -657,6 +660,13 @@ async def copy_async(
                         follow_symlinks=dereference,
                         remote_only=True,
                     )
+            except asyncssh.sftp.SFTPNoSuchFile as exc:
+                # note: one could just create directories, but aiida engine expects this behavior
+                # see `execmanager.py`::_copy_remote_files for more details
+                raise FileNotFoundError(
+                    f'The remote path {remotedestination} is not reachable,'
+                    f'perhaps the parent folder does not exists: {exc}'
+                )
             except asyncssh.sftp.SFTPFailure as exc:
                 raise OSError(f'Error while copying {remotesource} to {remotedestination}: {exc}')
         else:

src/aiida/transports/transport.py

@@ -9,7 +9,6 @@
 """Transport interface."""
 
 import abc
-import asyncio
 import fnmatch
 import os
 import re
@@ -1793,7 +1792,11 @@ class AsyncTransport(Transport):
     """
 
     def run_command_blocking(self, func, *args, **kwargs):
-        loop = asyncio.get_event_loop()
+        """The event loop must be the one of manager."""
+
+        from aiida.manage import get_manager
+
+        loop = get_manager().get_runner()
         return loop.run_until_complete(func(*args, **kwargs))
 
     def open(self):

tests/conftest.py

@@ -525,8 +525,10 @@ def runner(manager):
 
 
 @pytest.fixture
-def event_loop(manager):
-    """Get the event loop instance of the currently loaded profile.
+def event_loop(manager, aiida_profile_clean):
+    """Get the event loop instance of a cleaned profile.
+    This works, because ``aiida_profile_clean`` fixture, apart from loading a profile and cleaning it,
+    and also triggers ``manager.reset_profile()`` which clears the event loop.
 
     This is automatically called as a fixture for any test marked with ``@pytest.mark.asyncio``.
     """

tests/engine/daemon/test_execmanager.py

@@ -16,8 +16,8 @@
 from aiida.common.datastructures import CalcInfo, CodeInfo, FileCopyOperation, StashMode
 from aiida.common.folders import SandboxFolder
 from aiida.engine.daemon import execmanager
-from aiida.manage import get_manager
 from aiida.orm import CalcJobNode, FolderData, PortableCode, RemoteData, SinglefileData
+from aiida.plugins import entry_point
 from aiida.transports.plugins.local import LocalTransport
 
 
@@ -41,14 +41,19 @@ def file_hierarchy_simple():
     }
 
 
-@pytest.fixture(params=('core.local', 'core.ssh'))
-def node_and_calc_info(aiida_localhost, aiida_computer_ssh, aiida_code_installed, request):
+@pytest.fixture(
+    scope='function',
+    params=[name for name in entry_point.get_entry_point_names('aiida.transports') if name.startswith('core.')],
+)
+def node_and_calc_info(aiida_localhost, aiida_computer_ssh, aiida_computer_ssh_async, aiida_code_installed, request):
     """Return a ``CalcJobNode`` and associated ``CalcInfo`` instance."""
 
     if request.param == 'core.local':
         node = CalcJobNode(computer=aiida_localhost)
     elif request.param == 'core.ssh':
         node = CalcJobNode(computer=aiida_computer_ssh())
+    elif request.param == 'core.ssh_async':
+        node = CalcJobNode(computer=aiida_computer_ssh_async())
     else:
         raise ValueError(f'unsupported transport: {request.param}')
 
@@ -111,7 +116,8 @@ def test_hierarchy_utility(file_hierarchy, tmp_path, create_file_hierarchy, seri
         (['file_a.txt', 'file_u.txt', 'path/file_u.txt', ('path/sub/file_u.txt', '.', 3)], {'file_a.txt': 'file_a'}),
     ),
 )
-def test_retrieve_files_from_list(
+@pytest.mark.asyncio
+async def test_retrieve_files_from_list(
     tmp_path_factory,
     generate_calcjob_node,
     file_hierarchy,
@@ -125,15 +131,15 @@ def test_retrieve_files_from_list(
     target = tmp_path_factory.mktemp('target')
 
     create_file_hierarchy(file_hierarchy, source)
-    runner = get_manager().get_runner()
 
     with LocalTransport() as transport:
         node = generate_calcjob_node(workdir=source)
-        runner.loop.run_until_complete(execmanager.retrieve_files_from_list(node, transport, target, retrieve_list))
+        await execmanager.retrieve_files_from_list(node, transport, target, retrieve_list)
 
     assert serialize_file_hierarchy(target, read_bytes=False) == expected_hierarchy
 
 
+@pytest.mark.asyncio
 @pytest.mark.parametrize(
     ('local_copy_list', 'expected_hierarchy'),
     (
@@ -148,7 +154,7 @@ def test_retrieve_files_from_list(
         (['sub', 'target/another-sub'], {'target': {'another-sub': {'b': 'file_b'}}}),
     ),
 )
-def test_upload_local_copy_list(
+async def test_upload_local_copy_list(
     fixture_sandbox,
     node_and_calc_info,
     file_hierarchy_simple,
@@ -168,8 +174,7 @@ def test_upload_local_copy_list(
     calc_info.local_copy_list = [[folder.uuid] + local_copy_list]
 
     with node.computer.get_transport() as transport:
-        runner = get_manager().get_runner()
-        runner.loop.run_until_complete(execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox))
+        await execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox)
 
     # Check that none of the files were written to the repository of the calculation node, since they were communicated
     # through the ``local_copy_list``.
@@ -180,7 +185,8 @@ def test_upload_local_copy_list(
     assert written_hierarchy == expected_hierarchy
 
 
-def test_upload_local_copy_list_files_folders(
+@pytest.mark.asyncio
+async def test_upload_local_copy_list_files_folders(
     fixture_sandbox, node_and_calc_info, file_hierarchy, tmp_path, create_file_hierarchy, serialize_file_hierarchy
 ):
     """Test the ``local_copy_list`` functionality in ``upload_calculation``.
@@ -206,8 +212,7 @@ def test_upload_local_copy_list_files_folders(
     ]
 
     with node.computer.get_transport() as transport:
-        runner = get_manager().get_runner()
-        runner.loop.run_until_complete(execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox))
+        await execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox)
 
     # Check that none of the files were written to the repository of the calculation node, since they were communicated
     # through the ``local_copy_list``.
@@ -222,7 +227,8 @@ def test_upload_local_copy_list_files_folders(
     assert expected_hierarchy == written_hierarchy
 
 
-def test_upload_remote_symlink_list(
+@pytest.mark.asyncio
+async def test_upload_remote_symlink_list(
     fixture_sandbox, node_and_calc_info, file_hierarchy, tmp_path, create_file_hierarchy
 ):
     """Test the ``remote_symlink_list`` functionality in ``upload_calculation``.
@@ -238,8 +244,7 @@ def test_upload_remote_symlink_list(
     ]
 
     with node.computer.get_transport() as transport:
-        runner = get_manager().get_runner()
-        runner.loop.run_until_complete(execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox))
+        await execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox)
 
     filepath_workdir = pathlib.Path(node.get_remote_workdir())
     assert (filepath_workdir / 'file_a.txt').is_symlink()
@@ -270,7 +275,8 @@ def test_upload_remote_symlink_list(
         ),
     ),
 )
-def test_upload_file_copy_operation_order(node_and_calc_info, tmp_path, order, expected):
+@pytest.mark.asyncio
+async def test_upload_file_copy_operation_order(node_and_calc_info, tmp_path, order, expected):
     """Test the ``CalcInfo.file_copy_operation_order`` controls the copy order."""
     node, calc_info = node_and_calc_info
 
@@ -303,8 +309,7 @@ def test_upload_file_copy_operation_order(node_and_calc_info, tmp_path, order, e
         calc_info.file_copy_operation_order = order
 
     with node.computer.get_transport() as transport:
-        runner = get_manager().get_runner()
-        runner.loop.run_until_complete(execmanager.upload_calculation(node, transport, calc_info, sandbox, inputs))
+        await execmanager.upload_calculation(node, transport, calc_info, sandbox, inputs)
         filepath = pathlib.Path(node.get_remote_workdir()) / 'file.txt'
         assert filepath.is_file()
         assert filepath.read_text() == expected
@@ -377,6 +382,7 @@ def test_upload_file_copy_operation_order(node_and_calc_info, tmp_path, order, e
         ),
         # Only remote copy of a single file to the "pseudo" directory
         # -> Copy fails silently since target directory does not exist: final directory structure is empty
+        # COUNTER-INTUITIVE: the silent behavior is expected. See `execmanager.py`::_copy_remote_files for more details
         (
             {},
             (),
@@ -385,6 +391,7 @@ def test_upload_file_copy_operation_order(node_and_calc_info, tmp_path, order, e
             None,
         ),
         # -> Copy fails silently since target directory does not exist: final directory structure is empty
+        # COUNTER-INTUITIVE: the silent behavior is expected. See `execmanager.py`::_copy_remote_files for more details
         (
             {},
             (),
@@ -532,7 +539,8 @@ def test_upload_file_copy_operation_order(node_and_calc_info, tmp_path, order, e
         ),
     ],
 )
-def test_upload_combinations(
+@pytest.mark.asyncio
+async def test_upload_combinations(
     fixture_sandbox,
     node_and_calc_info,
     tmp_path,
@@ -598,27 +606,25 @@ def test_upload_combinations(
         calc_info.remote_copy_list.append(
             (node.computer.uuid, (sub_tmp_path_remote / source_path).as_posix(), target_path)
         )
-    runner = get_manager().get_runner()
     if expected_exception is not None:
         with pytest.raises(expected_exception):
             with node.computer.get_transport() as transport:
-                runner.loop.run_until_complete(
-                    execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox)
-                )
+                await execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox)
 
             filepath_workdir = pathlib.Path(node.get_remote_workdir())
 
             assert serialize_file_hierarchy(filepath_workdir, read_bytes=False) == expected_hierarchy
     else:
         with node.computer.get_transport() as transport:
-            runner.loop.run_until_complete(execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox))
+            await execmanager.upload_calculation(node, transport, calc_info, fixture_sandbox)
 
         filepath_workdir = pathlib.Path(node.get_remote_workdir())
 
         assert serialize_file_hierarchy(filepath_workdir, read_bytes=False) == expected_hierarchy
 
 
-def test_upload_calculation_portable_code(fixture_sandbox, node_and_calc_info, tmp_path):
+@pytest.mark.asyncio
+async def test_upload_calculation_portable_code(fixture_sandbox, node_and_calc_info, tmp_path):
     """Test ``upload_calculation`` with a ``PortableCode`` for different transports.
 
     Regression test for https://github.com/aiidateam/aiida-core/issues/6518
@@ -639,14 +645,11 @@ def test_upload_calculation_portable_code(fixture_sandbox, node_and_calc_info, t
     calc_info.codes_info = [code_info]
 
     with node.computer.get_transport() as transport:
-        runner = get_manager().get_runner()
-        runner.loop.run_until_complete(
-            execmanager.upload_calculation(
-                node,
-                transport,
-                calc_info,
-                fixture_sandbox,
-            )
+        await execmanager.upload_calculation(
+            node,
+            transport,
+            calc_info,
+            fixture_sandbox,
         )
 
 
@@ -664,7 +667,8 @@ def test_upload_calculation_portable_code(fixture_sandbox, node_and_calc_info, t
         StashMode.COMPRESS_TARXZ.value,
     ],
 )
-def test_stashing(
+@pytest.mark.asyncio
+async def test_stashing(
     generate_calcjob_node,
     stash_mode,
     file_hierarchy,
@@ -699,8 +703,6 @@ def test_stashing(
         },
     )
 
-    runner = get_manager().get_runner()
-
     def mock_get_authinfo(*args, **kwargs):
         class MockAuthInfo:
             def get_workdir(self, *args, **kwargs):
@@ -716,7 +718,7 @@ def get_workdir(self, *args, **kwargs):
     # Here we using local transport we test basic functionality of `stash_calculation`.
 
     with LocalTransport() as transport:
-        runner.loop.run_until_complete(execmanager.stash_calculation(node, transport))
+        await execmanager.stash_calculation(node, transport)
 
     if stash_mode != StashMode.COPY.value:
         # more detailed test on integrity of the zip file is in `test_all_plugins.py`
@@ -762,7 +764,7 @@ async def mock_compress_async(*args, **kwargs):
         # no error should be raised
         # the error should only be logged to EXEC_LOGGER.warning and exit with 0
         with caplog.at_level(logging.WARNING):
-            runner.loop.run_until_complete(execmanager.stash_calculation(node, transport))
+            await execmanager.stash_calculation(node, transport)
             assert any('failed to stash' in message for message in caplog.messages)
 
     # Ensure no files were created in the destination path after the error