Updated documentation. Added parallel commands example script. Added test for overriding hosts list and using iterator as hosts list.

Author: Dan

examples/parallel_commands.py

@@ -0,0 +1,22 @@
+from pssh import ParallelSSHClient
+import datetime
+
+output = []
+host = 'localhost'
+hosts = [host]
+client = ParallelSSHClient(hosts)
+
+# Run 10 five second sleeps
+cmds = ['sleep 5' for _ in xrange(10)]
+start = datetime.datetime.now()
+for cmd in cmds:
+    output.append(client.run_command(cmd, stop_on_errors=False))
+end = datetime.datetime.now()
+print "Started %s commands on %s host(s) in %s" % (
+    len(cmds), len(hosts), end-start,)
+start = datetime.datetime.now()
+for _output in output:
+    for line in _output[host]['stdout']:
+        print line
+end = datetime.datetime.now()
+print "All commands finished in %s" % (end-start,)

pssh/pssh_client.py

@@ -177,21 +177,11 @@ def __init__(self, hosts,
         >>> import paramiko
         >>> client_key = paramiko.RSAKey.from_private_key_file('user.key')
         >>> client = ParallelSSHClient(['myhost1', 'myhost2'], pkey=client_key)
-
-        **Example with expression as host list**
         
-        Any type of iterator may be used as host list, including generator and
-        list comprehension expressions.
+        **Multiple commands**
         
-        >>> hosts = ['dc1.myhost1', 'dc2.myhost2']
-        >>> client = ParallelSSHClient([h for h in hosts if h.find('dc1')])
-        >>> client.run_command(<..>)
-
-        **Overriding host list**
-
-        >>> client.hosts = ['otherhost']
-        >>> print client.run_command('exit 0')
-        >>> {'otherhost': {'exit_code':0}, <..>}
+        >>> for cmd in ['uname', 'whoami']:
+        ...    client.run_command(cmd)
         
         .. note ::
         
@@ -225,7 +215,7 @@ def __init__(self, hosts,
         self.timeout = timeout
         self.proxy_host, self.proxy_port = proxy_host, proxy_port
         # To hold host clients
-        self.host_clients = dict((host, None) for host in hosts)
+        self.host_clients = {}
         self.agent = agent
 
     def run_command(self, *args, **kwargs):
@@ -259,9 +249,9 @@ def run_command(self, *args, **kwargs):
         :raises: :mod:`pssh.exceptions.UnknownHostException` on DNS resolution error
         :raises: :mod:`pssh.exceptions.ConnectionErrorException` on error connecting
         :raises: :mod:`pssh.exceptions.SSHException` on other undefined SSH errors
-
+        
         **Example Usage**
-
+        
         **Simple run command**
         
         >>> output = client.run_command('ls -ltrh')
@@ -291,41 +281,79 @@ def run_command(self, *args, **kwargs):
         
         Capture stdout - **WARNING** - this will store the entirety of stdout
         into memory and may exhaust available memory if command output is
-        large enough:
+        large enough.
+        
+        Iterating over stdout/stderr by definition implies blocking until
+        command has finished. To only see output as it comes in without blocking
+        the host logger can be enabled - see `Enabling Host Logger` above.
         
         >>> for host in output:
         >>>     stdout = list(output[host]['stdout'])
         >>>     print "Complete stdout for host %s is %s" % (host, stdout,)
-
+        
+        **Expression as host list**
+        
+        Any type of iterator may be used as host list, including generator and
+        list comprehension expressions.
+        
+        >>> hosts = ['dc1.myhost1', 'dc2.myhost2']
+        # List comprehension
+        >>> client = ParallelSSHClient([h for h in hosts if h.find('dc1')])
+        # Generator
+        >>> client = ParallelSSHClient((h for h in hosts if h.find('dc1')))
+        # Filter
+        >>> client = ParallelSSHClient(filter(lambda h: h.find('dc1'), hosts))
+        >>> client.run_command(<..>)
+        
+        .. note ::
+        
+          Since iterators by design only iterate over a sequence once then stop,
+          `client.hosts` should be re-assigned after each call to `run_command`
+          when using iterators as target of `client.hosts`.
+        
+        **Overriding host list**
+        
+        Host list can be modified in place. Call to `run_command` will create
+        new connections as necessary and output will only contain output for
+        hosts command ran on.
+        
+        >>> client.hosts = ['otherhost']
+        >>> print client.run_command('exit 0')
+        >>> {'otherhost': {'exit_code':0}, <..>}
+        
         **Run multiple commands in parallel**
 
-        This short example demonstrates running long running commands in parallel
-        and how long it takes for all commands to start, blocking until they
-        complete and how long it takes for all commands to complete.
-        
-        See examples directory for complete example script. ::
+        This short example demonstrates running long running commands in
+        parallel, how long it takes for all commands to start, blocking until
+        they complete and how long it takes for all commands to complete.
         
-        output = []
+        See examples directory for complete script. ::
         
-        start = datetime.datetime.now()
-        cmds = ['sleep 5' for _ in xrange(10)]
-        for cmd in cmds:
-            output.append(client.run_command(cmd, stop_on_errors=False))
-        end = datetime.datetime.now()
-        print "Started %s commands in %s" % (len(cmds), end-start,)
-        start = datetime.datetime.now()
-        for _output in output:
-            for line in _output[host]['stdout']:
-                print line
-        end = datetime.datetime.now()
-        print "All commands finished in %s" % (end-start,)
+          output = []
+          host = 'localhost'
+          
+          # Run 10 five second sleeps
+          cmds = ['sleep 5' for _ in xrange(10)]
+          start = datetime.datetime.now()
+          for cmd in cmds:
+              output.append(client.run_command(cmd, stop_on_errors=False))
+          end = datetime.datetime.now()
+          print "Started %s commands in %s" % (len(cmds), end-start,)
+          start = datetime.datetime.now()
+          for _output in output:
+              for line in _output[host]['stdout']:
+                  print line
+          end = datetime.datetime.now()
+          print "All commands finished in %s" % (end-start,)
         
         *Output*
         
-        Started 10 commands in 0:00:00.428629
-        All commands finished in 0:00:05.014757
+        ::
         
-        **Example Output**
+          Started 10 commands in 0:00:00.428629
+          All commands finished in 0:00:05.014757
+        
+        **Output dictionary**
         
         ::
         
@@ -575,7 +603,7 @@ def copy_file(self, local_file, remote_file, recurse=False):
 
     def _copy_file(self, host, local_file, remote_file, recurse=False):
         """Make sftp client, copy file"""
-        if not self.host_clients[host]:
+        if not host in self.host_clients or not self.host_clients[host]:
             self.host_clients[host] = SSHClient(
                 host, user=self.user, password=self.password,
                 port=self.port, pkey=self.pkey,

tests/test_pssh_client.py

@@ -419,14 +419,51 @@ def test_pssh_hosts_more_than_pool_size(self):
         stdout = [list(output[k]['stdout']) for k in output]
         expected_stdout = [[self.fake_resp], [self.fake_resp]]
         self.assertEqual(len(hosts), len(output),
-                         msg="Did not get output from all hosts. Got output for \
-%s/%s hosts" % (len(output), len(hosts),))
+                         msg="Did not get output from all hosts. Got output for " \
+                         "%s/%s hosts" % (len(output), len(hosts),))
         self.assertEqual(expected_stdout, stdout,
                          msg="Did not get expected output from all hosts. \
                          Got %s - expected %s" % (stdout, expected_stdout,))
         del client
         del server2
-
+    
+    def test_pssh_hosts_iterator_hosts_modification(self):
+        """Test using iterator as host list and modifying host list in place"""
+        server2_socket = make_socket('127.0.0.2', port=self.listen_port)
+        server2_port = server2_socket.getsockname()[1]
+        server2 = start_server(server2_socket)
+        server3_socket = make_socket('127.0.0.3', port=self.listen_port)
+        server3_port = server3_socket.getsockname()[1]
+        server3 = start_server(server3_socket)
+        hosts = [self.host, '127.0.0.2']
+        client = ParallelSSHClient(iter(hosts),
+                                   port=self.listen_port,
+                                   pkey=self.user_key,
+                                   pool_size=1,
+                                   )
+        output = client.run_command(self.fake_cmd)
+        stdout = [list(output[k]['stdout']) for k in output]
+        expected_stdout = [[self.fake_resp], [self.fake_resp]]
+        self.assertEqual(len(hosts), len(output),
+                         msg="Did not get output from all hosts. Got output for " \
+                         "%s/%s hosts" % (len(output), len(hosts),))
+        # Run again without re-assigning host list, should do nothing
+        output = client.run_command(self.fake_cmd)
+        self.assertFalse(hosts[0] in output,
+                         msg="Expected no host output, got %s" % (output,))
+        self.assertFalse(output,
+                         msg="Expected empty output, got %s" % (output,))
+        # Re-assigning host list with new hosts should work
+        hosts = ['127.0.0.2', '127.0.0.3']
+        client.hosts = iter(hosts)
+        output = client.run_command(self.fake_cmd)
+        self.assertEqual(len(hosts), len(output),
+                         msg="Did not get output from all hosts. Got output for " \
+                         "%s/%s hosts" % (len(output), len(hosts),))
+        self.assertTrue(hosts[1] in output,
+                        msg="Did not get output for new host %s" % (hosts[1],))
+        del client, server2, server3
+    
     def test_ssh_proxy(self):
         """Test connecting to remote destination via SSH proxy
         client -> proxy -> destination