Added timeout exception raising on output reading time out. Added timeout file read test, updated tests

Author: Pan

Changelog.rst

@@ -1,6 +1,14 @@
 Change Log
 ============
 
+1.5.1
+++++++
+
+Fixes
+--------
+
+* Output ``pssh.exceptions.Timeout`` exception raising was not enabled.
+
 1.5.0
 ++++++
 

doc/advanced.rst

@@ -178,6 +178,38 @@ The native clients have timeout functionality on reading output and ``client.joi
 
 The client will raise a ``Timeout`` exception if remote commands have not finished within five seconds in the above examples.
 
+In some cases, such as when the remote command never terminates unless interrupted, it is necessary to use PTY and to close the channel to force the process to be terminated before a ``join`` sans timeout can complete. For example:
+
+.. code-block:: python
+
+   output = client.run_command('tail -f /var/log/messages', use_pty=True)
+   client.join(output, timeout=1)
+   # Closing channel which has PTY has the effect of terminating
+   # any running processes started on that channel.
+   for host in client.hosts:
+       client.host_clients[host].close_channel(output[host].channel)
+   client.join(output)
+
+Without a PTY, the ``join`` will complete but the remote process will be left running as per SSH protocol specifications.
+
+Output reading and Timeouts
+______________________________
+
+Furthermore, once reading output has timed out, it is necessary to restart the output generators as by Python design they only iterate once. This can be done as follows:
+
+.. code-block:: python
+
+   output = client.run_command(.., timeout=1)
+   for host, host_out in output.items():
+       try:
+           stdout = list(host_out.stdout)
+       except Timeout:
+           stdout_buf = client.host_clients[host].read_output_buffer(
+               client.host_clients[host].read_output(
+	           output[host].channel, timeout=1))
+	   # Reset generator to be able to gather new output
+           host_out.stdout = stdout_buf
+
 .. note::
 
    ``join`` with a timeout forces output to be consumed as otherwise the pending output will keep the channel open and make it appear as if command has not yet finished.

pssh/native/_ssh2.c

@@ -33,7 +33,7 @@ from ssh2.sftp_handle cimport SFTPHandle
 from ssh2.exceptions import SFTPIOError
 from ssh2.utils cimport to_bytes
 
-from ..exceptions import SessionError
+from ..exceptions import SessionError, Timeout
 
 
 cdef bytes LINESEP = b'\n'
@@ -54,7 +54,7 @@ def _read_output(Session session, read_func, timeout=None):
             _wait_select(_sock, _session, timeout)
             _size, _data = read_func()
             if timeout is not None and _size == LIBSSH2_ERROR_EAGAIN:
-                break
+                raise Timeout
         while _size > 0:
             while _pos < _size:
                 linesep = _data[:_size].find(LINESEP, _pos)

pssh/native/_ssh2.pyx

@@ -1193,14 +1193,43 @@ def test_join_timeout_set_no_timeout(self):
     def test_read_timeout(self):
         client = ParallelSSHClient([self.host], port=self.port,
                                    pkey=self.user_key)
-        output = client.run_command('sleep 2', timeout=1)
-        stdout = list(output[self.host].stdout)
+        output = client.run_command('sleep 2; echo me', timeout=1)
+        for host, host_out in output.items():
+            self.assertRaises(Timeout, list, host_out.stdout)
         self.assertFalse(output[self.host].channel.eof())
-        self.assertEqual(len(stdout), 0)
-        list(output[self.host].stdout)
-        list(output[self.host].stdout)
         client.join(output)
-        self.assertTrue(output[self.host].channel.eof())
+        for host, host_out in output.items():
+            stdout_buf = client.host_clients[self.host].read_output_buffer(
+                client.host_clients[self.host].read_output(
+                    output[self.host].channel, timeout=1))
+            host_out.stdout = stdout_buf
+            stdout = list(output[self.host].stdout)
+            self.assertEqual(len(stdout), 1)
+
+    def test_timeout_file_read(self):
+        dir_name = os.path.dirname(__file__)
+        _file = os.sep.join((dir_name, 'file_to_read'))
+        contents = [b'a line\n' for _ in range(50)]
+        with open(_file, 'wb') as fh:
+            fh.writelines(contents)
+        try:
+            output = self.client.run_command('tail -f %s' % (_file,),
+                                             use_pty=True,
+                                             timeout=1)
+            self.assertRaises(Timeout, self.client.join, output, timeout=1)
+            for host, host_out in output.items():
+                try:
+                    for line in host_out.stdout:
+                        pass
+                except Timeout:
+                    pass
+                else:
+                    raise Exception("Timeout should have been raised")
+            channel = output[self.host].channel
+            self.client.host_clients[self.host].close_channel(channel)
+            self.client.join(output)
+        finally:
+            os.unlink(_file)
 
     ## OpenSSHServer needs to run in its own thread for this test to work
     ##  Race conditions otherwise.