From 18c4db7bc7c507f3125cd5e30099c51b584e92f8 Mon Sep 17 00:00:00 2001 From: ChaDongWun <66347959+lovewave02@users.noreply.github.com> Date: Thu, 13 Aug 2026 14:25:03 +0000 Subject: [PATCH 1/5] feat(native): add SFTP client API --- Changelog.rst | 3 + .../libssh2_clients/test_sftp_client.py | 43 ++++ doc/api.rst | 1 + doc/native_sftp.rst | 27 +++ pssh/clients/native/__init__.py | 1 + pssh/clients/native/sftp.py | 112 +++++++++++ pssh/clients/native/single.py | 5 + tests/test_native_sftp.py | 183 ++++++++++++++++++ 8 files changed, 375 insertions(+) create mode 100644 ci/integration_tests/libssh2_clients/test_sftp_client.py create mode 100644 doc/native_sftp.rst create mode 100644 pssh/clients/native/sftp.py create mode 100644 tests/test_native_sftp.py diff --git a/Changelog.rst b/Changelog.rst index 4de37c94..1070c413 100644 --- a/Changelog.rst +++ b/Changelog.rst @@ -7,6 +7,9 @@ Change Log Changes -------- +* Added a native ``SFTPClient`` via ``SSHClient.open_sftp`` with public remote + directory, metadata, mutation and transfer operations, plus remote current + working directory support. * All local file operations now use a thread pool to improve local file I/O performance. This includes loading private key files from a local file path, identity authentication using local files as well as SFTP read/write operations on local files. diff --git a/ci/integration_tests/libssh2_clients/test_sftp_client.py b/ci/integration_tests/libssh2_clients/test_sftp_client.py new file mode 100644 index 00000000..64933325 --- /dev/null +++ b/ci/integration_tests/libssh2_clients/test_sftp_client.py @@ -0,0 +1,43 @@ +# This file is part of parallel-ssh. +# Copyright (C) 2014-2026 Panos Kittenis. +# Copyright (C) 2014-2026 parallel-ssh Contributors. +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation, version 2.1. + +import os +import shutil +import tempfile + +from .base_ssh2_case import SSH2TestCase + + +class SFTPClientTest(SSH2TestCase): + + def test_cwd_directory_and_transfer_operations(self): + remote_root = tempfile.mkdtemp(prefix='parallel-ssh-sftp-') + local_root = tempfile.mkdtemp(prefix='parallel-ssh-local-') + local_source = os.path.join(local_root, 'source.txt') + local_copy = os.path.join(local_root, 'copy.txt') + try: + with open(local_source, 'w') as handle: + handle.write('parallel-ssh') + sftp = self.client.open_sftp() + self.assertTrue(sftp.getcwd().startswith('/')) + sftp.chdir(remote_root) + self.assertEqual(sftp.getcwd(), os.path.realpath(remote_root)) + sftp.mkdir('nested') + self.assertIn('nested', sftp.listdir('.')) + sftp.put(local_source, 'nested/remote.txt') + self.assertIn('remote.txt', sftp.listdir('nested')) + sftp.get('nested/remote.txt', local_copy) + with open(local_copy) as handle: + self.assertEqual(handle.read(), 'parallel-ssh') + sftp.rename('nested/remote.txt', 'nested/renamed.txt') + sftp.remove('nested/renamed.txt') + sftp.rmdir('nested') + self.assertNotIn('nested', sftp.listdir('.')) + finally: + shutil.rmtree(remote_root, ignore_errors=True) + shutil.rmtree(local_root, ignore_errors=True) diff --git a/doc/api.rst b/doc/api.rst index 7f45c00a..3c710fa3 100644 --- a/doc/api.rst +++ b/doc/api.rst @@ -6,6 +6,7 @@ API Documentation native_parallel native_single + native_sftp ssh_parallel ssh_single base_parallel diff --git a/doc/native_sftp.rst b/doc/native_sftp.rst new file mode 100644 index 00000000..f2a38f45 --- /dev/null +++ b/doc/native_sftp.rst @@ -0,0 +1,27 @@ +Native SFTP Client +================== + +The native client can open a user-facing SFTP client that owns one reusable +SFTP channel and tracks a remote current working directory. + +.. code-block:: python + + from pssh.clients import SSHClient + + client = SSHClient('localhost') + sftp = client.open_sftp() + sftp.chdir('/srv/uploads') + sftp.mkdir('incoming') + sftp.put('local.txt', 'incoming/remote.txt') + print(sftp.listdir('incoming')) + sftp.get('incoming/remote.txt', 'downloaded.txt') + +Relative remote paths are resolved against ``sftp.getcwd()`` using POSIX path +semantics. The SFTP client is bound to its parent ``SSHClient`` connection. +This API is available for the native ``ssh2-python`` client only; the +``pssh.clients.ssh`` backend does not currently support SFTP. + +.. automodule:: pssh.clients.native.sftp + :members: + :undoc-members: + :member-order: groupwise diff --git a/pssh/clients/native/__init__.py b/pssh/clients/native/__init__.py index 5e5f19ad..6f9ae0c8 100644 --- a/pssh/clients/native/__init__.py +++ b/pssh/clients/native/__init__.py @@ -18,3 +18,4 @@ # flake8: noqa: F401 from .parallel import ParallelSSHClient from .single import SSHClient, logger +from .sftp import SFTPClient diff --git a/pssh/clients/native/sftp.py b/pssh/clients/native/sftp.py new file mode 100644 index 00000000..45aafb58 --- /dev/null +++ b/pssh/clients/native/sftp.py @@ -0,0 +1,112 @@ +# This file is part of parallel-ssh. +# Copyright (C) 2014-2026 Panos Kittenis. +# Copyright (C) 2014-2026 parallel-ssh Contributors. +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation, version 2.1. + +import posixpath + + +class SFTPClient(object): + """User-facing SFTP operations bound to one native SSH client.""" + + __slots__ = ('_client', '_sftp', '_cwd') + + def __init__(self, client, sftp=None): + self._client = client + self._sftp = client._make_sftp() if sftp is None else sftp + self._cwd = self._canonical_path('.') + + def _canonical_path(self, path): + return self._client.eagain(self._sftp.realpath, path) + + def _remote_path(self, path): + if not isinstance(path, str): + raise TypeError("Remote path must be a string.") + if not path: + return self._cwd + if posixpath.isabs(path): + return posixpath.normpath(path) + return posixpath.normpath(posixpath.join(self._cwd, path)) + + def getcwd(self): + """Get the current remote working directory.""" + return self._cwd + + def chdir(self, path): + """Change the current remote working directory.""" + target = self._canonical_path(self._remote_path(path)) + with self._client._sftp_openfh(self._sftp.opendir, target): + pass + self._cwd = target + return self._cwd + + def listdir(self, path='.', encoding='utf-8'): + """List names in a remote directory.""" + with self._client._sftp_openfh( + self._sftp.opendir, self._remote_path(path)) as dir_h: + entries = self._client._sftp_readdir(dir_h) + names = [entry.decode(encoding) for entry in entries] + return [name for name in names if name not in ('.', '..')] + + def stat(self, path): + """Return attributes for a remote path, following symbolic links.""" + return self._client.eagain(self._sftp.stat, self._remote_path(path)) + + def lstat(self, path): + """Return attributes for a remote path without following links.""" + return self._client.eagain(self._sftp.lstat, self._remote_path(path)) + + def mkdir(self, path): + """Create a remote directory and missing parent directories.""" + return self._client.mkdir(self._sftp, self._remote_path(path)) + + def rmdir(self, path): + """Remove an empty remote directory.""" + return self._client.eagain(self._sftp.rmdir, self._remote_path(path)) + + def rename(self, source, destination): + """Rename a remote path.""" + return self._client.eagain( + self._sftp.rename, + self._remote_path(source), + self._remote_path(destination), + ) + + def remove(self, path): + """Remove a remote file.""" + return self._client.eagain(self._sftp.unlink, self._remote_path(path)) + + unlink = remove + + def get(self, remote_file, local_file): + """Copy one remote file to a local path.""" + return self._client.sftp_get( + self._sftp, self._remote_path(remote_file), local_file) + + def put(self, local_file, remote_file): + """Copy one local file to a remote path.""" + return self._client.sftp_put( + self._sftp, local_file, self._remote_path(remote_file)) + + def copy_file(self, local_file, remote_file, recurse=False): + """Copy a local file or directory to a remote path.""" + return self._client.copy_file( + local_file, + self._remote_path(remote_file), + recurse=recurse, + sftp=self._sftp, + ) + + def copy_remote_file(self, remote_file, local_file, recurse=False, + encoding='utf-8'): + """Copy a remote file or directory to a local path.""" + return self._client.copy_remote_file( + self._remote_path(remote_file), + local_file, + recurse=recurse, + sftp=self._sftp, + encoding=encoding, + ) diff --git a/pssh/clients/native/single.py b/pssh/clients/native/single.py index ee1d9855..7a4446a1 100644 --- a/pssh/clients/native/single.py +++ b/pssh/clients/native/single.py @@ -35,6 +35,7 @@ LIBSSH2_SFTP_S_IXGRP, LIBSSH2_SFTP_S_IXOTH from .tunnel import FORWARDER +from .sftp import SFTPClient from ..base.single import BaseSSHClient, PollMixIn from ...constants import DEFAULT_RETRIES, RETRY_DELAY from ...exceptions import SessionError, SFTPError, \ @@ -441,6 +442,10 @@ def _make_sftp(self): raise SFTPError(ex) return sftp + def open_sftp(self): + """Open a user-facing SFTP client bound to this SSH session.""" + return SFTPClient(self) + def _mkdir(self, sftp, directory): """Make directory via SFTP channel. diff --git a/tests/test_native_sftp.py b/tests/test_native_sftp.py new file mode 100644 index 00000000..c24750a7 --- /dev/null +++ b/tests/test_native_sftp.py @@ -0,0 +1,183 @@ +# This file is part of parallel-ssh. +# Copyright (C) 2014-2026 Panos Kittenis. +# Copyright (C) 2014-2026 parallel-ssh Contributors. +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation, version 2.1. + +import unittest + +from pssh.clients.native.sftp import SFTPClient + + +class DirectoryHandle(object): + + def __init__(self): + self.closed = False + + def __enter__(self): + return self + + def __exit__(self, *_args): + self.closed = True + + +class SFTP(object): + + def __init__(self): + self.realpath_calls = [] + self.opendir_calls = [] + self.handles = [] + self.calls = [] + + def realpath(self, path): + self.realpath_calls.append(path) + return '/home/tester' if path == '.' else path + + def opendir(self, path): + self.opendir_calls.append(path) + handle = DirectoryHandle() + self.handles.append(handle) + return handle + + def stat(self, path): + self.calls.append(('stat', path)) + return 'stat-result' + + def lstat(self, path): + self.calls.append(('lstat', path)) + return 'lstat-result' + + def rmdir(self, path): + self.calls.append(('rmdir', path)) + return 0 + + def rename(self, source, destination): + self.calls.append(('rename', source, destination)) + return 0 + + def unlink(self, path): + self.calls.append(('unlink', path)) + return 0 + + +class SSHClient(object): + + def __init__(self, sftp): + self.sftp = sftp + + def _make_sftp(self): + return self.sftp + + def eagain(self, func, *args): + return func(*args) + + def _sftp_openfh(self, func, *args): + return func(*args) + + def _sftp_readdir(self, _handle): + return iter((b'.', b'..', b'file.txt', b'data')) + + def mkdir(self, sftp, path): + self.calls = getattr(self, 'calls', []) + self.calls.append(('mkdir', sftp, path)) + + def sftp_get(self, sftp, remote_file, local_file): + self.calls = getattr(self, 'calls', []) + self.calls.append(('get', sftp, remote_file, local_file)) + + def sftp_put(self, sftp, local_file, remote_file): + self.calls = getattr(self, 'calls', []) + self.calls.append(('put', sftp, local_file, remote_file)) + + def copy_file(self, local_file, remote_file, recurse=False, sftp=None): + self.calls = getattr(self, 'calls', []) + self.calls.append( + ('copy_file', sftp, local_file, remote_file, recurse)) + + def copy_remote_file(self, remote_file, local_file, recurse=False, + sftp=None, encoding='utf-8'): + self.calls = getattr(self, 'calls', []) + self.calls.append( + ('copy_remote_file', sftp, remote_file, local_file, + recurse, encoding)) + + +class NativeSFTPClientTest(unittest.TestCase): + + def setUp(self): + self.sftp = SFTP() + self.ssh_client = SSHClient(self.sftp) + self.client = SFTPClient(self.ssh_client) + + def test_initial_cwd_uses_server_realpath(self): + self.assertEqual(self.client.getcwd(), '/home/tester') + self.assertEqual(self.sftp.realpath_calls, ['.']) + + def test_remote_path_uses_posix_semantics(self): + self.assertEqual( + self.client._remote_path('../shared/./file'), '/home/shared/file') + self.assertEqual( + self.client._remote_path('/var//data/../log'), '/var/log') + + def test_chdir_canonicalizes_and_verifies_directory(self): + cwd = self.client.chdir('data') + + self.assertEqual(cwd, '/home/tester/data') + self.assertEqual(self.client.getcwd(), '/home/tester/data') + self.assertEqual(self.sftp.opendir_calls, ['/home/tester/data']) + self.assertTrue(self.sftp.handles[0].closed) + + def test_invalid_path_type_fails_before_transport(self): + with self.assertRaises(TypeError): + self.client.chdir(None) + + self.assertEqual(self.sftp.opendir_calls, []) + + def test_listdir_filters_navigation_entries(self): + self.assertEqual(self.client.listdir('data'), ['file.txt', 'data']) + self.assertEqual(self.sftp.opendir_calls, ['/home/tester/data']) + + def test_metadata_and_mutations_resolve_remote_paths(self): + self.assertEqual(self.client.stat('file'), 'stat-result') + self.assertEqual(self.client.lstat('../link'), 'lstat-result') + self.client.rmdir('empty') + self.client.rename('old', '../new') + self.client.remove('obsolete') + + self.assertEqual( + self.sftp.calls, + [ + ('stat', '/home/tester/file'), + ('lstat', '/home/link'), + ('rmdir', '/home/tester/empty'), + ('rename', '/home/tester/old', '/home/new'), + ('unlink', '/home/tester/obsolete'), + ], + ) + + def test_transfer_helpers_reuse_bound_channel_and_cwd(self): + self.client.mkdir('new/child') + self.client.get('remote.txt', 'local.txt') + self.client.put('local.bin', '../remote.bin') + self.client.copy_file('tree', 'remote-tree', recurse=True) + self.client.copy_remote_file( + 'remote-tree', 'local-tree', recurse=True, encoding='ascii') + + self.assertEqual( + self.ssh_client.calls, + [ + ('mkdir', self.sftp, '/home/tester/new/child'), + ('get', self.sftp, '/home/tester/remote.txt', 'local.txt'), + ('put', self.sftp, 'local.bin', '/home/remote.bin'), + ('copy_file', self.sftp, 'tree', + '/home/tester/remote-tree', True), + ('copy_remote_file', self.sftp, + '/home/tester/remote-tree', 'local-tree', True, 'ascii'), + ], + ) + + +if __name__ == '__main__': + unittest.main() From 5b43c07fbc69aeb6622b1ea623ad2375c3ac0f4a Mon Sep 17 00:00:00 2001 From: lovewave02 Date: Sat, 22 Aug 2026 19:16:31 +0900 Subject: [PATCH 2/5] Apply verified ParallelSSH PR #410 patch --- Changelog.rst | 5 +- pssh/clients/native/__init__.py | 2 +- pssh/clients/native/parallel.py | 135 ++++++++++++++++++++++ tests/test_native_parallel.py | 191 ++++++++++++++++++++++++++++++++ 4 files changed, 330 insertions(+), 3 deletions(-) create mode 100644 tests/test_native_parallel.py diff --git a/Changelog.rst b/Changelog.rst index 1070c413..e92b3624 100644 --- a/Changelog.rst +++ b/Changelog.rst @@ -7,9 +7,10 @@ Change Log Changes -------- -* Added a native ``SFTPClient`` via ``SSHClient.open_sftp`` with public remote +* Added native ``SFTPClient`` and ``ParallelSFTPClient`` APIs with public remote directory, metadata, mutation and transfer operations, plus remote current - working directory support. + working directory support. ``ParallelSFTPClient`` runs operations on all + configured hosts concurrently and preserves configured host order. * All local file operations now use a thread pool to improve local file I/O performance. This includes loading private key files from a local file path, identity authentication using local files as well as SFTP read/write operations on local files. diff --git a/pssh/clients/native/__init__.py b/pssh/clients/native/__init__.py index 6f9ae0c8..4b51a167 100644 --- a/pssh/clients/native/__init__.py +++ b/pssh/clients/native/__init__.py @@ -16,6 +16,6 @@ # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA # flake8: noqa: F401 -from .parallel import ParallelSSHClient +from .parallel import ParallelSFTPClient, ParallelSSHClient from .single import SSHClient, logger from .sftp import SFTPClient diff --git a/pssh/clients/native/parallel.py b/pssh/clients/native/parallel.py index 2612f548..c2a5bba5 100644 --- a/pssh/clients/native/parallel.py +++ b/pssh/clients/native/parallel.py @@ -17,7 +17,10 @@ import logging +from gevent import Timeout as GTimeout + from .single import SSHClient +from .sftp import SFTPClient from ..base.parallel import BaseParallelSSHClient from ..common import _validate_pkey from ...constants import DEFAULT_RETRIES, RETRY_DELAY @@ -513,3 +516,135 @@ def scp_recv(self, remote_file, local_file, recurse=False, copy_args=None, raise HostArgumentError( "Number of per-host copy arguments provided does not match " "number of hosts") + + +class ParallelSFTPClient(ParallelSSHClient): + """Run native SFTP operations on a collection of SSH hosts.""" + + @property + def hosts(self): + return self._hosts + + @hosts.setter + def hosts(self, hosts): + BaseParallelSSHClient.hosts.fset(self, hosts) + if hasattr(self, '_sftp_clients'): + self._sftp_clients.clear() + + def _make_sftp_client(self, host_i, host): + ssh_client = self._get_ssh_client(host_i, host) + return SFTPClient(ssh_client) + + def _get_sftp_client(self, host_i, host): + try: + clients = self._sftp_clients + except AttributeError: + clients = self._sftp_clients = {} + key = (host_i, host) + client = clients.get(key) + if client is None: + client = self._make_sftp_client(host_i, host) + clients[key] = client + return client + + @staticmethod + def _collect(tasks, stop_on_errors): + results = [] + for task in tasks: + try: + results.append(task.get()) + except (GTimeout, Exception) as exc: + if stop_on_errors: + raise + results.append(exc) + return results + + def _run_sftp_operation(self, host_i, host, operation, args, kwargs): + client = self._get_sftp_client(host_i, host) + result = getattr(client, operation)(*args, **kwargs) + if operation == 'listdir': + return list(result) + return result + + def _run_parallel(self, operation, args=(), kwargs=None, + stop_on_errors=True): + kwargs = {} if kwargs is None else kwargs + tasks = [self.pool.spawn( + self._run_sftp_operation, host_i, host, operation, args, kwargs) + for host_i, host in enumerate(self.hosts)] + return self._collect(tasks, stop_on_errors) + + def connect(self, stop_on_errors=True): + """Create and return one :class:`SFTPClient` per configured host. + + Connections are initialized concurrently using this client's pool and + cached for subsequent parallel operations. Results always follow + configured host order. When ``stop_on_errors`` is false, an exception + is returned in place of the failed host's client instead of being + raised. + + :param stop_on_errors: Raise SFTP initialization errors when true. + :type stop_on_errors: bool + :rtype: list(:class:`SFTPClient` or Exception) + """ + tasks = [self.pool.spawn(self._get_sftp_client, host_i, host) + for host_i, host in enumerate(self.hosts)] + return self._collect(tasks, stop_on_errors) + + def getcwd(self, stop_on_errors=True): + """Return the current remote directory for every host.""" + return self._run_parallel('getcwd', stop_on_errors=stop_on_errors) + + def chdir(self, path, stop_on_errors=True): + """Change the current remote directory on every host.""" + return self._run_parallel( + 'chdir', (path,), stop_on_errors=stop_on_errors) + + def listdir(self, path='.', encoding='utf-8', stop_on_errors=True): + """Return directory entry lists from every host.""" + return self._run_parallel( + 'listdir', (path, encoding), stop_on_errors=stop_on_errors) + + def stat(self, path, stop_on_errors=True): + """Return attributes for a remote path on every host.""" + return self._run_parallel( + 'stat', (path,), stop_on_errors=stop_on_errors) + + def lstat(self, path, stop_on_errors=True): + """Return attributes without following links on every host.""" + return self._run_parallel( + 'lstat', (path,), stop_on_errors=stop_on_errors) + + def mkdir(self, path, stop_on_errors=True): + """Create a remote directory on every host.""" + return self._run_parallel( + 'mkdir', (path,), stop_on_errors=stop_on_errors) + + def rmdir(self, path, stop_on_errors=True): + """Remove an empty remote directory on every host.""" + return self._run_parallel( + 'rmdir', (path,), stop_on_errors=stop_on_errors) + + def rename(self, source, destination, stop_on_errors=True): + """Rename a remote path on every host.""" + return self._run_parallel( + 'rename', (source, destination), stop_on_errors=stop_on_errors) + + def remove(self, path, stop_on_errors=True): + """Remove a remote file on every host.""" + return self._run_parallel( + 'remove', (path,), stop_on_errors=stop_on_errors) + + unlink = remove + + def get(self, remote_file, local_file, stop_on_errors=True): + """Copy a remote file from every host to a local path.""" + return self._run_parallel( + 'get', (remote_file, local_file), + stop_on_errors=stop_on_errors) + + def put(self, local_file, remote_file, stop_on_errors=True): + """Copy a local file to every host.""" + return self._run_parallel( + 'put', (local_file, remote_file), + stop_on_errors=stop_on_errors) diff --git a/tests/test_native_parallel.py b/tests/test_native_parallel.py new file mode 100644 index 00000000..6e80621d --- /dev/null +++ b/tests/test_native_parallel.py @@ -0,0 +1,191 @@ +# This file is part of parallel-ssh. +# Copyright (C) 2014-2025 Panos Kittenis. +# Copyright (C) 2014-2025 parallel-ssh Contributors. +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation, version 2.1. + +from unittest import TestCase +from unittest.mock import Mock, call, patch + +from gevent import Timeout + +from pssh.clients.native import ParallelSFTPClient +from pssh.clients.native import parallel + + +class DeferredTask: + + def __init__(self, func, *args): + self.func = func + self.args = args + + def get(self): + return self.func(*self.args) + + +class RecordingPool: + + def __init__(self): + self.calls = [] + + def spawn(self, func, *args): + self.calls.append(args) + return DeferredTask(func, *args) + + +class ParallelSFTPClientTest(TestCase): + + def make_client(self, hosts, host_clients): + client = Mock() + client.hosts = hosts + client.pool = RecordingPool() + client._get_ssh_client = Mock( + side_effect=lambda host_i, _host: host_clients[host_i]) + client._make_sftp_client = ParallelSFTPClient._make_sftp_client.__get__( + client) + client._get_sftp_client = ParallelSFTPClient._get_sftp_client.__get__( + client) + client._collect = ParallelSFTPClient._collect + client._sftp_clients = {} + return client + + @patch.object(parallel, 'SFTPClient') + def test_connect_creates_sftp_clients_in_host_order(self, sftp_client): + hosts = ['host-b', 'host-a', 'host-c'] + host_clients = [Mock(name=host) for host in hosts] + sftp_clients = [Mock(name='%s-sftp' % host) for host in hosts] + sftp_client.side_effect = sftp_clients + client = self.make_client(hosts, host_clients) + + result = ParallelSFTPClient.connect(client) + + self.assertEqual(sftp_clients, result) + self.assertEqual([(0, 'host-b'), (1, 'host-a'), (2, 'host-c')], + client.pool.calls) + self.assertEqual([call(host_client) for host_client in host_clients], + sftp_client.call_args_list) + + self.assertEqual(sftp_clients, ParallelSFTPClient.connect(client)) + self.assertEqual(len(hosts), sftp_client.call_count) + + @patch.object(parallel, 'SFTPClient') + def test_connect_returns_errors_in_host_order_when_not_stopping(self, + sftp_client): + hosts = ['first', 'failing', 'last'] + host_clients = [Mock(name=host) for host in hosts] + first = Mock(name='first-sftp') + error = RuntimeError('SFTP unavailable') + last = Mock(name='last-sftp') + sftp_client.side_effect = [first, error, last] + client = self.make_client(hosts, host_clients) + + result = ParallelSFTPClient.connect(client, stop_on_errors=False) + + self.assertEqual([first, error, last], result) + + @patch.object(parallel, 'SFTPClient') + def test_connect_raises_when_stopping_on_errors(self, sftp_client): + error = RuntimeError('SFTP unavailable') + sftp_client.side_effect = [Mock(name='first-sftp'), error] + client = self.make_client( + ['first', 'failing'], [Mock(name='first'), Mock(name='failing')]) + + with self.assertRaisesRegex(RuntimeError, 'SFTP unavailable'): + ParallelSFTPClient.connect(client, stop_on_errors=True) + + @patch.object(parallel, 'SFTPClient') + def test_operations_run_for_every_host_and_preserve_result_order( + self, sftp_client): + hosts = ['host-b', 'host-a'] + host_clients = [Mock(name=host) for host in hosts] + sftp_clients = [Mock(name='%s-sftp' % host) for host in hosts] + sftp_client.side_effect = sftp_clients + client = self.make_client(hosts, host_clients) + client._run_sftp_operation = \ + ParallelSFTPClient._run_sftp_operation.__get__(client) + client._run_parallel = ParallelSFTPClient._run_parallel.__get__(client) + sftp_clients[0].listdir.return_value = iter(['b-one', 'b-two']) + sftp_clients[1].listdir.return_value = iter(['a-one']) + + result = ParallelSFTPClient.listdir(client, 'data') + + self.assertEqual([['b-one', 'b-two'], ['a-one']], result) + self.assertEqual( + [call('data', 'utf-8'), call('data', 'utf-8')], + [sftp.listdir.call_args for sftp in sftp_clients], + ) + self.assertEqual(len(hosts), sftp_client.call_count) + + self.assertEqual( + [sftp.getcwd.return_value for sftp in sftp_clients], + ParallelSFTPClient.getcwd(client), + ) + self.assertEqual(len(hosts), sftp_client.call_count) + + @patch.object(parallel, 'SFTPClient') + def test_operation_errors_follow_stop_on_errors(self, sftp_client): + hosts = ['first', 'failing', 'last'] + sftp_clients = [Mock(name='%s-sftp' % host) for host in hosts] + error = RuntimeError('stat unavailable') + sftp_clients[0].stat.return_value = 'first-stat' + sftp_clients[1].stat.side_effect = error + sftp_clients[2].stat.return_value = 'last-stat' + sftp_client.side_effect = sftp_clients + client = self.make_client(hosts, [Mock(name=host) for host in hosts]) + client._run_sftp_operation = \ + ParallelSFTPClient._run_sftp_operation.__get__(client) + client._run_parallel = ParallelSFTPClient._run_parallel.__get__(client) + + result = ParallelSFTPClient.stat( + client, 'target', stop_on_errors=False) + + self.assertEqual(['first-stat', error, 'last-stat'], result) + + with self.assertRaisesRegex(RuntimeError, 'stat unavailable'): + ParallelSFTPClient.stat(client, 'target') + + def test_collect_can_return_gevent_timeout(self): + timeout = Timeout(1) + + def raise_timeout(): + raise timeout + + result = ParallelSFTPClient._collect( + [DeferredTask(raise_timeout)], stop_on_errors=False) + + self.assertEqual([timeout], result) + + def test_public_operations_forward_arguments(self): + client = Mock() + operations = [ + ('chdir', ('directory',), ('chdir', ('directory',))), + ('stat', ('file',), ('stat', ('file',))), + ('lstat', ('link',), ('lstat', ('link',))), + ('mkdir', ('directory',), ('mkdir', ('directory',))), + ('rmdir', ('directory',), ('rmdir', ('directory',))), + ('rename', ('old', 'new'), ('rename', ('old', 'new'))), + ('remove', ('file',), ('remove', ('file',))), + ('unlink', ('file',), ('remove', ('file',))), + ('get', ('remote', 'local'), ('get', ('remote', 'local'))), + ('put', ('local', 'remote'), ('put', ('local', 'remote'))), + ] + + for method, args, forwarded in operations: + with self.subTest(method=method): + getattr(ParallelSFTPClient, method)(client, *args) + client._run_parallel.assert_called_once_with( + forwarded[0], forwarded[1], stop_on_errors=True) + client._run_parallel.reset_mock() + + def test_changing_hosts_invalidates_sftp_clients(self): + client = object.__new__(ParallelSFTPClient) + client._hosts = ['old-host'] + client._host_clients = {(0, 'old-host'): Mock()} + client._sftp_clients = {(0, 'old-host'): Mock()} + + client.hosts = ['new-host'] + + self.assertEqual({}, client._host_clients) + self.assertEqual({}, client._sftp_clients) From f331a75a8e96f4774c82241d3dfe4f4f00f68ff8 Mon Sep 17 00:00:00 2001 From: lovewave02 Date: Tue, 1 Sep 2026 02:35:59 +0900 Subject: [PATCH 3/5] Implement single-host SFTPClient with generator-based listdir --- Changelog.rst | 7 +- pssh/clients/native/__init__.py | 2 +- pssh/clients/native/parallel.py | 135 ---------------------- pssh/clients/native/sftp.py | 9 +- pssh/clients/native/single.py | 14 ++- tests/test_native_parallel.py | 191 -------------------------------- tests/test_native_sftp.py | 14 ++- tests/test_native_single.py | 50 +++++++++ 8 files changed, 78 insertions(+), 344 deletions(-) delete mode 100644 tests/test_native_parallel.py diff --git a/Changelog.rst b/Changelog.rst index e92b3624..5e957ebf 100644 --- a/Changelog.rst +++ b/Changelog.rst @@ -7,10 +7,9 @@ Change Log Changes -------- -* Added native ``SFTPClient`` and ``ParallelSFTPClient`` APIs with public remote - directory, metadata, mutation and transfer operations, plus remote current - working directory support. ``ParallelSFTPClient`` runs operations on all - configured hosts concurrently and preserves configured host order. +* Added a native ``SFTPClient`` API with public remote directory, metadata, + mutation and transfer operations, plus remote current working directory + support. * All local file operations now use a thread pool to improve local file I/O performance. This includes loading private key files from a local file path, identity authentication using local files as well as SFTP read/write operations on local files. diff --git a/pssh/clients/native/__init__.py b/pssh/clients/native/__init__.py index 4b51a167..6f9ae0c8 100644 --- a/pssh/clients/native/__init__.py +++ b/pssh/clients/native/__init__.py @@ -16,6 +16,6 @@ # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA # flake8: noqa: F401 -from .parallel import ParallelSFTPClient, ParallelSSHClient +from .parallel import ParallelSSHClient from .single import SSHClient, logger from .sftp import SFTPClient diff --git a/pssh/clients/native/parallel.py b/pssh/clients/native/parallel.py index c2a5bba5..2612f548 100644 --- a/pssh/clients/native/parallel.py +++ b/pssh/clients/native/parallel.py @@ -17,10 +17,7 @@ import logging -from gevent import Timeout as GTimeout - from .single import SSHClient -from .sftp import SFTPClient from ..base.parallel import BaseParallelSSHClient from ..common import _validate_pkey from ...constants import DEFAULT_RETRIES, RETRY_DELAY @@ -516,135 +513,3 @@ def scp_recv(self, remote_file, local_file, recurse=False, copy_args=None, raise HostArgumentError( "Number of per-host copy arguments provided does not match " "number of hosts") - - -class ParallelSFTPClient(ParallelSSHClient): - """Run native SFTP operations on a collection of SSH hosts.""" - - @property - def hosts(self): - return self._hosts - - @hosts.setter - def hosts(self, hosts): - BaseParallelSSHClient.hosts.fset(self, hosts) - if hasattr(self, '_sftp_clients'): - self._sftp_clients.clear() - - def _make_sftp_client(self, host_i, host): - ssh_client = self._get_ssh_client(host_i, host) - return SFTPClient(ssh_client) - - def _get_sftp_client(self, host_i, host): - try: - clients = self._sftp_clients - except AttributeError: - clients = self._sftp_clients = {} - key = (host_i, host) - client = clients.get(key) - if client is None: - client = self._make_sftp_client(host_i, host) - clients[key] = client - return client - - @staticmethod - def _collect(tasks, stop_on_errors): - results = [] - for task in tasks: - try: - results.append(task.get()) - except (GTimeout, Exception) as exc: - if stop_on_errors: - raise - results.append(exc) - return results - - def _run_sftp_operation(self, host_i, host, operation, args, kwargs): - client = self._get_sftp_client(host_i, host) - result = getattr(client, operation)(*args, **kwargs) - if operation == 'listdir': - return list(result) - return result - - def _run_parallel(self, operation, args=(), kwargs=None, - stop_on_errors=True): - kwargs = {} if kwargs is None else kwargs - tasks = [self.pool.spawn( - self._run_sftp_operation, host_i, host, operation, args, kwargs) - for host_i, host in enumerate(self.hosts)] - return self._collect(tasks, stop_on_errors) - - def connect(self, stop_on_errors=True): - """Create and return one :class:`SFTPClient` per configured host. - - Connections are initialized concurrently using this client's pool and - cached for subsequent parallel operations. Results always follow - configured host order. When ``stop_on_errors`` is false, an exception - is returned in place of the failed host's client instead of being - raised. - - :param stop_on_errors: Raise SFTP initialization errors when true. - :type stop_on_errors: bool - :rtype: list(:class:`SFTPClient` or Exception) - """ - tasks = [self.pool.spawn(self._get_sftp_client, host_i, host) - for host_i, host in enumerate(self.hosts)] - return self._collect(tasks, stop_on_errors) - - def getcwd(self, stop_on_errors=True): - """Return the current remote directory for every host.""" - return self._run_parallel('getcwd', stop_on_errors=stop_on_errors) - - def chdir(self, path, stop_on_errors=True): - """Change the current remote directory on every host.""" - return self._run_parallel( - 'chdir', (path,), stop_on_errors=stop_on_errors) - - def listdir(self, path='.', encoding='utf-8', stop_on_errors=True): - """Return directory entry lists from every host.""" - return self._run_parallel( - 'listdir', (path, encoding), stop_on_errors=stop_on_errors) - - def stat(self, path, stop_on_errors=True): - """Return attributes for a remote path on every host.""" - return self._run_parallel( - 'stat', (path,), stop_on_errors=stop_on_errors) - - def lstat(self, path, stop_on_errors=True): - """Return attributes without following links on every host.""" - return self._run_parallel( - 'lstat', (path,), stop_on_errors=stop_on_errors) - - def mkdir(self, path, stop_on_errors=True): - """Create a remote directory on every host.""" - return self._run_parallel( - 'mkdir', (path,), stop_on_errors=stop_on_errors) - - def rmdir(self, path, stop_on_errors=True): - """Remove an empty remote directory on every host.""" - return self._run_parallel( - 'rmdir', (path,), stop_on_errors=stop_on_errors) - - def rename(self, source, destination, stop_on_errors=True): - """Rename a remote path on every host.""" - return self._run_parallel( - 'rename', (source, destination), stop_on_errors=stop_on_errors) - - def remove(self, path, stop_on_errors=True): - """Remove a remote file on every host.""" - return self._run_parallel( - 'remove', (path,), stop_on_errors=stop_on_errors) - - unlink = remove - - def get(self, remote_file, local_file, stop_on_errors=True): - """Copy a remote file from every host to a local path.""" - return self._run_parallel( - 'get', (remote_file, local_file), - stop_on_errors=stop_on_errors) - - def put(self, local_file, remote_file, stop_on_errors=True): - """Copy a local file to every host.""" - return self._run_parallel( - 'put', (local_file, remote_file), - stop_on_errors=stop_on_errors) diff --git a/pssh/clients/native/sftp.py b/pssh/clients/native/sftp.py index 45aafb58..abf76fe1 100644 --- a/pssh/clients/native/sftp.py +++ b/pssh/clients/native/sftp.py @@ -16,7 +16,7 @@ class SFTPClient(object): def __init__(self, client, sftp=None): self._client = client - self._sftp = client._make_sftp() if sftp is None else sftp + self._sftp = client.make_sftp_client() if sftp is None else sftp self._cwd = self._canonical_path('.') def _canonical_path(self, path): @@ -47,9 +47,10 @@ def listdir(self, path='.', encoding='utf-8'): """List names in a remote directory.""" with self._client._sftp_openfh( self._sftp.opendir, self._remote_path(path)) as dir_h: - entries = self._client._sftp_readdir(dir_h) - names = [entry.decode(encoding) for entry in entries] - return [name for name in names if name not in ('.', '..')] + for entry in self._client._sftp_readdir(dir_h): + name = entry.decode(encoding) + if name not in ('.', '..'): + yield name def stat(self, path): """Return attributes for a remote path, following symbolic links.""" diff --git a/pssh/clients/native/single.py b/pssh/clients/native/single.py index 7a4446a1..a5a55ac1 100644 --- a/pssh/clients/native/single.py +++ b/pssh/clients/native/single.py @@ -435,13 +435,15 @@ def eagain(self, func, *args, **kwargs): def _make_sftp_eagain(self): return self.eagain(self.session.sftp_init) - def _make_sftp(self): + def make_sftp_client(self): try: sftp = self._make_sftp_eagain() except Exception as ex: raise SFTPError(ex) return sftp + _make_sftp = make_sftp_client + def open_sftp(self): """Open a user-facing SFTP client bound to this SSH session.""" return SFTPClient(self) @@ -493,7 +495,7 @@ def copy_file(self, local_file, remote_file, recurse=False, sftp=None): :raises: :py:class:`IOError` on local file IO errors :raises: :py:class:`OSError` on local OS errors like permission denied """ - sftp = self._make_sftp() if sftp is None else sftp + sftp = self.make_sftp_client() if sftp is None else sftp if os.path.isdir(local_file) and recurse: return self._copy_dir(local_file, remote_file, sftp) elif os.path.isdir(local_file) and not recurse: @@ -595,7 +597,7 @@ def copy_remote_file(self, remote_file, local_file, recurse=False, :raises: :py:class:`IOError` on local file IO errors :raises: :py:class:`OSError` on local OS errors like permission denied """ - sftp = self._make_sftp() if sftp is None else sftp + sftp = self.make_sftp_client() if sftp is None else sftp try: self.eagain(sftp.stat, remote_file) except (SFTPHandleError, SFTPProtocolError): @@ -670,7 +672,7 @@ def scp_recv(self, remote_file, local_file, recurse=False, sftp=None, :raises: :py:class:`OSError` on local OS errors like permission denied. """ if recurse: - sftp = self._make_sftp() if sftp is None else sftp + sftp = self.make_sftp_client() if sftp is None else sftp return self._scp_recv_recursive(remote_file, local_file, sftp, encoding=encoding) elif local_file.endswith('/'): remote_filename = remote_file.rsplit('/')[-1] @@ -734,7 +736,7 @@ def scp_send(self, local_file, remote_file, recurse=False, sftp=None): :raises: :py:class:`OSError` on local OS errors like permission denied """ if os.path.isdir(local_file) and recurse: - sftp = self._make_sftp() if sftp is None else sftp + sftp = self.make_sftp_client() if sftp is None else sftp return self._scp_send_dir(local_file, remote_file, sftp) elif os.path.isdir(local_file) and not recurse: raise ValueError("Recurse must be True if local_file is a " @@ -742,7 +744,7 @@ def scp_send(self, local_file, remote_file, recurse=False, sftp=None): if recurse: destination = self._remote_paths_split(remote_file) if destination is not None: - sftp = self._make_sftp() if sftp is None else sftp + sftp = self.make_sftp_client() if sftp is None else sftp try: self.eagain(sftp.stat, destination) except (SFTPHandleError, SFTPProtocolError): diff --git a/tests/test_native_parallel.py b/tests/test_native_parallel.py deleted file mode 100644 index 6e80621d..00000000 --- a/tests/test_native_parallel.py +++ /dev/null @@ -1,191 +0,0 @@ -# This file is part of parallel-ssh. -# Copyright (C) 2014-2025 Panos Kittenis. -# Copyright (C) 2014-2025 parallel-ssh Contributors. -# -# This library is free software; you can redistribute it and/or -# modify it under the terms of the GNU Lesser General Public -# License as published by the Free Software Foundation, version 2.1. - -from unittest import TestCase -from unittest.mock import Mock, call, patch - -from gevent import Timeout - -from pssh.clients.native import ParallelSFTPClient -from pssh.clients.native import parallel - - -class DeferredTask: - - def __init__(self, func, *args): - self.func = func - self.args = args - - def get(self): - return self.func(*self.args) - - -class RecordingPool: - - def __init__(self): - self.calls = [] - - def spawn(self, func, *args): - self.calls.append(args) - return DeferredTask(func, *args) - - -class ParallelSFTPClientTest(TestCase): - - def make_client(self, hosts, host_clients): - client = Mock() - client.hosts = hosts - client.pool = RecordingPool() - client._get_ssh_client = Mock( - side_effect=lambda host_i, _host: host_clients[host_i]) - client._make_sftp_client = ParallelSFTPClient._make_sftp_client.__get__( - client) - client._get_sftp_client = ParallelSFTPClient._get_sftp_client.__get__( - client) - client._collect = ParallelSFTPClient._collect - client._sftp_clients = {} - return client - - @patch.object(parallel, 'SFTPClient') - def test_connect_creates_sftp_clients_in_host_order(self, sftp_client): - hosts = ['host-b', 'host-a', 'host-c'] - host_clients = [Mock(name=host) for host in hosts] - sftp_clients = [Mock(name='%s-sftp' % host) for host in hosts] - sftp_client.side_effect = sftp_clients - client = self.make_client(hosts, host_clients) - - result = ParallelSFTPClient.connect(client) - - self.assertEqual(sftp_clients, result) - self.assertEqual([(0, 'host-b'), (1, 'host-a'), (2, 'host-c')], - client.pool.calls) - self.assertEqual([call(host_client) for host_client in host_clients], - sftp_client.call_args_list) - - self.assertEqual(sftp_clients, ParallelSFTPClient.connect(client)) - self.assertEqual(len(hosts), sftp_client.call_count) - - @patch.object(parallel, 'SFTPClient') - def test_connect_returns_errors_in_host_order_when_not_stopping(self, - sftp_client): - hosts = ['first', 'failing', 'last'] - host_clients = [Mock(name=host) for host in hosts] - first = Mock(name='first-sftp') - error = RuntimeError('SFTP unavailable') - last = Mock(name='last-sftp') - sftp_client.side_effect = [first, error, last] - client = self.make_client(hosts, host_clients) - - result = ParallelSFTPClient.connect(client, stop_on_errors=False) - - self.assertEqual([first, error, last], result) - - @patch.object(parallel, 'SFTPClient') - def test_connect_raises_when_stopping_on_errors(self, sftp_client): - error = RuntimeError('SFTP unavailable') - sftp_client.side_effect = [Mock(name='first-sftp'), error] - client = self.make_client( - ['first', 'failing'], [Mock(name='first'), Mock(name='failing')]) - - with self.assertRaisesRegex(RuntimeError, 'SFTP unavailable'): - ParallelSFTPClient.connect(client, stop_on_errors=True) - - @patch.object(parallel, 'SFTPClient') - def test_operations_run_for_every_host_and_preserve_result_order( - self, sftp_client): - hosts = ['host-b', 'host-a'] - host_clients = [Mock(name=host) for host in hosts] - sftp_clients = [Mock(name='%s-sftp' % host) for host in hosts] - sftp_client.side_effect = sftp_clients - client = self.make_client(hosts, host_clients) - client._run_sftp_operation = \ - ParallelSFTPClient._run_sftp_operation.__get__(client) - client._run_parallel = ParallelSFTPClient._run_parallel.__get__(client) - sftp_clients[0].listdir.return_value = iter(['b-one', 'b-two']) - sftp_clients[1].listdir.return_value = iter(['a-one']) - - result = ParallelSFTPClient.listdir(client, 'data') - - self.assertEqual([['b-one', 'b-two'], ['a-one']], result) - self.assertEqual( - [call('data', 'utf-8'), call('data', 'utf-8')], - [sftp.listdir.call_args for sftp in sftp_clients], - ) - self.assertEqual(len(hosts), sftp_client.call_count) - - self.assertEqual( - [sftp.getcwd.return_value for sftp in sftp_clients], - ParallelSFTPClient.getcwd(client), - ) - self.assertEqual(len(hosts), sftp_client.call_count) - - @patch.object(parallel, 'SFTPClient') - def test_operation_errors_follow_stop_on_errors(self, sftp_client): - hosts = ['first', 'failing', 'last'] - sftp_clients = [Mock(name='%s-sftp' % host) for host in hosts] - error = RuntimeError('stat unavailable') - sftp_clients[0].stat.return_value = 'first-stat' - sftp_clients[1].stat.side_effect = error - sftp_clients[2].stat.return_value = 'last-stat' - sftp_client.side_effect = sftp_clients - client = self.make_client(hosts, [Mock(name=host) for host in hosts]) - client._run_sftp_operation = \ - ParallelSFTPClient._run_sftp_operation.__get__(client) - client._run_parallel = ParallelSFTPClient._run_parallel.__get__(client) - - result = ParallelSFTPClient.stat( - client, 'target', stop_on_errors=False) - - self.assertEqual(['first-stat', error, 'last-stat'], result) - - with self.assertRaisesRegex(RuntimeError, 'stat unavailable'): - ParallelSFTPClient.stat(client, 'target') - - def test_collect_can_return_gevent_timeout(self): - timeout = Timeout(1) - - def raise_timeout(): - raise timeout - - result = ParallelSFTPClient._collect( - [DeferredTask(raise_timeout)], stop_on_errors=False) - - self.assertEqual([timeout], result) - - def test_public_operations_forward_arguments(self): - client = Mock() - operations = [ - ('chdir', ('directory',), ('chdir', ('directory',))), - ('stat', ('file',), ('stat', ('file',))), - ('lstat', ('link',), ('lstat', ('link',))), - ('mkdir', ('directory',), ('mkdir', ('directory',))), - ('rmdir', ('directory',), ('rmdir', ('directory',))), - ('rename', ('old', 'new'), ('rename', ('old', 'new'))), - ('remove', ('file',), ('remove', ('file',))), - ('unlink', ('file',), ('remove', ('file',))), - ('get', ('remote', 'local'), ('get', ('remote', 'local'))), - ('put', ('local', 'remote'), ('put', ('local', 'remote'))), - ] - - for method, args, forwarded in operations: - with self.subTest(method=method): - getattr(ParallelSFTPClient, method)(client, *args) - client._run_parallel.assert_called_once_with( - forwarded[0], forwarded[1], stop_on_errors=True) - client._run_parallel.reset_mock() - - def test_changing_hosts_invalidates_sftp_clients(self): - client = object.__new__(ParallelSFTPClient) - client._hosts = ['old-host'] - client._host_clients = {(0, 'old-host'): Mock()} - client._sftp_clients = {(0, 'old-host'): Mock()} - - client.hosts = ['new-host'] - - self.assertEqual({}, client._host_clients) - self.assertEqual({}, client._sftp_clients) diff --git a/tests/test_native_sftp.py b/tests/test_native_sftp.py index c24750a7..44a82467 100644 --- a/tests/test_native_sftp.py +++ b/tests/test_native_sftp.py @@ -7,6 +7,7 @@ # License as published by the Free Software Foundation, version 2.1. import unittest +from types import GeneratorType from pssh.clients.native.sftp import SFTPClient @@ -67,7 +68,7 @@ class SSHClient(object): def __init__(self, sftp): self.sftp = sftp - def _make_sftp(self): + def make_sftp_client(self): return self.sftp def eagain(self, func, *args): @@ -135,9 +136,16 @@ def test_invalid_path_type_fails_before_transport(self): self.assertEqual(self.sftp.opendir_calls, []) - def test_listdir_filters_navigation_entries(self): - self.assertEqual(self.client.listdir('data'), ['file.txt', 'data']) + def test_listdir_filters_names_and_holds_handle_during_iteration(self): + names = self.client.listdir('data') + + self.assertIsInstance(names, GeneratorType) + self.assertEqual(self.sftp.opendir_calls, []) + self.assertEqual(next(names), 'file.txt') self.assertEqual(self.sftp.opendir_calls, ['/home/tester/data']) + self.assertFalse(self.sftp.handles[0].closed) + self.assertEqual(list(names), ['data']) + self.assertTrue(self.sftp.handles[0].closed) def test_metadata_and_mutations_resolve_remote_paths(self): self.assertEqual(self.client.stat('file'), 'stat-result') diff --git a/tests/test_native_single.py b/tests/test_native_single.py index a50e241c..4f18f50a 100644 --- a/tests/test_native_single.py +++ b/tests/test_native_single.py @@ -11,8 +11,10 @@ import unittest +from unittest.mock import Mock, patch from pssh.clients.native.single import SSHClient +from pssh.exceptions import SFTPError from pssh.output import HostOutput @@ -41,6 +43,54 @@ def get_exit_status(self): class NativeSingleClientTest(unittest.TestCase): + def test_make_sftp_client_returns_channel_and_wraps_errors(self): + client = object.__new__(SSHClient) + sftp = object() + client._make_sftp_eagain = lambda: sftp + + self.assertIs(client.make_sftp_client(), sftp) + + error = RuntimeError('sftp init failed') + + def raise_error(): + raise error + + client._make_sftp_eagain = raise_error + + with self.assertRaises(SFTPError) as raised: + client.make_sftp_client() + + self.assertIs(raised.exception.args[0], error) + + def test_transfer_helpers_create_sftp_client(self): + client = Mock(spec=SSHClient) + client.host = 'host' + sftp = Mock() + client.make_sftp_client.return_value = sftp + client._remote_paths_split.return_value = None + client._sftp_openfh.side_effect = SFTPError + client._scp_recv_recursive.return_value = 'received' + client._scp_send_dir.return_value = 'sent' + client.eagain.side_effect = lambda func, *args: func(*args) + + with patch('pssh.clients.native.single.os.path.isdir') as isdir: + isdir.return_value = False + SSHClient.copy_file(client, 'local', 'remote') + SSHClient.copy_remote_file(client, 'remote', 'local') + self.assertEqual( + SSHClient.scp_recv( + client, 'remote', 'local', recurse=True), 'received') + isdir.return_value = True + self.assertEqual( + SSHClient.scp_send( + client, 'local', 'remote', recurse=True), 'sent') + isdir.return_value = False + client._remote_paths_split.return_value = '/remote' + SSHClient.scp_send( + client, 'local', 'remote/file', recurse=True) + + self.assertEqual(client.make_sftp_client.call_count, 5) + def test_wait_finished_waits_for_close_before_exit_status(self): client = object.__new__(SSHClient) client.eagain = lambda func: func() From 7a0a629a388aee72ed9856c7d84c18c65161e620 Mon Sep 17 00:00:00 2001 From: lovewave02 Date: Tue, 1 Sep 2026 04:45:34 +0900 Subject: [PATCH 4/5] Fix native SCP receive timeout on Python 3.12 --- pssh/clients/native/single.py | 8 ++++++- tests/test_native_single.py | 43 +++++++++++++++++++++++++++++++++-- 2 files changed, 48 insertions(+), 3 deletions(-) diff --git a/pssh/clients/native/single.py b/pssh/clients/native/single.py index a5a55ac1..5b7078fb 100644 --- a/pssh/clients/native/single.py +++ b/pssh/clients/native/single.py @@ -697,10 +697,16 @@ def _scp_recv(self, remote_file, local_file): try: total = 0 while total < fileinfo.st_size: - size, data = file_chan.read(size=fileinfo.st_size - total) + size = min(self._BUF_SIZE, fileinfo.st_size - total) + size, data = file_chan.read(size=size) if size == LIBSSH2_ERROR_EAGAIN: self.poll() continue + if size == 0: + raise SCPError( + "Unexpected EOF while receiving %s (%s of %s bytes)" % ( + remote_file, total, fileinfo.st_size), + remote_file, self.host) total += size local_fh.write(data) finally: diff --git a/tests/test_native_single.py b/tests/test_native_single.py index 4f18f50a..5cb8d7f1 100644 --- a/tests/test_native_single.py +++ b/tests/test_native_single.py @@ -11,10 +11,10 @@ import unittest -from unittest.mock import Mock, patch +from unittest.mock import Mock, call, patch from pssh.clients.native.single import SSHClient -from pssh.exceptions import SFTPError +from pssh.exceptions import SCPError, SFTPError from pssh.output import HostOutput @@ -43,6 +43,45 @@ def get_exit_status(self): class NativeSingleClientTest(unittest.TestCase): + @patch('pssh.clients.native.single.FileObjectThread') + def test_scp_recv_rejects_unexpected_eof(self, file_object): + client = object.__new__(SSHClient) + client.host = '127.0.0.1' + client.session = Mock() + client.poll = Mock() + channel = Mock() + channel.read.return_value = (0, b'') + fileinfo = Mock(st_size=4) + client.session.scp_recv2.return_value = (channel, fileinfo) + + with self.assertRaises(SCPError): + client._scp_recv('remote', 'local') + + channel.read.assert_called_once_with(size=4) + client.poll.assert_not_called() + file_object.return_value.write.assert_not_called() + file_object.return_value.flush.assert_called_once_with() + file_object.return_value.close.assert_called_once_with() + channel.close.assert_called_once_with() + + @patch('pssh.clients.native.single.FileObjectThread') + def test_scp_recv_limits_read_size_to_buffer(self, file_object): + client = object.__new__(SSHClient) + client._BUF_SIZE = 3 + client.session = Mock() + client.poll = Mock() + channel = Mock() + channel.read.side_effect = [(3, b'one'), (1, b'!')] + fileinfo = Mock(st_size=4) + client.session.scp_recv2.return_value = (channel, fileinfo) + + client._scp_recv('remote', 'local') + + self.assertEqual(channel.read.call_args_list, + [call(size=3), call(size=1)]) + client.poll.assert_not_called() + file_object.return_value.write.assert_has_calls([call(b'one'), call(b'!')]) + def test_make_sftp_client_returns_channel_and_wraps_errors(self): client = object.__new__(SSHClient) sftp = object() From 21807f00f90c25e6bd0de97a881b34ed9ccb3095 Mon Sep 17 00:00:00 2001 From: lovewave02 Date: Tue, 1 Sep 2026 08:07:10 +0900 Subject: [PATCH 5/5] Fix native SCP receive timeout on Python 3.12 --- pssh/clients/native/single.py | 4 +++- tests/test_native_single.py | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/pssh/clients/native/single.py b/pssh/clients/native/single.py index 5b7078fb..92397f46 100644 --- a/pssh/clients/native/single.py +++ b/pssh/clients/native/single.py @@ -95,6 +95,8 @@ class SSHClient(BaseSSHClient): """ssh2-python (libssh2) based non-blocking SSH client.""" # 2MB buffer _BUF_SIZE = 2048 * 1024 + # Keep SCP receives in bounded chunks instead of requesting the whole file. + _SCP_RECV_BUF_SIZE = 64 * 1024 def __init__(self, host, user=None, password=None, port=None, @@ -697,7 +699,7 @@ def _scp_recv(self, remote_file, local_file): try: total = 0 while total < fileinfo.st_size: - size = min(self._BUF_SIZE, fileinfo.st_size - total) + size = min(self._SCP_RECV_BUF_SIZE, fileinfo.st_size - total) size, data = file_chan.read(size=size) if size == LIBSSH2_ERROR_EAGAIN: self.poll() diff --git a/tests/test_native_single.py b/tests/test_native_single.py index 5cb8d7f1..435a38f8 100644 --- a/tests/test_native_single.py +++ b/tests/test_native_single.py @@ -67,7 +67,7 @@ def test_scp_recv_rejects_unexpected_eof(self, file_object): @patch('pssh.clients.native.single.FileObjectThread') def test_scp_recv_limits_read_size_to_buffer(self, file_object): client = object.__new__(SSHClient) - client._BUF_SIZE = 3 + client._SCP_RECV_BUF_SIZE = 3 client.session = Mock() client.poll = Mock() channel = Mock()