Added device connectivity checking to get_device_type (#34)

* Added device connectivity checking to get_device_type

* Changed pxssh "unable to determine device type" i integer.

* Fixed/added unit tests.

pylib/aeon/utils/__init__.py

@@ -3,6 +3,9 @@
 # This source code is licensed under End User License Agreement found in the
 # LICENSE file at http://www.apstra.com/community/eula
 
+import socket
+
+import paramiko
 import pexpect
 from pexpect import pxssh
 
@@ -30,14 +33,26 @@ def get_device(target=None, user='admin', passwd='admin', nos_only=False):
         'ubuntu': aeon.ubuntu.device.Device,
         'centos': aeon.centos.device.Device
     }
+    try:
+        # Use paramiko to check if device is reachable because pxssh is bad at that.
+        ssh = paramiko.SSHClient()
+        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        ssh.connect(target, username=user, password=passwd, timeout=5)
+        ssh.close()
+    except socket.error:
+        raise TargetError('Device unreachable: %s' % target)
+    except paramiko.AuthenticationException:
+        raise TargetError('Authentication error: %s' % target)
+    except paramiko.SSHException:
+        raise TargetError('Error logging in: %s' % target)
+
     try:
         nos = ''
         session = pxssh.pxssh(timeout=10, options={'StrictHostKeyChecking': "no",
                                        'UserKnownHostsFile': "/dev/null"})
         session.force_password = True
         session.PROMPT = '\r\n.*#|\r\n.*$'
         session.login(target, user, password=passwd, auto_prompt_reset=False, login_timeout=10)
-
         session.sendline('show version')
         i = session.expect(['Cisco', 'Arista', 'command not found', 'not installed', pexpect.TIMEOUT], timeout=10)
         if i == 0:
@@ -54,8 +69,8 @@ def get_device(target=None, user='admin', passwd='admin', nos_only=False):
                 nos = 'ubuntu'
             if i == 2:
                 nos = 'centos'
-            if i == 3:
-                raise TargetError('Unable to determine device type for %s' % target)
+        if i == 4:
+            raise TargetError('Unable to determine device type for %s' % target)
 
     except pxssh.ExceptionPxssh as e:
         raise TargetError("Error logging in to {target} : {error}".format(target=target, error=e))

setup.py

