wip - test netbox server creation:

dev-requirements.txt

@@ -5,3 +5,5 @@ pytest-cov
 flake8
 flake8-isort
 tox
+mock
+pyfakefs

netbox_agent/cli.py

@@ -18,12 +18,13 @@
 
 
 def run(config):
-    manufacturer = dmidecode.get_by_type('Chassis')[0].get('Manufacturer')
+    dmi = dmidecode.parse()
+    manufacturer = dmidecode.get_by_type(dmi, 'Chassis')[0].get('Manufacturer')
 
     try:
-        server = MANUFACTURERS[manufacturer](dmi=dmidecode)
+        server = MANUFACTURERS[manufacturer](dmi=dmi)
     except KeyError:
-        server = GenericHost
+        server = GenericHost(dmi=dmi)
 
     if config.debug:
         server.print_debug()

netbox_agent/ipmi.py

@@ -35,14 +35,20 @@ class IPMI():
     """
 
     def __init__(self):
-        self.ret, self.output = subprocess.getstatusoutput('ipmitool lan print')
+        try:
+            self.ret, self.output = subprocess.getstatusoutput('ipmitool lan print')
+        except Exception as e:
+            self.ret = 42
+            self.output = str(e)
+
         if self.ret != 0:
             logging.error('Cannot get ipmi info: {}'.format(self.output))
 
     def parse(self):
-        ret = {}
         if self.ret != 0:
-            return ret
+            return None
+
+        ret = {}
         for line in self.output.splitlines():
             key = line.split(':')[0].strip()
             value = ':'.join(line.split(':')[1:]).strip()

netbox_agent/lshw.py

@@ -103,13 +103,14 @@ def find_storage(self, obj):
             self.disks.append(d)
 
     def find_cpus(self, obj):
-        c = {}
-        c["product"] = obj["product"]
-        c["vendor"] = obj["vendor"]
-        c["description"] = obj["description"]
-        c["location"] = obj["slot"]
-
-        self.cpus.append(c)
+        if "product" in obj:
+            c = {}
+            c["product"] = obj["product"]
+            c["vendor"] = obj["vendor"]
+            c["description"] = obj["description"]
+            c["location"] = obj["slot"]
+
+            self.cpus.append(c)
 
     def find_memories(self, obj):
         if "children" not in obj:

netbox_agent/network.py

@@ -234,6 +234,9 @@ def reset_vlan_on_interface(self, nic, interface):
 
     def create_or_update_ipmi(self):
         ipmi = self.get_ipmi()
+        if ipmi is None:
+            return None
+
         mac = ipmi['MAC Address']
         ip = ipmi['IP Address']
         netmask = ipmi['Subnet Mask']

netbox_agent/power.py

@@ -1,5 +1,6 @@
 import logging
 
+import netbox_agent.dmidecode as dmidecode
 from netbox_agent.config import netbox_instance as nb
 
 PSU_DMI_TYPE = 39
@@ -16,7 +17,7 @@ def __init__(self, server=None):
 
     def get_power_supply(self):
         power_supply = []
-        for psu in self.server.dmi.get_by_type(PSU_DMI_TYPE):
+        for psu in dmidecode.get_by_type(self.server.dmi, PSU_DMI_TYPE):
             if 'Present' not in psu['Status'] or psu['Status'] == 'Not Present':
                 continue
 

netbox_agent/vendors/dell.py

@@ -20,7 +20,7 @@ def get_blade_slot(self):
         `        Location In Chassis: Slot 03`
         """
         if self.is_blade():
-            return self.dmi.get_by_type('Baseboard')[0].get('Location In Chassis').strip()
+            return self.baseboard[0].get('Location In Chassis').strip()
         return None
 
     def get_chassis_name(self):
@@ -30,12 +30,12 @@ def get_chassis_name(self):
 
     def get_chassis(self):
         if self.is_blade():
-            return self.dmi.get_by_type('Chassis')[0]['Version'].strip()
+            return self.chassis[0]['Version'].strip()
         return self.get_product_name()
 
     def get_chassis_service_tag(self):
         if self.is_blade():
-            return self.dmi.get_by_type('Chassis')[0]['Serial Number'].strip()
+            return self.chassis[0]['Serial Number'].strip()
         return self.get_service_tag()
 
     def get_power_consumption(self):

netbox_agent/vendors/generic.py

@@ -5,7 +5,7 @@
 class GenericHost(ServerBase):
     def __init__(self, *args, **kwargs):
         super(GenericHost, self).__init__(*args, **kwargs)
-        self.manufacturer = dmidecode.get_by_type('Baseboard')[0].get('Manufacturer')
+        self.manufacturer = dmidecode.get_by_type(self.dmi, 'Baseboard')[0].get('Manufacturer')
 
     def is_blade(self):
         return None

