Disc (#402)

* Fix socket shutdown on disconnect - resolves #274 
* Updated tests
* Updated clients, changelog
* Move disconnect to internal function
* Refactor ci tests. Updated disconnect on clients
* Refactor for keep alive
* Added example script
* Updated setup.cfg
* Updated tunnel socket failure tests. Added tunnel test.
* Added shutdown tests, updated changelog
* Updated tunnel

Author: pkittenis

Changelog.rst

@@ -1,5 +1,32 @@
 Change Log
-============
+==========
+
+2.14.0
+++++++
+
+Changes
+-------
+
+* Handle disconnects better to allow for file descriptor reuse for both clients.
+* Parallel clients no longer forcefully disconnect their clients at de-allocation -
+  now done by each individual ``SSHClient`` instead when that ``SSHClient`` goes out of scope.
+  This allows reading of output and anything associated with output, exit codes et al,
+  to work as long as one of either the client or an associated output object is alive.
+* ``SSHClient.disconnect`` is now a no-op and deprecated - handled by object de-allocation.
+* ``SSHClient.eagain`` is now a public function - wrapper for polling socket and calling a given socket using function.
+* ``SSHClient.eagain_write`` is now a public function - wrapper for polling socket and calling a given socket using
+  write function.
+* ``SSHClient``, ``TunnelServer`` and ``LocalForwarder`` now use their own gevent pools for greenlets spawned so they
+  are cleaned up correctly at shutdown.
+
+Fixes
+------
+
+* Forwarder threads used for proxies would not exit gracefully at interpreter shutdown, sometimes causing segfaults.
+* Client, both parallel and single, going out of scope would cause reading output from existing output objects
+  to break.
+* Explicitly calling ``SSHClient.disconnect`` would sometimes cause segfaults at interpreter shutdown.
+* Keepalives being configured on native client would keep client in scope forever.
 
 
 2.13.0
@@ -418,7 +445,7 @@ Fixes
 ------
 
 * Reading from output streams with timeout via `run_command(<..>, timeout=<timeout>)` would raise timeout early when
-trying to read from a stream with no data written to it while other streams have pending data - #180.
+  trying to read from a stream with no data written to it while other streams have pending data - #180.
 
 
 1.12.0

ci/integration_tests/native/__init__.py renamed to ci/integration_tests/libssh2_clients/__init__.py

@@ -1779,7 +1779,7 @@ def test_client_disconnect(self):
         client.join(consume_output=True)
         single_client = list(client._host_clients.values())[0]
         del client
-        self.assertEqual(single_client.session, None)
+        self.assertIsNotNone(single_client.session)
 
     def test_client_disconnect_error(self):
         def disc():
@@ -1788,7 +1788,7 @@ def disc():
                                    pkey=self.user_key, num_retries=1)
         output = client.run_command(self.cmd)
         client.join(output)
-        client._host_clients[(0, self.host)].disconnect = disc
+        client._host_clients[(0, self.host)]._disconnect = disc
         del client
 
     def test_multiple_join_timeout(self):

ci/integration_tests/native/base_ssh2_case.py renamed to ci/integration_tests/libssh2_clients/base_ssh2_case.py

@@ -16,23 +16,22 @@
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 
 import os
+import pytest
 import shutil
 import subprocess
 import tempfile
 from datetime import datetime
-from hashlib import sha256
-from tempfile import NamedTemporaryFile
-from unittest.mock import MagicMock, call, patch
-
-import pytest
 from gevent import sleep, spawn, Timeout as GTimeout, socket
+from hashlib import sha256
 from pytest import raises
 from ssh2.exceptions import (SocketDisconnectError, BannerRecvError, SocketRecvError,
                              AgentConnectionError, AgentListIdentitiesError,
                              AgentAuthenticationError, AgentGetIdentityError, SFTPProtocolError,
                              AuthenticationError as SSH2AuthenticationError,
                              )
 from ssh2.session import Session
+from tempfile import NamedTemporaryFile
+from unittest.mock import MagicMock, call, patch
 
 from pssh.clients.native import SSHClient
 from pssh.exceptions import (AuthenticationException, ConnectionErrorException,
@@ -272,16 +271,14 @@ class _SSHClient(SSHClient):
                            pkey=self.user_key,
                            num_retries=1,
                            allow_agent=False)
-        client.disconnect()
         client.pkey = None
-        del client.session
-        del client.sock
+        client._disconnect()
         client._connect(self.host, self.port)
         client._init_session()
         client.IDENTITIES = (self.user_key,)
         # Default identities auth only should succeed
         client._identity_auth()
