Add OPX support and unit tests. (#44)

pylib/aeon/opx/__init__.py

@@ -0,0 +1,4 @@
+# Copyright 2014-present, Apstra, Inc. All rights reserved.
+#
+# This source code is licensed under End User License Agreement found in the
+# LICENSE file at http://www.apstra.com/community/eula

pylib/aeon/opx/device.py

@@ -0,0 +1,52 @@
+# Copyright 2014-present, Apstra, Inc. All rights reserved.
+#
+# This source code is licensed under End User License Agreement found in the
+# LICENSE file at http://www.apstra.com/community/eula
+
+from aeon.base.device import BaseDevice
+from aeon.cumulus.connector import Connector
+
+
+__all__ = ['Device']
+
+
+class Device(BaseDevice):
+    OS_NAME = 'OPX'
+
+    def __init__(self, target, **kwargs):
+        """
+        :param target: hostname or ipaddr of target device
+        :param kwargs:
+            'user' : login user-name, defaults to "admin"
+            'passwd': login password, defaults to "admin
+        """
+        BaseDevice.__init__(self, target, Connector, **kwargs)
+
+    def get_mac_address(self):
+        good, got = self.api.execute(['ip link show'])
+        data = got[0]['stdout']
+        macaddr = data.partition('link/ether ')[-1].split()[0]
+        return macaddr.upper()
+
+    def gather_facts(self):
+
+        facts = self.facts
+        facts['os_name'] = self.OS_NAME
+
+        good, got = self.api.execute([
+            'hostname',
+            """grep -oP '^OS_VERSION="\K.*\d' /etc/OPX-release-version""",
+            """grep -oP '^PLATFORM="\K.*\w' /etc/OPX-release-version"""
+        ])
+
+        facts['fqdn'] = got[0]['stdout'].strip()
+        facts['hostname'] = facts['fqdn']
+        facts['os_version'] = got[1]['stdout'].strip()
+        facts['virtual'] = bool('vm' in got[2]['stdout'].lower())
+        facts['vendor'] = 'OPX'
+        facts['serial_number'] = self.get_mac_address().replace(':', '')
+        facts['mac_address'] = self.get_mac_address()
+        facts['hw_model'] = got[2]['stdout'].strip()
+        facts['hw_part_number'] = None
+        facts['hw_version'] = None
+        facts['service_tag'] = None

setup.py

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

tests/test_opx.py

@@ -0,0 +1,83 @@
+import mock
+import pytest
+from pylib.aeon.opx import device
+from pylib.aeon.cumulus import connector
+
+
+g_facts = {
+    'hw_version': None,
+    'hw_part_number': None,
+    'hostname': 'opx221_vm',
+    'serial_number': '525400A5EC36',
+    'fqdn': 'opx221_vm',
+    'os_version': '2.2.1',
+    'virtual': True,
+    'hw_model': 'S6000-VM',
+    'vendor': 'OPX',
+    'mac_address': '52:54:00:A5:EC:36',
+    'os_name': 'OPX',
+    'service_tag': None
+}
+
+ip_link_show_out = '''
+1: lo: <LOOPBACK> mtu 65536 qdisc noop state DOWN mode DEFAULT group default
+    link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
+2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc pfifo_fast state UP mode DEFAULT group default qlen 1000
+    link/ether 52:54:00:a5:ec:36 brd ff:ff:ff:ff:ff:ff
+'''
+
+hostname_out = "opx221_vm"
+grep_version_out = '2.2.1'
+grep_platform_out = 'S6000-VM'
+
+
+@mock.patch('pylib.aeon.opx.connector.paramiko.SSHClient')
+@pytest.fixture()
+def opx_connector(mock_ssh):
+    hostname = '1.1.1.1'
+    port = 22
+    proto = 'ssh'
+    user = 'test_user'
+    passwd = 'test_passwd'
+    con = connector.Connector(hostname, port=port, proto=proto, user=user, passwd=passwd)
+    return con
+
+
+@mock.patch('pylib.aeon.opx.device.BaseDevice.probe')
+@mock.patch('pylib.aeon.opx.device.Connector')
+@pytest.fixture()
+def opx_device(mock_connector, mock_probe, request):
+    def mock_execute(args, **kwargs):
+        results = []
+        for arg in args:
+            # os_version
+            if arg == """grep -oP '^OS_VERSION="\K.*\d' /etc/OPX-release-version""":
+                results.append({'stdout': grep_version_out})
+            # platform
+            if arg == """grep -oP '^PLATFORM="\K.*\w' /etc/OPX-release-version""":
+                results.append({'stdout': grep_platform_out})
+            # hostname
+            elif arg == 'hostname':
+                results.append({'stdout': hostname_out})
+            elif arg =='ip link show':
+                results.append({'stdout': ip_link_show_out})
+        return True, results
+    mock_connector.return_value.execute.side_effect = mock_execute
+    mock_probe.return_value = True, 10
+    target = '1.1.1.1'
+    user = 'test_user'
+    passwd = 'test_passwd'
+    dev = device.Device(target, user=user, passwd=passwd)
+    return dev
+
+
+def test_opx_device(opx_device):
+    dev = opx_device
+    assert dev.OS_NAME == 'OPX'
+    assert dev.DEFAULT_PROBE_TIMEOUT == 10
+    assert dev.user == 'test_user'
+    assert dev.passwd == 'test_passwd'
+    assert dev.facts == g_facts
+
+
+