netbox_agent/vendors/hp.py

@@ -1,3 +1,4 @@
+import netbox_agent.dmidecode as dmidecode
 from netbox_agent.server import ServerBase
 
 
@@ -19,7 +20,7 @@ def _find_rack_locator(self):
         """
         # FIXME: make a dmidecode function get_by_dminame() ?
         if self.is_blade():
-            locator = self.dmi.get_by_type(204)
+            locator = dmidecode.get_by_type(self.dmi, 204)
             if self.get_product_name() == 'ProLiant BL460c Gen10':
                 locator = locator[0]['Strings']
                 return {

netbox_agent/vendors/qct.py

@@ -7,12 +7,12 @@ def __init__(self, *args, **kwargs):
         self.manufacturer = 'QCT'
 
     def is_blade(self):
-        return 'Location In Chassis' in self.dmi.get_by_type('Baseboard')[0].keys()
+        return 'Location In Chassis' in self.baseboard[0].keys()
 
     def get_blade_slot(self):
         if self.is_blade():
             return 'Slot {}'.format(
-                self.dmi.get_by_type('Baseboard')[0].get('Location In Chassis').strip()
+                self.baseboard[0].get('Location In Chassis').strip()
             )
         return None
 
@@ -23,10 +23,10 @@ def get_chassis_name(self):
 
     def get_chassis(self):
         if self.is_blade():
-            return self.dmi.get_by_type('Chassis')[0]['Version'].strip()
+            return self.chassis[0]['Version'].strip()
         return self.get_product_name()
 
     def get_chassis_service_tag(self):
         if self.is_blade():
-            return self.dmi.get_by_type('Chassis')[0]['Serial Number'].strip()
+            return self.chassis[0]['Serial Number'].strip()
         return self.get_service_tag()

tests/conftest.py

@@ -15,7 +15,7 @@ def get_fixture_paths(path):
 
 
 def parametrize_with_fixtures(path, base_path='tests/fixtures',
-                              argname='fixture', only_filenames=None):
+                              argname='dmi_fixture', only_filenames=None):
     path = os.path.join(base_path, path)
     fixture_paths = get_fixture_paths(path)
     argvalues = []

tests/constants.py

@@ -0,0 +1 @@
+DEFAULT_DATACENTER = 'DC1'

tests/fixtures/dmidecode/HP_SL4540_Gen8

@@ -829,7 +829,7 @@ System Power Supply
 	Location: Not Specified
 	Name: Power Supply 2
 	Manufacturer: HP
-	Serial Number: 4242
+	Serial Number: 4243
 	Asset Tag: Not Specified
 	Model Part Number: 656364-B21                      
 	Revision: Not Specified
@@ -846,7 +846,7 @@ System Power Supply
 	Location: Not Specified
 	Name: Power Supply 3
 	Manufacturer: HP
-	Serial Number: 4242
+	Serial Number: 4244
 	Asset Tag: Not Specified
 	Model Part Number: 656364-B21                      
 	Revision: Not Specified
@@ -863,7 +863,7 @@ System Power Supply
 	Location: Not Specified
 	Name: Power Supply 4
 	Manufacturer: HP
-	Serial Number: 4242
+	Serial Number: 4245
 	Asset Tag: Not Specified
 	Model Part Number: 656364-B21                      
 	Revision: Not Specified

tests/network.py

@@ -6,15 +6,15 @@
     'lldp/', only_filenames=[
         'dedibox1.txt',
     ])
-def test_lldp_parse_with_port_desc(fixture):
-    lldp = LLDP(fixture)
+def test_lldp_parse_with_port_desc(dmi_fixture):
+    lldp = LLDP(dmi_fixture)
     assert lldp.get_switch_port('enp1s0f0') == 'RJ-9'
 
 
 @parametrize_with_fixtures(
     'lldp/', only_filenames=[
         'qfx.txt',
     ])
-def test_lldp_parse_without_ifname(fixture):
-    lldp = LLDP(fixture)
+def test_lldp_parse_without_ifname(dmi_fixture):
+    lldp = LLDP(dmi_fixture)
     assert lldp.get_switch_port('eth0') == 'xe-0/0/1'

tests/server.py

@@ -1,11 +1,20 @@
+
+import mock
+import netifaces
+
+from netbox_agent.config import config
+from netbox_agent.config import netbox_instance as nb
 from netbox_agent.dmidecode import parse
 from netbox_agent.server import ServerBase
+from netbox_agent.vendors.hp import HPHost
 from tests.conftest import parametrize_with_fixtures
+from tests.constants import DEFAULT_DATACENTER
+from tests.utils import setup_netbox
 
 
 @parametrize_with_fixtures('dmidecode/')
-def test_init(fixture):
-    dmi = parse(fixture)
+def test_init(dmi_fixture):
+    dmi = parse(dmi_fixture)
     server = ServerBase(dmi)
     assert server
 
@@ -17,8 +26,8 @@ def test_init(fixture):
         'HP_DL380p_Gen8',
         'HP_SL4540_Gen8'
     ])
-def test_hp_service_tag(fixture):
-    dmi = parse(fixture)
+def test_hp_service_tag(dmi_fixture):
+    dmi = parse(dmi_fixture)
     server = ServerBase(dmi)
     assert server.get_service_tag() == '4242'
 
@@ -27,8 +36,8 @@ def test_hp_service_tag(fixture):
     'dmidecode/', only_filenames=[
         'unknown.txt'
     ])
-def test_generic_host_service_tag(fixture):
-    dmi = parse(fixture)
+def test_generic_host_service_tag(dmi_fixture):
+    dmi = parse(dmi_fixture)
     server = ServerBase(dmi)
     assert server.get_service_tag() == '42'
 
@@ -37,7 +46,56 @@ def test_generic_host_service_tag(fixture):
     'dmidecode/', only_filenames=[
         'unknown.txt'
     ])
-def test_generic_host_product_name(fixture):
-    dmi = parse(fixture)
+def test_generic_host_product_name(dmi_fixture):
+    dmi = parse(dmi_fixture)
     server = ServerBase(dmi)
     assert server.get_product_name() == 'SR'
+
+
+@mock.patch('netifaces.ifaddresses')
+@mock.patch('netifaces.interfaces')
+@parametrize_with_fixtures(
+    'dmidecode/', only_filenames=[
+        'HP_SL4540_Gen8',
+    ], argname='dmi_fixture')
+def test_create_server(
+        mock_interfaces,
+        mock_ifaddresses,
+        fs,
+        dmi_fixture,
+):
+    fake_addresses = {}
+    fake_addresses[netifaces.AF_INET] = [{'addr': '42.42.42.42', 'netmask': '255.255.255.0'}]
+    fake_addresses[netifaces.AF_LINK] = [{'addr': 'a8:1e:84:f2:9e:69'}]
+
+    mock_interfaces.return_value = ['enp1s0f0']
+    mock_ifaddresses.return_value = fake_addresses
+
+    dmi = parse(dmi_fixture)
+    server = HPHost(dmi)
+
+    setup_netbox(
+        DEFAULT_DATACENTER,
+        'Server',
+        'HP',
+        server.get_product_name(),
+    )
+
+    # Create fake /sys/class/net directory with fake interface and MAC addr
+    fs.create_file('/tmp/enp1s0f0/address', contents='a8:1e:84:f2:9e:69')
+    fs.create_symlink('/sys/class/net/enp1s0f0', '/tmp/enp1s0f0')
+
+    server.netbox_create(config)
+
+    # Check serial tag is correct
+    assert server.get_service_tag() == '4242'
+    network_card = server.network.get_netbox_network_card({'name': 'enp1s0f0', 'mac': None})
+
+    # Check network card is correct
+    assert network_card.name == 'enp1s0f0'
+
+    # Check IP on network card
+    ips = nb.ipam.ip_addresses.filter(
+        interface_id=network_card.id
+    )
+    assert ips[0].address == '42.42.42.42/24'

tests/utils.py

@@ -0,0 +1,51 @@
+
+import pynetbox
+
+from netbox_agent.config import netbox_instance as nb
+
+
+def setup_netbox(dc, device_role, manufacturer, model):
+    try:
+        nb_dc = nb.dcim.sites.create(
+            name=dc,
+            slug=dc.lower(),
+        )
+    except pynetbox.RequestError:
+        nb_dc = nb.dcim.sites.get(
+            slug=dc.lower()
+        )
+
+    try:
+        nb_manufacturer = nb.dcim.manufacturers.create(
+            name=manufacturer,
+            slug=manufacturer.lower(),
+        )
+    except pynetbox.RequestError:
+        nb_manufacturer = nb.dcim.manufacturers.get(
+            slug=manufacturer.lower()
+        )
+
+    try:
+        nb_device_role = nb.dcim.device_roles.create(
+            name=device_role,
+            slug=device_role.lower(),
+            color='f44336',
+        )
+    except pynetbox.RequestError:
+        nb_device_role = nb.dcim.device_roles.get(
+            slug=device_role.lower(),
+        )
+
+    try:
+        nb_device_type = nb.dcim.device_types.create(
+            model=model,
+            slug=model.lower().replace(' ', '_'),
+            device_role=nb_device_role.id,
+            manufacturer=nb_manufacturer.id,
+        )
+    except pynetbox.RequestError:
+        nb_device_type = nb.dcim.device_types.get(
+            slug=model.lower().replace(' ', '_'),
+        )
+
+    return nb_dc, nb_manufacturer, nb_device_role, nb_device_type