-        client.disconnect()
+        client._disconnect()
         client._connect(self.host, self.port)
         client._init_session()
         # Auth should succeed
@@ -360,9 +357,37 @@ def test_handshake_fail(self):
         client = SSHClient(self.host, port=self.port,
                            pkey=self.user_key,
                            num_retries=1)
-        client.session.disconnect()
+        client.eagain(client.session.disconnect)
         self.assertRaises((SocketDisconnectError, BannerRecvError, SocketRecvError), client._init_session)
 
+    @patch('gevent.socket.socket')
+    @patch('pssh.clients.native.single.Session')
+    def test_sock_shutdown_fail(self, mock_sess, mock_sock):
+        sess = MagicMock()
+        sock = MagicMock()
+        mock_sess.return_value = sess
+        mock_sock.return_value = sock
+
+        hand_mock = MagicMock()
+        sess.handshake = hand_mock
+        retries = 2
+        client = SSHClient(self.host, port=self.port,
+                           num_retries=retries,
+                           timeout=.1,
+                           retry_delay=.1,
+                           _auth_thread_pool=False,
+                           allow_agent=False,
+                           )
+        self.assertIsInstance(client, SSHClient)
+        hand_mock.side_effect = AuthenticationError
+        sock.closed = False
+        sock.detach = MagicMock()
+        sock.detach.side_effect = Exception
+        self.assertRaises(AuthenticationError, client._init_session)
+        self.assertEqual(sock.detach.call_count, retries)
+        client._disconnect()
+        self.assertIsNone(client.sock)
+
     def test_stdout_parsing(self):
         dir_list = os.listdir(os.path.expanduser('~'))
         host_out = self.client.run_command('ls -la')
@@ -438,15 +463,30 @@ def test_multiple_clients_exec_terminates_channels(self):
         # and break subsequent sessions even on different socket and
         # session
         def scope_killer():
-            for _ in range(5):
+            for _ in range(20):
                 client = SSHClient(self.host, port=self.port,
                                    pkey=self.user_key,
                                    num_retries=1,
                                    allow_agent=False)
                 host_out = client.run_command(self.cmd)
                 output = list(host_out.stdout)
                 self.assertListEqual(output, [self.resp])
-                client.disconnect()
+
+        scope_killer()
+
+    def test_multiple_clients_exec_terminates_channels_explicit_disc(self):
+        # Explicit disconnects should not affect subsequent connections
+        def scope_killer():
+            for _ in range(20):
+                client = SSHClient(self.host, port=self.port,
+                                   pkey=self.user_key,
+                                   num_retries=1,
+                                   allow_agent=False)
+                host_out = client.run_command(self.cmd)
+                output = list(host_out.stdout)
+                self.assertListEqual(output, [self.resp])
+                client._disconnect()
+
         scope_killer()
 
     def test_agent_auth_exceptions(self):
@@ -1036,7 +1076,8 @@ def _make_sftp():
         client._make_sftp_eagain = _make_sftp
         self.assertRaises(SFTPError, client._make_sftp)
 
-    def test_disconnect_exc(self):
+    @patch('pssh.clients.native.single.Session')
+    def test_disconnect_exc(self, mock_sess):
         class DiscError(Exception):
             pass
 
@@ -1091,5 +1132,36 @@ def test_many_short_lived_commands(self):
             duration = end.total_seconds()
             self.assertTrue(duration < timeout * 0.9, msg=f"Duration of instant cmd is {duration}")
 
-    # TODO
-    # * read output callback
+    def test_output_client_scope(self):
+        """Output objects should keep client alive while they are in scope even if client is not."""
+        def make_client_run():
+            client = SSHClient(self.host, port=self.port,
+                               pkey=self.user_key,
+                               num_retries=1,
+                               allow_agent=False,
+                               )
+            host_out = client.run_command("%s; exit 1" % (self.cmd,))
+            return host_out
+
+        output = make_client_run()
+        stdout = list(output.stdout)
+        self.assertListEqual(stdout, [self.resp])
+        self.assertEqual(output.exit_code, 1)
+
+    def test_output_client_scope_disconnect(self):
+        """Calling deprecated .disconnect on client that also goes out of scope should not break reading
+        any unread output."""
+        def make_client_run():
+            client = SSHClient(self.host, port=self.port,
+                               pkey=self.user_key,
+                               num_retries=1,
+                               allow_agent=False,
+                               )
+            host_out = client.run_command("%s; exit 1" % (self.cmd,))
+            client.disconnect()
+            return host_out
+
+        output = make_client_run()
+        stdout = list(output.stdout)
+        self.assertListEqual(stdout, [self.resp])
+        self.assertEqual(output.exit_code, 1)