Sftp tests (#273)

* Added sftp tests for errors opening local and remote files and directories and sftp init.
* Added disconnect exception test.
* Test cleanup
* Moved polling code to base client
* Added libssh client tests
* Updated tests for faster test running (60-70% speedup)
* Updated setup.cfg

Author: pkittenis

pssh/clients/base/single.py

@@ -27,7 +27,8 @@
 
 from gevent import sleep, socket, Timeout as GTimeout
 from gevent.hub import Hub
-from gevent.select import poll
+from gevent.select import poll, POLLIN, POLLOUT
+
 from ssh2.utils import find_eol
 from ssh2.exceptions import AgentConnectionError, AgentListIdentitiesError, \
     AgentAuthenticationError, AgentGetIdentityError
@@ -244,9 +245,12 @@ def open_shell(self, encoding='utf-8', read_timeout=None):
     def _shell(self, channel):
         raise NotImplementedError
 
+    def _disconnect_eagain(self):
+        self._eagain(self.session.disconnect)
+
     def _connect_init_session_retry(self, retries):
         try:
-            self.session.disconnect()
+            self._disconnect_eagain()
         except Exception:
             pass
         self.session = None
@@ -664,3 +668,15 @@ def _poll_socket(self, events, timeout=None):
         poller = poll()
         poller.register(self.sock, eventmask=events)
         poller.poll(timeout=timeout)
+
+    def _poll_errcodes(self, directions_func, inbound, outbound, timeout=None):
+        timeout = self.timeout if timeout is None else timeout
+        directions = directions_func()
+        if directions == 0:
+            return
+        events = 0
+        if directions & inbound:
+            events = POLLIN
+        if directions & outbound:
+            events |= POLLOUT
+        self._poll_socket(events, timeout=timeout)

pssh/clients/native/single.py

@@ -21,7 +21,6 @@
 from warnings import warn
 
 from gevent import sleep, spawn, get_hub
-from gevent.select import POLLIN, POLLOUT
 from ssh2.error_codes import LIBSSH2_ERROR_EAGAIN
 from ssh2.exceptions import SFTPHandleError, SFTPProtocolError, \
     Timeout as SSH2Timeout
@@ -163,11 +162,14 @@ def _connect_proxy(self, proxy_host, proxy_port, proxy_pkey,
         return proxy_local_port
 
     def disconnect(self):
-        """Disconnect session, close socket if needed."""
+        """Attempt to disconnect session.
+
+        Any errors on calling disconnect are suppressed by this function.
+        """
         self._keepalive_greenlet = None
         if self.session is not None:
             try:
-                self._eagain(self.session.disconnect)
+                self._disconnect_eagain()
             except Exception:
                 pass
             self.session = None
@@ -316,10 +318,13 @@ def close_channel(self, channel):
     def _eagain(self, func, *args, **kwargs):
         return self._eagain_errcode(func, LIBSSH2_ERROR_EAGAIN, *args, **kwargs)
 
+    def _make_sftp_eagain(self):
+        return self._eagain(self.session.sftp_init)
+
     def _make_sftp(self):
         """Make SFTP client from open transport"""
         try:
-            sftp = self._eagain(self.session.sftp_init)
+            sftp = self._make_sftp_eagain()
         except Exception as ex:
             raise SFTPError(ex)
         return sftp
@@ -486,6 +491,27 @@ def copy_remote_file(self, remote_file, local_file, recurse=False,
         logger.info("Copied local file %s from remote destination %s:%s",
                     local_file, self.host, remote_file)
 
+    def _scp_recv_recursive(self, remote_file, local_file, sftp, encoding='utf-8'):
+        try:
+            self._eagain(sftp.stat, remote_file)
+        except (SFTPHandleError, SFTPProtocolError):
+            msg = "Remote file or directory %s does not exist"
+            logger.error(msg, remote_file)
+            raise SCPError(msg, remote_file)
+        try:
+            dir_h = self._sftp_openfh(sftp.opendir, remote_file)
+        except SFTPError:
+            # remote_file is not a dir, scp file
+            return self.scp_recv(remote_file, local_file, encoding=encoding)
+        try:
+            os.makedirs(local_file)
+        except OSError:
+            pass
+        file_list = self._sftp_readdir(dir_h)
+        return self._scp_recv_dir(file_list, remote_file,
+                                  local_file, sftp,
+                                  encoding=encoding)
+
     def scp_recv(self, remote_file, local_file, recurse=False, sftp=None,
                  encoding='utf-8'):
         """Copy remote file to local host via SCP.
@@ -505,33 +531,13 @@ def scp_recv(self, remote_file, local_file, recurse=False, sftp=None,
           enabled.
         :type encoding: str
 
-        :raises: :py:class:`pssh.exceptions.SCPError` when a directory is
-          supplied to ``local_file`` and ``recurse`` is not set.
         :raises: :py:class:`pssh.exceptions.SCPError` on errors copying file.
         :raises: :py:class:`IOError` on local file IO errors.
         :raises: :py:class:`OSError` on local OS errors like permission denied.
         """
         if recurse:
             sftp = self._make_sftp() if sftp is None else sftp
-            try:
-                self._eagain(sftp.stat, remote_file)
-            except (SFTPHandleError, SFTPProtocolError):
-                msg = "Remote file or directory %s does not exist"
-                logger.error(msg, remote_file)
-                raise SCPError(msg, remote_file)
-            try:
-                dir_h = self._sftp_openfh(sftp.opendir, remote_file)
-            except SFTPError:
-                pass
-            else:
-                try:
-                    os.makedirs(local_file)
-                except OSError:
-                    pass
-                file_list = self._sftp_readdir(dir_h)
-                return self._scp_recv_dir(file_list, remote_file,
-                                          local_file, sftp,
-                                          encoding=encoding)
+            return self._scp_recv_recursive(remote_file, local_file, sftp, encoding=encoding)
         elif local_file.endswith('/'):
             remote_filename = remote_file.rsplit('/')[-1]
             local_file += remote_filename
@@ -561,11 +567,6 @@ def _scp_recv(self, remote_file, local_file):
                     continue
                 total += size
                 local_fh.write(data)
-            if total != fileinfo.st_size:
-                msg = "Error copying data from remote file %s on host %s. " \
-                      "Copied %s out of %s total bytes"
-                raise SCPError(msg, remote_file, self.host, total,
-                               fileinfo.st_size)
         finally:
             local_fh.close()
             file_chan.close()
@@ -690,16 +691,12 @@ def poll(self, timeout=None):
         Blocks current greenlet only if socket has pending read or write operations
         in the appropriate direction.
         """
-        timeout = self.timeout if timeout is None else timeout
-        directions = self.session.block_directions()
-        if directions == 0:
-            return
-        events = 0
-        if directions & LIBSSH2_SESSION_BLOCK_INBOUND:
-            events = POLLIN
-        if directions & LIBSSH2_SESSION_BLOCK_OUTBOUND:
-            events |= POLLOUT
-        self._poll_socket(events, timeout=timeout)
+        self._poll_errcodes(
+            self.session.block_directions,
+            LIBSSH2_SESSION_BLOCK_INBOUND,
+            LIBSSH2_SESSION_BLOCK_OUTBOUND,
+            timeout=timeout,
+        )
 
     def _eagain_write(self, write_func, data, timeout=None):
         """Write data with given write_func for an ssh2-python session while

pssh/clients/ssh/single.py

@@ -18,7 +18,6 @@
 import logging
 
 from gevent import sleep, spawn, Timeout as GTimeout, joinall
-from gevent.select import POLLIN, POLLOUT
 from ssh import options
 from ssh.session import Session, SSH_READ_PENDING, SSH_WRITE_PENDING
 from ssh.key import import_privkey_file, import_cert_file, copy_cert_to_privkey
@@ -145,7 +144,7 @@ def _init_session(self, retries=1):
         self.session.set_socket(self.sock)
         logger.debug("Session started, connecting with existing socket")
         try:
-            self.session.connect()
+            self._session_connect()
         except Exception as ex:
             if retries < self.num_retries:
                 return self._connect_init_session_retry(retries=retries+1)
@@ -155,6 +154,9 @@ def _init_session(self, retries=1):
             ex.port = self.port
             raise ex
 
+    def _session_connect(self):
+        self.session.connect()
+
     def auth(self):
         if self.gssapi_auth or (self.gssapi_server_identity or self.gssapi_client_identity):
             try:
@@ -166,8 +168,6 @@ def auth(self):
         return super(SSHClient, self).auth()
 
     def _password_auth(self):
-        if not self.password:
-            raise AuthenticationError("All authentication methods failed")
         try:
             self.session.userauth_password(self.password)
         except Exception as ex:
@@ -270,7 +270,6 @@ def wait_finished(self, host_output, timeout=None):
         with GTimeout(seconds=timeout, exception=Timeout):
             joinall((host_output.buffers.stdout.reader, host_output.buffers.stderr.reader))
         logger.debug("Readers finished, closing channel")
-        # Close channel
         self.close_channel(channel)
 
     def finished(self, channel):
@@ -305,16 +304,12 @@ def close_channel(self, channel):
 
     def poll(self, timeout=None):
         """ssh-python based co-operative gevent poll on session socket."""
-        timeout = self.timeout if timeout is None else timeout
-        directions = self.session.get_poll_flags()
-        if directions == 0:
-            return
-        events = 0
-        if directions & SSH_READ_PENDING:
-            events = POLLIN
-        if directions & SSH_WRITE_PENDING:
-            events |= POLLOUT
-        self._poll_socket(events, timeout=timeout)
+        self._poll_errcodes(
+            self.session.get_poll_flags,
+            SSH_READ_PENDING,
+            SSH_WRITE_PENDING,
+            timeout=timeout,
+        )
 
     def _eagain(self, func, *args, **kwargs):
         """Run function given and handle EAGAIN for an ssh-python session"""

setup.cfg

@@ -9,3 +9,8 @@ universal = 1
 
 [flake8]
 max-line-length = 100
+
+[tool:pytest]
+# addopts=--cov=pssh --cov-append --cov-report=term --cov-report=term-missing
+testpaths =
+    tests

tests/native/base_ssh2_case.py

@@ -60,6 +60,7 @@ def setUpClass(cls):
                                pkey=PKEY_FILENAME,
                                num_retries=1,
                                identity_auth=False,
+                               retry_delay=.1,
                                )
 
     @classmethod