Added automatic device type detection (#33)

* Added automatic device type detection

* Fixed test failures

* Added ability to check nos only. Does not return object

README.md

@@ -8,3 +8,39 @@ AEON-VENOS's primary purpose:
  - Create a common Python object model to interact with different vendors' APIs.
  - Gather facts from those devices.
  - Provide a communications channel to push and pull information from those devices.
+
+
+## Usage
+```python
+from aeon.eos.device import Device
+dev = Device('10.0.0.100', user='user', passwd='passwd')
+
+dev.facts
+{'os_name': 'eos', 'vendor': 'arista', 'hw_part_number': None, 'mac_address': u'52:54:00:e8:6c:ca', 
+'serial_number': u'525400E86CCA', 'chassis_id': None, 'fqdn': u'localhost.com', 'os_version': u'4.18.4F', 
+'virtual': True, 'hw_model': u'vEOS', 'hw_version': u'', 'hw_part_version': None, 'hostname': u'localhost'}
+
+dev.api.execute('show version')
+{u'memTotal': 3887540, u'version': u'4.18.4F', u'internalVersion': u'4.18.4F-5927901.4184F', u'serialNumber': u'', 
+u'systemMacAddress': u'52:54:00:e8:6c:ca', u'bootupTimestamp': 1506689787.23, u'memFree': 2463080, 
+u'modelName': u'vEOS', u'architecture': u'i386', u'isIntlVersion': False, 
+u'internalBuildId': u'763e44bf-fb89-4c3e-b735-c373d7a65ee4', u'hardwareRevision': u''}
+```
+
+**Automatically determine device type**
+```python
+from aeon.utils import get_device
+dev = get_device('10.0.0.100', user='user', passwd='passwd')
+dev.facts
+{'os_name': 'eos', 'vendor': 'arista', 'hw_part_number': None, 'mac_address': u'52:54:00:e8:6c:ca', 
+'serial_number': u'525400E86CCA', 'chassis_id': None, 'fqdn': u'localhost.com', 'os_version': u'4.18.4F', 
+'virtual': True, 'hw_model': u'vEOS', 'hw_version': u'', 'hw_part_version': None, 'hostname': u'localhost'}
+
+```
+
+**Get NOS name only**
+```python
+from aeon.utils import get_device
+get_device('10.0.0.100', user='user', passwd='passwd', nos_only=True)
+'eos'
+```

pylib/aeon/utils/__init__.py

@@ -2,3 +2,68 @@
 #
 # This source code is licensed under End User License Agreement found in the
 # LICENSE file at http://www.apstra.com/community/eula
+
+import pexpect
+from pexpect import pxssh
+
+import aeon.nxos.device
+import aeon.eos.device
+import aeon.cumulus.device
+import aeon.ubuntu.device
+import aeon.centos.device
+from aeon.exceptions import TargetError
+
+
+def get_device(target=None, user='admin', passwd='admin', nos_only=False):
+    """
+    Automatically determine device type based on device interrogation.
+    :param target: IP address or hostname of target
+    :param user: Username to login to target
+    :param passwd: Password to login to target
+    :param nos_only: Only check for device nos and return as string, do not return device object
+    :return: Device object
+    """
+    dev_table = {
+        'nxos': aeon.nxos.device.Device,
+        'eos': aeon.eos.device.Device,
+        'cumulus': aeon.cumulus.device.Device,
+        'ubuntu': aeon.ubuntu.device.Device,
+        'centos': aeon.centos.device.Device
+    }
+    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:
+            nos = 'nxos'
+        if i == 1:
+            nos = 'eos'
+
+        if i == 2 or i == 3:
+            session.sendline('cat /proc/version')
+            i = session.expect(['cumulus', 'Ubuntu', 'Red Hat'], timeout=10)
+            if i == 0:
+                nos = 'cumulus'
+            if i == 1:
+                nos = 'ubuntu'
+            if i == 2:
+                nos = 'centos'
+            if i == 3:
+                raise TargetError('Unable to determine device type for %s' % target)
+
+    except pxssh.ExceptionPxssh as e:
+        return "Error logging in: %s" % e
+
+    finally:
+        session.close()
+
+    if nos_only:
+        return nos
+    else:
+        return dev_table[nos](target, user=user, passwd=passwd)

requirements.txt

@@ -2,3 +2,4 @@ requests
 lxml
 pyeapi
 paramiko<2.0.0
+pexpect==4.2.1

setup.py

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

tests/test_get_device.py

@@ -0,0 +1,112 @@
+from mock import patch, Mock, call
+import pytest
+from pexpect.pxssh import ExceptionPxssh
+from aeon.exceptions import TargetError
+
+from aeon.utils import get_device
+
+dev_info = {'target': '1.1.1.1',
+       'user': 'test_user',
+       'passwd': 'test_passwd'}
+nos_only_params = [True, False]
+
+
+@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):
+    mock_pxssh.return_value.expect.return_value = 0
+    get_device(target=dev_info['target'], user=dev_info['user'], passwd=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.aeon.cumulus.device.Device')
+@patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
+def test_get_device_sendline(mock_pxssh, mock_nxos_device):
+    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.aeon.nxos.device.Device')
+@patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
+def test_get_device_nxos(mock_pxssh, mock_nxos_device):
+    device_name = 'nxos_device'
+    mock_pxssh.return_value.expect.return_value = 0
+    mock_nxos_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.aeon.nxos.device.Device')
+@patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
+def test_get_device_nos_only(mock_pxssh, mock_nxos_device):
+    device_name = 'nxos'
+    mock_pxssh.return_value.expect.return_value = 0
+    mock_nxos_device.return_value = device_name
+    dev = get_device(target=dev_info['target'],
+                     user=dev_info['user'],
+                     passwd=dev_info['passwd'],
+                     nos_only=True)
+    assert dev == device_name
+
+@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):
+    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.aeon.cumulus.device.Device')
+@patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
+def test_get_device_cumulus(mock_pxssh, mock_cumulus_device):
+    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.aeon.ubuntu.device.Device')
+@patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
+def test_get_device_ubuntu(mock_pxssh, mock_ubuntu_device):
+    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.aeon.centos.device.Device')
+@patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
+def test_get_device_centos(mock_pxssh, mock_centos_device):
+    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.pexpect.pxssh.pxssh')
+def test_get_device_target_error(mock_pxssh):
+    mock_pxssh.return_value.expect.side_effect = [2, 3]
+    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'])
+
+
+@patch('pylib.aeon.utils.pexpect.pxssh.pxssh')
+def test_get_device_exception(mock_pxssh):
+    exception_msg = 'Test Exception Message'
+    mock_pxssh.return_value.login.side_effect = ExceptionPxssh(exception_msg)
+    dev = get_device(target=dev_info['target'], user=dev_info['user'], passwd=dev_info['passwd'])
+    assert dev == 'Error logging in: %s' % exception_msg