From f4c55f5f78cd7fde9a5b4a4e48cad2159fd666b2 Mon Sep 17 00:00:00 2001 From: Ali Khosravi Date: Tue, 18 Feb 2025 09:57:06 +0100 Subject: [PATCH] `Transport`: feat `compress` & `extract` methods (#6743) - Introduce two new abstract methods `compress` & `extract` for transport plugins. - Implement those methods for all transport plugins in aiida-core --- src/aiida/transports/plugins/ssh_async.py | 125 +++++++++++- src/aiida/transports/transport.py | 228 ++++++++++++++++++++++ tests/transports/test_all_plugins.py | 221 ++++++++++++++++++++- 3 files changed, 571 insertions(+), 3 deletions(-) diff --git a/src/aiida/transports/plugins/ssh_async.py b/src/aiida/transports/plugins/ssh_async.py index 5dd8d32107..7db731680a 100644 --- a/src/aiida/transports/plugins/ssh_async.py +++ b/src/aiida/transports/plugins/ssh_async.py @@ -14,7 +14,7 @@ import glob import os from pathlib import Path, PurePath -from typing import Optional +from typing import Optional, Union import asyncssh import click @@ -764,6 +764,129 @@ async def copytree_async( """ return await self.copy_async(remotesource, remotedestination, dereference, recursive=True, preserve=preserve) + async def compress_async( + self, + format: str, + remotesources: Union[TransportPath, list[TransportPath]], + remotedestination: TransportPath, + root_dir: TransportPath, + overwrite: bool = True, + dereference: bool = False, + ): + """Compress a remote directory. + + This method supports `remotesources` with glob patterns. + + :param format: format of compression, should support: 'tar', 'tar.gz', 'tar.bz', 'tar.xz' + :param remotesources: path (list of paths) to the remote directory(ies) (and/)or file(s) to compress + :param remotedestination: path to the remote destination file (including file name). + :param root_dir: the path that compressed files will be relative to. + :param overwrite: if True, overwrite the file at remotedestination if it already exists. + :param dereference: if True, follow symbolic links. + Compress where they point to, instead of the links themselves. + + :raises ValueError: if format is not supported + :raises OSError: if remotesource does not exist, or a matching file/folder cannot be found + :raises OSError: if remotedestination already exists and overwrite is False. Or if it is a directory. + :raises OSError: if cannot create remotedestination + :raises OSError: if root_dir is not a directory + """ + if not await self.isdir_async(root_dir): + raise OSError(f'The relative root {root_dir} does not exist, or is not a directory.') + + if await self.isdir_async(remotedestination): + raise OSError(f'The remote destination {remotedestination} is a directory, should include a filename.') + + if not overwrite and await self.path_exists_async(remotedestination): + raise OSError(f'The remote destination {remotedestination} already exists.') + + if format not in ['tar', 'tar.gz', 'tar.bz2', 'tar.xz']: + raise ValueError(f'Unsupported compression format: {format}') + + compression_flag = { + 'tar': '', + 'tar.gz': 'z', + 'tar.bz2': 'j', + 'tar.xz': 'J', + }[format] + + if not isinstance(remotesources, list): + remotesources = [remotesources] + + copy_list = [] + + for source in remotesources: + if self.has_magic(source): + try: + copy_list += await self.glob_async(source) + except asyncssh.sftp.SFTPNoSuchFile: + raise OSError( + f'Either the remote path {source} does not exist, or a matching file/folder not found.' + ) + else: + if not await self.path_exists_async(source): + raise OSError(f'The remote path {source} does not exist') + + copy_list.append(source) + + copy_items = ' '.join([str(Path(item).relative_to(root_dir)) for item in copy_list]) + # note: order of the flags is important + tar_command = ( + f"tar -c{compression_flag!s}{'h' if dereference else ''}f {remotedestination!s} -C {root_dir!s} " + + copy_items + ) + + retval, stdout, stderr = await self.exec_command_wait_async(tar_command) + + if retval == 0: + if stderr.strip(): + self.logger.warning(f'There was nonempty stderr in the tar command: {stderr}') + else: + self.logger.error( + "Problem executing tar. Exit code: {}, stdout: '{}', stderr: '{}', command: '{}'".format( + retval, stdout, stderr, tar_command + ) + ) + raise OSError(f'Error while creating the tar archive. Exit code: {retval}') + + async def extract_async( + self, remotesource: TransportPath, remotedestination: TransportPath, overwrite: bool = True + ): + """Extract a remote archive. + + Does not accept glob patterns, as it doesn't make much sense and we don't have a usecase for it. + + :param remotesource: path to the remote archive to extract + :param remotedestination: path to the remote destination directory + :param overwrite: if True, overwrite the file at remotedestination if it already exists + (we don't have a usecase for False, sofar. The parameter is kept for clarity.) + + :raises OSError: if the remotesource does not exist. + :raises OSError: if the extraction fails. + """ + if not overwrite: + raise NotImplementedError('The overwrite=False is not implemented yet') + + if not await self.path_exists_async(remotesource): + raise OSError(f'The remote path {remotesource} does not exist') + + await self.makedirs_async(remotedestination, ignore_existing=True) + + tar_command = f'tar -xf {remotesource!s} -C {remotedestination!s} ' + + retval, stdout, stderr = await self.exec_command_wait_async(tar_command) + + if retval == 0: + if stderr.strip(): + self.logger.warning(f'There was nonempty stderr in the tar command: {stderr}') + else: + self.logger.error( + "Problem executing tar. Exit code: {}, stdout: '{}', " "stderr: '{}', command: '{}'".format( + retval, stdout, stderr, tar_command + ) + ) + raise OSError(f'Error while extracting the tar archive. Exit code: {retval}') + async def exec_command_wait_async( self, command: str, diff --git a/src/aiida/transports/transport.py b/src/aiida/transports/transport.py index 2f19669a18..7b1d758526 100644 --- a/src/aiida/transports/transport.py +++ b/src/aiida/transports/transport.py @@ -965,6 +965,50 @@ def glob0(self, dirname, basename): return [basename] return [] + @abc.abstractmethod + def compress( + self, + format: str, + remotesources: Union[TransportPath, list[TransportPath]], + remotedestination: TransportPath, + root_dir: TransportPath, + overwrite: bool = True, + dereference: bool = False, + ): + """Compress a remote directory. + + This method supports `remotesources` with glob patterns. + + :param format: format of compression, should support: 'tar', 'tar.gz', 'tar.bz', 'tar.xz' + :param remotesources: path (list of paths) to the remote directory(ies) (and/)or file(s) to compress + :param remotedestination: path to the remote destination file (including file name). + :param root_dir: the path that compressed files will be relative to. + :param overwrite: if True, overwrite the file at remotedestination if it already exists. + :param dereference: if True, follow symbolic links. + Compress where they point to, instead of the links themselves. + + :raises ValueError: if format is not supported + :raises OSError: if remotesource does not exist, or a matching file/folder cannot be found + :raises OSError: if remotedestination already exists and overwrite is False. Or if it is a directory. + :raises OSError: if cannot create remotedestination + :raises OSError: if root_dir is not a directory + """ + + @abc.abstractmethod + def extract(self, remotesource: TransportPath, remotedestination: TransportPath, overwrite: bool = True): + """Extract a remote archive. + + Does not accept glob patterns, as it doesn't make much sense and we don't have a usecase for it. + + :param remotesource: path to the remote archive to extract + :param remotedestination: path to the remote destination directory + :param overwrite: if True, overwrite the file at remotedestination if it already exists + (we don't have a usecase for False, sofar. The parameter is kept for clarity.) + + :raises OSError: if the remotesource does not exist. + :raises OSError: if the extraction fails. + """ + ## aiida-core engine is ultimately moving towards async, so this is a step in that direction. @abc.abstractmethod @@ -1416,6 +1460,52 @@ async def glob_async(self, pathname: TransportPath): :return: a list of paths matching the pattern. """ + @abc.abstractmethod + async def compress_async( + self, + format: str, + remotesources: Union[TransportPath, list[TransportPath]], + remotedestination: TransportPath, + root_dir: TransportPath, + overwrite: bool = True, + dereference: bool = False, + ): + """Compress a remote directory. + + This method supports `remotesources` with glob patterns. + + :param format: format of compression, should support: 'tar', 'tar.gz', 'tar.bz', 'tar.xz' + :param remotesources: path (list of paths) to the remote directory(ies) (and/)or file(s) to compress + :param remotedestination: path to the remote destination file (including file name). + :param root_dir: the path that compressed files will be relative to. + :param overwrite: if True, overwrite the file at remotedestination if it already exists. + :param dereference: if True, follow symbolic links. + Compress where they point to, instead of the links themselves. + + :raises ValueError: if format is not supported + :raises OSError: if remotesource does not exist, or a matching file/folder cannot be found + :raises OSError: if remotedestination already exists and overwrite is False. Or if it is a directory. + :raises OSError: if cannot create remotedestination + :raises OSError: if root_dir is not a directory + """ + + @abc.abstractmethod + async def extract_async( + self, remotesource: TransportPath, remotedestination: TransportPath, overwrite: bool = True + ): + """Extract a remote archive. + + Does not accept glob patterns, as it doesn't make much sense and we don't have a usecase for it. + + :param remotesource: path to the remote archive to extract + :param remotedestination: path to the remote destination directory + :param overwrite: if True, overwrite the file at remotedestination if it already exists + (we don't have a usecase for False, sofar. The parameter is kept for clarity.) + + :raises OSError: if the remotesource does not exist. + :raises OSError: if the extraction fails. + """ + class BlockingTransport(Transport): """Abstract class for a generic blocking transport. @@ -1425,6 +1515,128 @@ class BlockingTransport(Transport): This is done by awaiting the sync methods. """ + def compress( + self, + format: str, + remotesources: Union[TransportPath, list[TransportPath]], + remotedestination: TransportPath, + root_dir: TransportPath, + overwrite: bool = True, + dereference: bool = False, + ): + # The following implementation works for all blocking transoprt plugins + """Compress a remote directory. + + This method supports `remotesources` with glob patterns. + + :param format: format of compression, should support: 'tar', 'tar.gz', 'tar.bz', 'tar.xz' + :param remotesources: path (list of paths) to the remote directory(ies) (and/)or file(s) to compress + :param remotedestination: path to the remote destination file (including file name). + :param root_dir: the path that compressed files will be relative to. + :param overwrite: if True, overwrite the file at remotedestination if it already exists. + :param dereference: if True, follow symbolic links. + Compress where they point to, instead of the links themselves. + + :raises ValueError: if format is not supported + :raises OSError: if remotesource does not exist, or a matching file/folder cannot be found + :raises OSError: if remotedestination already exists and overwrite is False. Or if it is a directory. + :raises OSError: if cannot create remotedestination + :raises OSError: if root_dir is not a directory + """ + if not self.isdir(root_dir): + raise OSError(f'The relative root {root_dir} does not exist, or is not a directory.') + + if self.isdir(remotedestination): + raise OSError(f'The remote destination {remotedestination} is a directory, should include a filename.') + + if not overwrite and self.path_exists(remotedestination): + raise OSError(f'The remote destination {remotedestination} already exists.') + + if format not in ['tar', 'tar.gz', 'tar.bz2', 'tar.xz']: + raise ValueError(f'Unsupported compression format: {type}') + + compression_flag = { + 'tar': '', + 'tar.gz': 'z', + 'tar.bz2': 'j', + 'tar.xz': 'J', + }[format] + + if not isinstance(remotesources, list): + remotesources = [remotesources] + + copy_list = [] + + for source in remotesources: + if self.has_magic(source): + copy_list = self.glob(source) + if not copy_list: + raise OSError( + f'Either the remote path {source} does not exist, or a matching file/folder not found.' + ) + else: + if not self.path_exists(source): + raise OSError(f'The remote path {source} does not exist') + + copy_list.append(source) + + copy_items = ' '.join([str(Path(item).relative_to(root_dir)) for item in copy_list]) + # note: order of the flags is important + tar_command = ( + f"tar -c{compression_flag!s}{'h' if dereference else ''}f {remotedestination!s} -C {root_dir!s} " + + copy_items + ) + + retval, stdout, stderr = self.exec_command_wait(tar_command) + + if retval == 0: + if stderr.strip(): + self.logger.warning(f'There was nonempty stderr in the tar command: {stderr}') + else: + self.logger.error( + "Problem executing tar. Exit code: {}, stdout: '{}', " "stderr: '{}', command: '{}'".format( + retval, stdout, stderr, tar_command + ) + ) + raise OSError(f'Error while creating the tar archive. Exit code: {retval}') + + def extract(self, remotesource, remotedestination, overwrite=True, *args, **kwargs): + # The following implementation works for all blocking transoprt plugins + """Extract a remote archive. + + Does not accept glob patterns, as it doesn't make much sense and we don't have a usecase for it. + + :param remotesource: path to the remote archive to extract + :param remotedestination: path to the remote destination directory + :param overwrite: if True, overwrite the file at remotedestination if it already exists + (we don't have a usecase for False, sofar. The parameter is kept for clarity.) + + :raises OSError: if the remotesource does not exist. + :raises OSError: if the extraction fails. + """ + if not overwrite: + raise NotImplementedError('The overwrite=False is not implemented yet') + + if not self.path_exists(remotesource): + raise OSError(f'The remote path {remotesource} does not exist') + + self.makedirs(remotedestination, ignore_existing=True) + + tar_command = f'tar -xf {remotesource!s} -C {remotedestination!s} ' + + retval, stdout, stderr = self.exec_command_wait(tar_command) + + if retval == 0: + if stderr.strip(): + self.logger.warning(f'There was nonempty stderr in the tar command: {stderr}') + else: + self.logger.error( + "Problem executing tar. Exit code: {}, stdout: '{}', " "stderr: '{}', command: '{}'".format( + retval, stdout, stderr, tar_command + ) + ) + raise OSError(f'Error while extracting the tar archive. Exit code: {retval}') + async def open_async(self): """Counterpart to open() that is async.""" return self.open() @@ -1561,6 +1773,16 @@ async def glob_async(self, pathname): """Counterpart to glob() that is async.""" return self.glob(pathname) + async def compress_async( + self, format, remotesources, remotedestination, root_dir, overwrite=True, dereference=False + ): + """Counterpart to compress() that is async.""" + return self.compress(format, remotesources, remotedestination, root_dir, overwrite, dereference) + + async def extract_async(self, remotesource, remotedestination, overwrite=True): + """Counterpart to extract() that is async.""" + return self.extract(remotesource, remotedestination, overwrite) + class AsyncTransport(Transport): """An abstract base class for asynchronous transports. @@ -1673,6 +1895,12 @@ def glob(self, *args, **kwargs): def normalize(self, *args, **kwargs): return self.run_command_blocking(self.normalize_async, *args, **kwargs) + def extract(self, *args, **kwargs): + return self.run_command_blocking(self.extract_async, *args, **kwargs) + + def compress(self, *args, **kwargs): + return self.run_command_blocking(self.compress_async, *args, **kwargs) + class TransportInternalError(InternalError): """Raised if there is a transport error that is raised to an internal error (e.g. diff --git a/tests/transports/test_all_plugins.py b/tests/transports/test_all_plugins.py index 805058ce6d..4ab0a5d032 100644 --- a/tests/transports/test_all_plugins.py +++ b/tests/transports/test_all_plugins.py @@ -34,13 +34,21 @@ @pytest.fixture(scope='function') def tmp_path_remote(tmp_path_factory): - """Mock the remote tmp path using tmp_path_factory to create folder start with prefix 'remote'""" + """Provides a tmp path for mocking remote computer directory environment. + + Local and remote directories must be kept separate to ensure proper functionality testing. + The created folder starts with prefix 'remote' + """ return tmp_path_factory.mktemp('remote') @pytest.fixture(scope='function') def tmp_path_local(tmp_path_factory): - """Mock the local tmp path using tmp_path_factory to create folder start with prefix 'local'""" + """Provides a tmp path for mocking local computer directory environment. + + Local and remote directories must be kept separate to ensure proper functionality testing. + The created folder starts with prefix 'local' + """ return tmp_path_factory.mktemp('local') @@ -1257,3 +1265,212 @@ def test_rename(custom_transport, tmp_path_remote): # Perform rename operation if new file already exists with pytest.raises(OSError, match='already exist|destination exists'): transport.rename(new_file, another_file) + + +def test_compress_error_handling(custom_transport: Transport, tmp_path_remote: Path, monkeypatch: pytest.MonkeyPatch): + """Test that the compress method raises an error according to instructions given in the abstract method.""" + with custom_transport as transport: + # if the format is not supported + with pytest.raises(ValueError, match='Unsupported compression format'): + transport.compress('unsupported_format', tmp_path_remote, tmp_path_remote / 'archive.tar', '/') + + # if the remotesource does not exist + with pytest.raises(OSError, match=f"{tmp_path_remote / 'non_existing'} does not exist"): + transport.compress('tar', tmp_path_remote / 'non_existing', tmp_path_remote / 'archive.tar', '/') + + # if a matching pattern if remote source is not found + with pytest.raises(OSError, match='does not exist, or a matching file/folder not found'): + transport.compress('tar', tmp_path_remote / 'non_existing*', tmp_path_remote / 'archive.tar', '/') + + # if the remotedestination already exists + Path(tmp_path_remote / 'already_exist.tar').touch() + with pytest.raises( + OSError, match=f"The remote destination {tmp_path_remote / 'already_exist.tar'} already exists." + ): + transport.compress('tar', tmp_path_remote, tmp_path_remote / 'already_exist.tar', '/', overwrite=False) + + # if the remotedestination is a directory, raise a sensible error. + with pytest.raises(OSError, match=' is a directory, should include a filename.'): + transport.compress('tar', tmp_path_remote, tmp_path_remote, '/') + + # if the root_dir is not a directory + with pytest.raises( + OSError, + match=f"The relative root {tmp_path_remote / 'non_existing_folder'} does not exist, or is not a directory.", + ): + transport.compress( + 'tar', tmp_path_remote, tmp_path_remote / 'archive.tar', tmp_path_remote / 'non_existing_folder' + ) + + # if creating the tar file fails should raise an OSError + # I do this by monkey patching transport.mock_exec_command_wait and transport.exec_command_wait_async + def mock_exec_command_wait(*args, **kwargs): + return 1, b'', b'' + + async def mock_exec_command_wait_async(*args, **kwargs): + return 1, b'', b'' + + monkeypatch.setattr(transport, 'exec_command_wait', mock_exec_command_wait) + monkeypatch.setattr(transport, 'exec_command_wait_async', mock_exec_command_wait_async) + + with pytest.raises(OSError, match='Error while creating the tar archive.'): + Path(tmp_path_remote / 'file').touch() + transport.compress('tar', tmp_path_remote, tmp_path_remote / 'archive.tar', '/') + + +@pytest.mark.parametrize('format', ['tar', 'tar.gz', 'tar.bz2', 'tar.xz']) +@pytest.mark.parametrize('dereference', [True, False]) +@pytest.mark.parametrize('file_hierarchy', [{'file.txt': 'file', 'folder': {'file_1': '1'}}]) +def test_compress_basic( + custom_transport: Transport, + format: str, + dereference: bool, + create_file_hierarchy: callable, + file_hierarchy: dict, + tmp_path_remote: Path, + tmp_path_local: Path, +) -> None: + """Test the basic functionality of the compress method. + This test will create a file hierarchy on the remote machine, compress and download it, + and then extract it on the local machine. The content of the files will be checked + to ensure that the compression and extraction was successful.""" + remote = tmp_path_remote / 'root' + remote.mkdir() + create_file_hierarchy(file_hierarchy, remote) + Path(remote / 'symlink').symlink_to(remote / 'file.txt') + + archive_name = 'archive.' + format + + # 1) basic functionality + with custom_transport as transport: + transport.compress(format, remote, tmp_path_remote / archive_name, root_dir=remote, dereference=dereference) + + transport.get(tmp_path_remote / archive_name, tmp_path_local / archive_name) + + shutil.unpack_archive(tmp_path_local / archive_name, tmp_path_local / 'extracted') + + extracted = [ + str(path.relative_to(tmp_path_local / 'extracted')) for path in (tmp_path_local / 'extracted').rglob('*') + ] + + assert sorted(extracted) == sorted(['file.txt', 'folder', 'folder/file_1', 'symlink']) + + # test that the content of the files is the same + assert Path(tmp_path_local / 'extracted' / 'file.txt').read_text() == 'file' + assert Path(tmp_path_local / 'extracted' / 'folder' / 'file_1').read_text() == '1' + + # test the symlink + if dereference: + assert not os.path.islink(tmp_path_local / 'extracted' / 'symlink') + assert Path(tmp_path_local / 'extracted' / 'symlink').read_text() == 'file' + else: + assert os.path.islink(tmp_path_local / 'extracted' / 'symlink') + assert os.readlink(tmp_path_local / 'extracted' / 'symlink') == str(remote / 'file.txt') + + +@pytest.mark.parametrize('format', ['tar', 'tar.gz', 'tar.bz2', 'tar.xz']) +@pytest.mark.parametrize( + 'file_hierarchy', + [ + { + 'file.txt': 'file', + 'folder_1': {'file_1': '1', 'file_11': '11', 'file_2': '2'}, + 'folder_2': {'file_1': '1', 'file_11': '11', 'file_2': '2'}, + } + ], +) +def test_compress_glob( + custom_transport: Transport, + create_file_hierarchy: callable, + file_hierarchy: dict, + format: str, + tmp_path_remote: Path, + tmp_path_local: Path, +) -> None: + """Test the glob functionality of the compress method. + + It is similar to :func:`test_compress_basic` but specifies a glob pattern for the remote source allowing us to test + the resolving mechanism separately.""" + + remote = tmp_path_remote / 'root' + remote.mkdir() + create_file_hierarchy(file_hierarchy, remote) + + archive_name = 'archive_glob.' + format + + with custom_transport as transport: + transport.compress( + format, + remote / 'folder*' / 'file_1*', + tmp_path_remote / archive_name, + root_dir=remote, + ) + + transport.get(tmp_path_remote / archive_name, tmp_path_local / archive_name) + + shutil.unpack_archive(tmp_path_local / archive_name, tmp_path_local / 'extracted_glob') + + extracted = [ + str(path.relative_to(tmp_path_local / 'extracted_glob')) + for path in (tmp_path_local / 'extracted_glob').rglob('*') + ] + assert sorted(extracted) == sorted( + ['folder_1', 'folder_2', 'folder_1/file_1', 'folder_1/file_11', 'folder_2/file_1', 'folder_2/file_11'] + ) + + +@pytest.mark.parametrize('format', ['tar', 'tar.gz', 'tar.bz2', 'tar.xz']) +@pytest.mark.parametrize('file_hierarchy', [{'file.txt': 'file', 'folder_1': {'file_1': '1'}}]) +def test_extract( + custom_transport: Transport, + create_file_hierarchy: callable, + file_hierarchy: dict, + format: str, + tmp_path_remote: Path, + tmp_path_local: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + local = tmp_path_local / 'root' + local.mkdir() + create_file_hierarchy(file_hierarchy, local) + + shutil_mapping_format = {'tar': 'tar', 'tar.gz': 'gztar', 'tar.bz2': 'bztar', 'tar.xz': 'xztar'} + + archive_path = shutil.make_archive(str(tmp_path_local / 'archive'), shutil_mapping_format[format], root_dir=local) + + archive_name = archive_path.split('/')[-1] + with custom_transport as transport: + transport.put(archive_path, tmp_path_remote / archive_name) + + transport.extract(tmp_path_remote / archive_name, tmp_path_remote / 'extracted') + + transport.get(tmp_path_remote / 'extracted', tmp_path_local / 'extracted') + + extracted = [ + str(path.relative_to(tmp_path_local / 'extracted')) for path in (tmp_path_local / 'extracted').rglob('*') + ] + + assert sorted(extracted) == sorted(['file.txt', 'folder_1', 'folder_1/file_1']) + + assert Path(tmp_path_local / 'extracted' / 'file.txt').read_text() == 'file' + assert Path(tmp_path_local / 'extracted' / 'folder_1' / 'file_1').read_text() == '1' + + # should raise if remotesource does not exist + with pytest.raises(OSError, match='does not exist'): + with custom_transport as transport: + transport.extract(tmp_path_remote / 'non_existing', tmp_path_remote / 'extracted') + + # should raise OSError in case extraction fails + # I do this by monkey patching transport.exec_command_wait and transport.exec_command_wait_async + def mock_exec_command_wait(*args, **kwargs): + return 1, b'', b'' + + async def mock_exec_command_wait_async(*args, **kwargs): + return 1, b'', b'' + + monkeypatch.setattr(transport, 'exec_command_wait', mock_exec_command_wait) + monkeypatch.setattr(transport, 'exec_command_wait_async', mock_exec_command_wait_async) + + with pytest.raises(OSError, match='Error while extracting the tar archive.'): + with custom_transport as transport: + transport.extract(tmp_path_remote / archive_name, tmp_path_remote / 'extracted_1')