Host attr (#256)

* Migrate hosts to property, add removal of clients on set.
* Added iterator support for host list reassignment.
* Added tests.
* Updated documentation.
* Updated changelog.
* Removed unused params.

Author: pkittenis

Changelog.rst

@@ -10,6 +10,7 @@ Changes
 * Added interactive shell support to single and parallel clients - see `documentation <https://parallel-ssh.readthedocs.io/en/latest/advanced.html#interactive-shells>`_.
 * Added ``pssh.utils.enable_debug_logger`` function.
 * ``ParallelSSHClient`` timeout parameter is now also applied to *starting* remote commands via ``run_command``.
+* Assigning to ``ParallelSSHClient.hosts`` cleans up clients of hosts no longer in host list - #220
 
 Fixes
 -----

doc/advanced.rst

@@ -788,18 +788,21 @@ Iterators and filtering
 Any type of iterator may be used as hosts list, including generator and list comprehension expressions.
 
 :List comprehension:
+
    .. code-block:: python
 
       hosts = ['dc1.myhost1', 'dc2.myhost2']
       client = ParallelSSHClient([h for h in hosts if h.find('dc1')])
 
 :Generator:
+
    .. code-block:: python
 
       hosts = ['dc1.myhost1', 'dc2.myhost2']
       client = ParallelSSHClient((h for h in hosts if h.find('dc1')))
 
 :Filter:
+
    .. code-block:: python
 
       hosts = ['dc1.myhost1', 'dc2.myhost2']
@@ -808,12 +811,20 @@ Any type of iterator may be used as hosts list, including generator and list com
 
 .. note ::
 
-    Since generators by design only iterate over a sequence once then stop, ``client.hosts`` should be re-assigned after each call to ``run_command`` when using generators as target of ``client.hosts``.
+    Assigning a generator to host list is possible as shown above, and the generator is consumed into a list on assignment.
+
+    Multiple calls to ``run_command`` will use the same hosts read from the provided generator.
+
 
 Overriding hosts list
 =======================
 
-Hosts list can be modified in place. A call to ``run_command`` will create new connections as necessary and output will only contain output for the hosts ``run_command`` executed on.
+Hosts list can be modified in place.
+
+A call to ``run_command`` will create new connections as necessary and output will only be returned for the hosts ``run_command`` executed on.
+
+Clients for hosts that are no longer on the host list are removed on host list assignment. Reading output from hosts removed from host list is feasible, as long as their output objects or interactive shells are in scope.
+
 
 .. code-block:: python
 
@@ -825,3 +836,31 @@ Hosts list can be modified in place. A call to ``run_command`` will create new c
        host='otherhost'
        exit_code=None
        <..>
+
+
+When wanting to reassign host list frequently, it is best to sort or otherwise ensure order is maintained to avoid reconnections on hosts that are still in the host list but in a different order.
+
+For example, the following will cause reconnections on both hosts, though both are still in the list.
+
+.. code-block:: python
+
+   client.hosts = ['host1', 'host2']
+   client.hosts = ['host2', 'host1']
+
+
+In such cases it would be best to maintain order to avoid reconnections. This is also true when adding or removing hosts in host list.
+
+No change in clients occurs in the following case.
+
+.. code-block:: python
+
+   client.hosts = sorted(['host1', 'host2'])
+   client.hosts = sorted(['host2', 'host1'])
+
+
+Clients for hosts that would be removed by a reassignment can be calculated with:
+
+.. code-block:: python
+
+   set(enumerate(client.hosts)).difference(
+       set(enumerate(new_hosts)))

pssh/clients/base/parallel.py

@@ -43,14 +43,10 @@ def __init__(self, hosts, user=None, password=None, port=None, pkey=None,
                  timeout=120, pool_size=10,
                  host_config=None, retry_delay=RETRY_DELAY,
                  identity_auth=True):
-        if isinstance(hosts, str) or isinstance(hosts, bytes):
-            raise TypeError(
-                "Hosts must be list or other iterable, not string. "
-                "For example: ['localhost'] not 'localhost'.")
         self.allow_agent = allow_agent
         self.pool_size = pool_size
         self.pool = gevent.pool.Pool(size=self.pool_size)
-        self.hosts = hosts
+        self._hosts = self._validate_hosts(hosts)
         self.user = user
         self.password = password
         self.port = port
@@ -64,6 +60,31 @@ def __init__(self, hosts, user=None, password=None, port=None, pkey=None,
         self.identity_auth = identity_auth
         self._check_host_config()
 
+    def _validate_hosts(self, _hosts):
+        if _hosts is None:
+            raise ValueError
+        elif isinstance(_hosts, str) or isinstance(_hosts, bytes):
+            raise TypeError(
+                "Hosts must be list or other iterable, not string. "
+                "For example: ['localhost'] not 'localhost'.")
+        elif hasattr(_hosts, '__next__') or hasattr(_hosts, 'next'):
+            _hosts = list(_hosts)
+        return _hosts
+
+    @property
+    def hosts(self):
+        return self._hosts
+
+    @hosts.setter
+    def hosts(self, _hosts):
+        _hosts = self._validate_hosts(_hosts)
+        cur_vals = set(enumerate(self._hosts))
+        new_vals = set(enumerate(_hosts))
+        to_remove = cur_vals.difference(new_vals)
+        for i, host in to_remove:
+            self._host_clients.pop((i, host), None)
+        self._hosts = _hosts
+
     def _check_host_config(self):
         if self.host_config is None:
             return

pssh/clients/base/single.py

@@ -154,10 +154,6 @@ def __init__(self, host,
         self.pkey = _validate_pkey_path(pkey, self.host)
         self.identity_auth = identity_auth
         self._keepalive_greenlet = None
-        self._stdout_buffer = None
-        self._stderr_buffer = None
-        self._stdout_reader = None
-        self._stderr_reader = None
         self._init()
 
     def _init(self):

tests/native/test_parallel_client.py

@@ -836,10 +836,10 @@ def test_pssh_hosts_iterator_hosts_modification(self):
         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
+        # Run again without re-assigning host list, should run the same
         output = client.run_command(self.cmd)
-        self.assertEqual(len(output), 0)
-        # Re-assigning host list with new hosts should work
+        self.assertEqual(len(output), len(hosts))
+        # Re-assigning host list with new hosts should also work
         hosts = ['127.0.0.2', '127.0.0.3']
         client.hosts = iter(hosts)
         output = client.run_command(self.cmd)
@@ -1044,8 +1044,7 @@ def test_host_config_bad_entries(self):
         hosts = ['localhost', 'localhost']
         host_config = [HostConfig()]
         self.assertRaises(ValueError, ParallelSSHClient, hosts, host_config=host_config)
-        # Can't sanity check generators
-        ParallelSSHClient(iter(hosts), host_config=host_config)
+        self.assertRaises(ValueError, ParallelSSHClient, iter(hosts), host_config=host_config)
 
     def test_pssh_client_override_allow_agent_authentication(self):
         """Test running command with allow_agent set to False"""
@@ -1262,6 +1261,67 @@ def test_retries(self):
         client.hosts = [host]
         self.assertRaises(UnknownHostException, client.run_command, self.cmd)
 
+    def test_setting_hosts(self):
+        host2 = '127.0.0.3'
+        server2 = OpenSSHServer(host2, port=self.port)
+        server2.start_server()
+        client = ParallelSSHClient(
+            [self.host], port=self.port,
+            num_retries=1, retry_delay=1,
+            pkey=self.user_key,
+        )
+        joinall(client.connect_auth())
+        _client = list(client._host_clients.values())[0]
+        client.hosts = [self.host]
+        joinall(client.connect_auth())
+        try:
+            self.assertEqual(len(client._host_clients), 1)
+            _client_after = list(client._host_clients.values())[0]
+            self.assertEqual(id(_client), id(_client_after))
+            client.hosts = ['127.0.0.2', self.host, self.host]
+            self.assertEqual(len(client._host_clients), 0)
+            joinall(client.connect_auth())
+            self.assertEqual(len(client._host_clients), 2)
+            client.hosts = ['127.0.0.2', self.host, self.host]
+            self.assertListEqual([(1, self.host), (2, self.host)],
+                                 sorted(list(client._host_clients.keys())))
+            self.assertEqual(len(client._host_clients), 2)
+            hosts = [self.host, self.host, host2]
+            client.hosts = hosts
+            joinall(client.connect_auth())
+            self.assertListEqual([(0, self.host), (1, self.host), (2, host2)],
+                                 sorted(list(client._host_clients.keys())))
+            self.assertEqual(len(client._host_clients), 3)
+            hosts = [host2, self.host, self.host]
+            client.hosts = hosts
+            joinall(client.connect_auth())
+            self.assertListEqual([(0, host2), (1, self.host), (2, self.host)],
+                                 sorted(list(client._host_clients.keys())))
+            self.assertEqual(len(client._host_clients), 3)
+            client.hosts = [self.host]
+            self.assertEqual(len(client._host_clients), 0)
+            joinall(client.connect_auth())
+            self.assertEqual(len(client._host_clients), 1)
+            client.hosts = [self.host, host2]
+            joinall(client.connect_auth())
+            self.assertListEqual([(0, self.host), (1, host2)],
+                                 sorted(list(client._host_clients.keys())))
+            self.assertEqual(len(client._host_clients), 2)
+            try:
+                client.hosts = None
+            except ValueError:
+                pass
+            else:
+                raise AssertionError
+            try:
+                client.hosts = ''
+            except TypeError:
+                pass
+            else:
+                raise AssertionError
+        finally:
+            server2.stop()
+
     def test_unknown_host_failure(self):
         """Test connection error failure case - ConnectionErrorException"""
         host = ''.join([random.choice(string.ascii_letters) for n in range(8)])