@@ -22,7 +22,7 @@ def requirements(filename):
 
 setup(
     name="aeon-venos",
-    version="0.9.5",
+    version="0.9.6",
     author="Jeremy Schulman",
     url='https://github.com/Apstra/aeon-venos',
     author_email="jeremy@apstra.com",

tests/test_get_device.py

@@ -1,6 +1,8 @@
+import socket
 from mock import patch, Mock, call
 import pytest
 from pexpect.pxssh import ExceptionPxssh
+from paramiko import AuthenticationException, SSHException
 from aeon.exceptions import TargetError
 
 from aeon.utils import get_device
@@ -11,30 +13,37 @@
 nos_only_params = [True, False]
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.aeon.nxos.device.Device')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_login(mock_pxssh, mock_nxos_device):
+def test_get_device_login(mock_pxssh, mock_nxos_device, mock_paramiko):
     mock_pxssh.return_value.expect.return_value = 0
     get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
+    mock_paramiko.return_value.connect.assert_called_with(dev_info['target'],
+                                                     username=dev_info['user'],
+                                                     timeout=5,
+                                                     password=dev_info['passwd'])
     mock_pxssh.return_value.login.assert_called_with(dev_info['target'],
                                                      dev_info['user'],
                                                      auto_prompt_reset=False,
                                                      login_timeout=10,
                                                      password=dev_info['passwd'])
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.aeon.cumulus.device.Device')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_sendline(mock_pxssh, mock_nxos_device):
+def test_get_device_sendline(mock_pxssh, mock_nxos_device, mock_paramiko):
     mock_pxssh.return_value.expect.side_effect = [2, 0]
     get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
     calls = [call('show version'), call('cat /proc/version')]
     mock_pxssh.return_value.sendline.assert_has_calls(calls)
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.aeon.nxos.device.Device')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_nxos(mock_pxssh, mock_nxos_device):
+def test_get_device_nxos(mock_pxssh, mock_nxos_device, mock_paramiko):
     device_name = 'nxos_device'
     mock_pxssh.return_value.expect.return_value = 0
     mock_nxos_device.return_value = device_name
@@ -44,9 +53,10 @@ def test_get_device_nxos(mock_pxssh, mock_nxos_device):
     assert dev == device_name
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.aeon.nxos.device.Device')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_nos_only(mock_pxssh, mock_nxos_device):
+def test_get_device_nos_only(mock_pxssh, mock_nxos_device, mock_paramiko):
     device_name = 'nxos'
     mock_pxssh.return_value.expect.return_value = 0
     mock_nxos_device.return_value = device_name
@@ -56,58 +66,88 @@ def test_get_device_nos_only(mock_pxssh, mock_nxos_device):
                      nos_only=True)
     assert dev == device_name
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.aeon.eos.device.Device')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_eos(mock_pxssh, mock_eos_device):
+def test_get_device_eos(mock_pxssh, mock_eos_device, mock_paramiko):
     device_name = 'eos_device'
     mock_pxssh.return_value.expect.return_value = 1
     mock_eos_device.return_value = device_name
     dev = get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
     assert dev == device_name
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.aeon.cumulus.device.Device')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_cumulus(mock_pxssh, mock_cumulus_device):
+def test_get_device_cumulus(mock_pxssh, mock_cumulus_device, mock_paramiko):
     device_name = 'cumulus_device'
     mock_pxssh.return_value.expect.side_effect = [2, 0]
     mock_cumulus_device.return_value = device_name
     dev = get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
     assert dev == device_name
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.aeon.ubuntu.device.Device')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_ubuntu(mock_pxssh, mock_ubuntu_device):
+def test_get_device_ubuntu(mock_pxssh, mock_ubuntu_device, mock_paramiko):
     device_name = 'ubuntu_device'
     mock_pxssh.return_value.expect.side_effect = [2, 1]
     mock_ubuntu_device.return_value = device_name
     dev = get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
     assert dev == device_name
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.aeon.centos.device.Device')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_centos(mock_pxssh, mock_centos_device):
+def test_get_device_centos(mock_pxssh, mock_centos_device, mock_paramiko):
     device_name = 'centos_device'
     mock_pxssh.return_value.expect.side_effect = [2, 2]
     mock_centos_device.return_value = device_name
     dev = get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
     assert dev == device_name
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_target_error(mock_pxssh):
-    mock_pxssh.return_value.expect.side_effect = [2, 3]
+def test_get_device_target_error(mock_pxssh, mock_ssh):
+    mock_pxssh.return_value.expect.side_effect = [2, 4]
     with pytest.raises(TargetError) as e:
         dev = get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
-        assert e.message('Unable to determine device type for %s' % dev_info['target'])
+    assert e.value.message == 'Unable to determine device type for %s' % dev_info['target']
 
 
+@patch('pylib.aeon.utils.paramiko.SSHClient')
 @patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
-def test_get_device_exception(mock_pxssh):
+def test_get_device_exception(mock_pxssh, mock_ssh):
     exception_msg = 'Test Exception Message'
     mock_pxssh.return_value.login.side_effect = ExceptionPxssh(exception_msg)
     with pytest.raises(TargetError) as e:
         dev = get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
-        assert e.message('Error logging in: %s' % dev_info['target'])
+    assert e.value.message == 'Error logging in to {target} : {error}'.format(target=dev_info['target'], error=exception_msg)
+
+
+@patch('pylib.aeon.utils.paramiko.SSHClient')
+def test_get_device_socket_error(mock_paramiko):
+    mock_paramiko.return_value.connect.side_effect = socket.error
+    with pytest.raises(TargetError) as e:
+        get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
+    assert e.value.message == ('Device unreachable: %s' % dev_info['target'])
+
+
+@patch('pylib.aeon.utils.paramiko.SSHClient')
+def test_get_device_paramiko_authentication_exception(mock_paramiko):
+    mock_paramiko.return_value.connect.side_effect = AuthenticationException
+    with pytest.raises(TargetError) as e:
+        get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
+    assert e.value.message == ('Authentication error: %s' % dev_info['target'])
+
+
+@patch('pylib.aeon.utils.paramiko.SSHClient')
+def test_get_device_paramiko_ssh_exception(mock_paramiko):
+    mock_paramiko.return_value.connect.side_effect = SSHException
+    with pytest.raises(TargetError) as e:
+        get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
+    assert e.value.message == ('Error logging in: %s' % dev_info['target'])