Merge pull request 'Fix disk size inventory' (#310) from ribetm/disk_size into main

Reviewed-on: #310

.envrc

@@ -0,0 +1 @@
+use nix

.gitignore

@@ -182,3 +182,5 @@ dmypy.json
 
 netbox-docker
 /.vscode
+.direnv
+.pre-commit-config.yaml

default.nix

@@ -0,0 +1,55 @@
+{
+  sources ? import ./npins,
+  pkgs ? import sources.nixpkgs { },
+}:
+
+let
+  checks = (import sources.git-hooks).run {
+    src = ./.;
+
+    hooks =
+      {
+        commitizen.enable = true;
+      }
+      // (pkgs.lib.genAttrs
+        [
+          "black"
+          "isort"
+          "ruff"
+        ]
+        (hook: {
+          enable = true;
+          stages = [ "pre-push" ];
+        })
+      );
+  };
+
+  python3 = pkgs.python3.override {
+    packageOverrides = self: _: {
+      netifaces2 = self.callPackage ./nix/netifaces2.nix { };
+    };
+  };
+in
+
+{
+  devShell = pkgs.mkShell {
+    name = "netbox-agent.dev";
+
+    packages = [
+      (python3.withPackages (ps: [
+        ps.pynetbox
+        ps.netaddr
+        ps.netifaces2
+        ps.pyyaml
+        ps.jsonargparse
+        ps.python-slugify
+        ps.packaging
+        ps.distro
+      ]))
+    ] ++ checks.enabledPackages;
+
+    shellHook = ''
+      ${checks.shellHook}
+    '';
+  };
+}

netbox_agent/__init__.py

@@ -1,6 +1,7 @@
-from pkg_resources import DistributionNotFound, get_distribution
+from importlib.metadata import PackageNotFoundError
+from importlib.metadata import version as _get_version
 
 try:
-    __version__ = get_distribution(__name__).version
-except DistributionNotFound:
+    __version__ = _get_version(__name__)
+except PackageNotFoundError:
     pass

netbox_agent/cli.py

@@ -1,4 +1,5 @@
 from packaging import version
+
 import netbox_agent.dmidecode as dmidecode
 from netbox_agent.config import config
 from netbox_agent.config import netbox_instance as nb
@@ -11,12 +12,12 @@ from netbox_agent.vendors.supermicro import SupermicroHost
 from netbox_agent.virtualmachine import VirtualMachine, is_vm
 
 MANUFACTURERS = {
-    'Dell Inc.': DellHost,
-    'HP': HPHost,
-    'HPE': HPHost,
-    'Supermicro': SupermicroHost,
-    'Quanta Cloud Technology Inc.': QCTHost,
-    'Generic': GenericHost,
+    "Dell Inc.": DellHost,
+    "HP": HPHost,
+    "HPE": HPHost,
+    "Supermicro": SupermicroHost,
+    "Quanta Cloud Technology Inc.": QCTHost,
+    "Generic": GenericHost,
 }
 
 
@@ -25,21 +26,29 @@ def run(config):
 
     if config.virtual.enabled or is_vm(dmi):
         if not config.virtual.cluster_name:
-            raise Exception('virtual.cluster_name parameter is mandatory because it\'s a VM')
+            raise Exception(
+                "virtual.cluster_name parameter is mandatory because it's a VM"
+            )
         server = VirtualMachine(dmi=dmi)
     else:
-        manufacturer = dmidecode.get_by_type(dmi, 'Chassis')[0].get('Manufacturer')
+        manufacturer = dmidecode.get_by_type(dmi, "Chassis")[0].get("Manufacturer")
         try:
             server = MANUFACTURERS[manufacturer](dmi=dmi)
         except KeyError:
             server = GenericHost(dmi=dmi)
 
-    if version.parse(nb.version) < version.parse('3.7'):
-        print('netbox-agent is not compatible with Netbox prior to version 3.7')
+    if version.parse(nb.version) < version.parse("3.7"):
+        print("netbox-agent is not compatible with Netbox prior to version 3.7")
         return False
 
-    if config.register or config.update_all or config.update_network or \
-       config.update_location or config.update_inventory or config.update_psu:
+    if (
+        config.register
+        or config.update_all
+        or config.update_network
+        or config.update_location
+        or config.update_inventory
+        or config.update_psu
+    ):
         server.netbox_create_or_update(config)
     if config.debug:
         server.print_debug()
@@ -47,8 +56,8 @@ def run(config):
 
 
 def main():
-    return run(config)
+    return 0 if run(config) else 1
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()

netbox_agent/config.py

@@ -10,85 +10,148 @@ import urllib3
 def get_config():
     p = jsonargparse.ArgumentParser(
         default_config_files=[
-            '/etc/netbox_agent.yaml',
-            '~/.config/netbox_agent.yaml',
-            '~/.netbox_agent.yaml',
+            "/etc/netbox_agent.yaml",
+            "~/.config/netbox_agent.yaml",
+            "~/.netbox_agent.yaml",
         ],
-        prog='netbox_agent',
+        prog="netbox_agent",
         description="Netbox agent to run on your infrastructure's servers",
-        env_prefix='NETBOX_AGENT_',
-        default_env=True
+        env_prefix="NETBOX_AGENT_",
+        default_env=True,
     )
-    p.add_argument('-c', '--config', action=jsonargparse.ActionConfigFile)
+    p.add_argument("-c", "--config", action=jsonargparse.ActionConfigFile)
 
-    p.add_argument('-r', '--register', action='store_true', help='Register server to Netbox')
-    p.add_argument('-u', '--update-all', action='store_true', help='Update all infos in Netbox')
-    p.add_argument('-d', '--debug', action='store_true', help='Print debug infos')
-    p.add_argument('--update-network', action='store_true', help='Update network')
-    p.add_argument('--update-inventory', action='store_true', help='Update inventory')
-    p.add_argument('--update-location', action='store_true', help='Update location')
-    p.add_argument('--update-psu', action='store_true', help='Update PSU')
-    p.add_argument('--purge-old-devices', action='store_true',
-                   help='Purge existing (old ?) devices having same name but different serial')
-    p.add_argument('--expansion-as-device', action='store_true',
-                   help='Manage blade expansions as external devices')
+    p.add_argument(
+        "-r", "--register", action="store_true", help="Register server to Netbox"
+    )
+    p.add_argument(
+        "-u", "--update-all", action="store_true", help="Update all infos in Netbox"
+    )
+    p.add_argument("-d", "--debug", action="store_true", help="Print debug infos")
+    p.add_argument("--update-network", action="store_true", help="Update network")
+    p.add_argument("--update-inventory", action="store_true", help="Update inventory")
+    p.add_argument("--update-location", action="store_true", help="Update location")
+    p.add_argument("--update-psu", action="store_true", help="Update PSU")
+    p.add_argument(
+        "--purge-old-devices",
+        action="store_true",
+        help="Purge existing (old ?) devices having same name but different serial",
+    )
+    p.add_argument(
+        "--expansion-as-device",
+        action="store_true",
+        help="Manage blade expansions as external devices",
+    )
 
-    p.add_argument('--log_level', default='debug')
-    p.add_argument('--netbox.ssl_ca_certs_file', help='SSL CA certificates file')
-    p.add_argument('--netbox.url', help='Netbox URL')
-    p.add_argument('--netbox.token', help='Netbox API Token')
-    p.add_argument('--netbox.ssl_verify', default=True, action='store_true',
-                   help='Disable SSL verification')
-    p.add_argument('--virtual.enabled', action='store_true', help='Is a virtual machine or not')
-    p.add_argument('--virtual.cluster_name', help='Cluster name of VM')
-    p.add_argument('--hostname_cmd', default=None,
-                   help="Command to output hostname, used as Device's name in netbox")
-    p.add_argument('--device.platform', default=None,
-                   help='Override device platform. Here we use OS distribution.')
-    p.add_argument('--device.tags', default=r'',
-                   help='tags to use for a host')
-    p.add_argument('--preserve-tags', action='store_true', help='Append new unique tags, preserve those already present')
-    p.add_argument('--device.custom_fields', default=r'',
-                   help='custom_fields to use for a host, eg: field1=v1,field2=v2')
-    p.add_argument('--device.blade_role', default=r'Blade',
-                   help='role to use for a blade server')
-    p.add_argument('--device.chassis_role', default=r'Server Chassis',
-                   help='role to use for a chassis')
-    p.add_argument('--device.server_role', default=r'Server',
-                   help='role to use for a server')
-    p.add_argument('--tenant.driver',
-                   help='tenant driver, ie cmd, file')
-    p.add_argument('--tenant.driver_file',
-                   help='tenant driver custom driver file path')
-    p.add_argument('--tenant.regex',
-                   help='tenant regex to extract Netbox tenant slug')
-    p.add_argument('--datacenter_location.driver',
-                   help='Datacenter location driver, ie: cmd, file')
-    p.add_argument('--datacenter_location.driver_file',
-                   help='Datacenter location custom driver file path')
-    p.add_argument('--datacenter_location.regex',
-                   help='Datacenter location regex to extract Netbox DC slug')
-    p.add_argument('--rack_location.driver', help='Rack location driver, ie: cmd, file')
-    p.add_argument('--rack_location.driver_file', help='Rack location custom driver file path')
-    p.add_argument('--rack_location.regex', help='Rack location regex to extract Rack name')
-    p.add_argument('--slot_location.driver', help='Slot location driver, ie: cmd, file')
-    p.add_argument('--slot_location.driver_file', help='Slot location custom driver file path')
-    p.add_argument('--slot_location.regex', help='Slot location regex to extract slot name')
-    p.add_argument('--network.ignore_interfaces', default=r'(dummy.*|docker.*)',
-                   help='Regex to ignore interfaces')
-    p.add_argument('--network.ignore_ips', default=r'^(127\.0\.0\..*|fe80.*|::1.*)',
-                   help='Regex to ignore IPs')
-    p.add_argument('--network.ipmi', default=True, help='Enable gathering IPMI information')
-    p.add_argument('--network.lldp', help='Enable auto-cabling feature through LLDP infos')
-    p.add_argument('--inventory', action='store_true',
-                   help='Enable HW inventory (CPU, Memory, RAID Cards, Disks) feature')
-    p.add_argument('--process-virtual-drives', action='store_true',
-                   help='Process virtual drives information from RAID '
-                        'controllers to fill disk custom_fields')
-    p.add_argument('--force-disk-refresh', action='store_true',
-                   help='Forces disks detection reprocessing')
-    p.add_argument('--dump-disks-map',
-                   help='File path to dump physical/virtual disks map')
+    p.add_argument("--log_level", default="debug")
+    p.add_argument("--netbox.ssl_ca_certs_file", help="SSL CA certificates file")
+    p.add_argument("--netbox.url", help="Netbox URL")
+    p.add_argument("--netbox.token", help="Netbox API Token")
+    p.add_argument(
+        "--netbox.ssl_verify",
+        default=True,
+        action="store_true",
+        help="Disable SSL verification",
+    )
+    p.add_argument(
+        "--virtual.enabled", action="store_true", help="Is a virtual machine or not"
+    )
+    p.add_argument("--virtual.cluster_name", help="Cluster name of VM")
+    p.add_argument(
+        "--hostname_cmd",
+        default=None,
+        help="Command to output hostname, used as Device's name in netbox",
+    )
+    p.add_argument(
+        "--device.platform",
+        default=None,
+        help="Override device platform. Here we use OS distribution.",
+    )
+    p.add_argument("--device.tags", default=r"", help="tags to use for a host")
+    p.add_argument(
+        "--preserve-tags",
+        action="store_true",
+        help="Append new unique tags, preserve those already present",
+    )
+    p.add_argument(
+        "--device.custom_fields",
+        default=r"",
+        help="custom_fields to use for a host, eg: field1=v1,field2=v2",
+    )
+    p.add_argument(
+        "--device.blade_role", default=r"Blade", help="role to use for a blade server"
+    )
+    p.add_argument(
+        "--device.chassis_role",
+        default=r"Server Chassis",
+        help="role to use for a chassis",
+    )
+    p.add_argument(
+        "--device.server_role", default=r"Server", help="role to use for a server"
+    )
+    p.add_argument("--tenant.driver", help="tenant driver, ie cmd, file")
+    p.add_argument("--tenant.driver_file", help="tenant driver custom driver file path")
+    p.add_argument("--tenant.regex", help="tenant regex to extract Netbox tenant slug")
+    p.add_argument(
+        "--datacenter_location.driver", help="Datacenter location driver, ie: cmd, file"
+    )
+    p.add_argument(
+        "--datacenter_location.driver_file",
+        help="Datacenter location custom driver file path",
+    )
+    p.add_argument(
+        "--datacenter_location.regex",
+        help="Datacenter location regex to extract Netbox DC slug",
+    )
+    p.add_argument("--rack_location.driver", help="Rack location driver, ie: cmd, file")
+    p.add_argument(
+        "--rack_location.driver_file", help="Rack location custom driver file path"
+    )
+    p.add_argument(
+        "--rack_location.regex", help="Rack location regex to extract Rack name"
+    )
+    p.add_argument("--slot_location.driver", help="Slot location driver, ie: cmd, file")
+    p.add_argument(
+        "--slot_location.driver_file", help="Slot location custom driver file path"
+    )
+    p.add_argument(
+        "--slot_location.regex", help="Slot location regex to extract slot name"
+    )
+    p.add_argument(
+        "--network.ignore_interfaces",
+        default=r"(dummy.*|docker.*)",
+        help="Regex to ignore interfaces",
+    )
+    p.add_argument(
+        "--network.ignore_ips",
+        default=r"^(127\.0\.0\..*|fe80.*|::1.*)",
+        help="Regex to ignore IPs",
+    )
+    p.add_argument(
+        "--network.ipmi", default=True, help="Enable gathering IPMI information"
+    )
+    p.add_argument(
+        "--network.lldp", help="Enable auto-cabling feature through LLDP infos"
+    )
+    p.add_argument(
+        "--inventory",
+        action="store_true",
+        help="Enable HW inventory (CPU, Memory, RAID Cards, Disks) feature",
+    )
+    p.add_argument(
+        "--process-virtual-drives",
+        action="store_true",
+        help="Process virtual drives information from RAID "
+        "controllers to fill disk custom_fields",
+    )
+    p.add_argument(
+        "--force-disk-refresh",
+        action="store_true",
+        help="Forces disks detection reprocessing",
+    )
+    p.add_argument(
+        "--dump-disks-map", help="File path to dump physical/virtual disks map"
+    )
 
     options = p.parse_args()
     return options
@@ -99,7 +162,7 @@ config = get_config()
 
 def get_netbox_instance():
     if config.netbox.url is None or config.netbox.token is None:
-        logging.error('Netbox URL and token are mandatory')
+        logging.error("Netbox URL and token are mandatory")
         sys.exit(1)
 
     nb = pynetbox.api(

netbox_agent/dmidecode.py

@@ -5,55 +5,57 @@ import sys
 
 from netbox_agent.misc import is_tool
 
-_handle_re = _re.compile('^Handle\\s+(.+),\\s+DMI\\s+type\\s+(\\d+),\\s+(\\d+)\\s+bytes$')
-_in_block_re = _re.compile('^\\t\\t(.+)$')
-_record_re = _re.compile('\\t(.+):\\s+(.+)$')
-_record2_re = _re.compile('\\t(.+):$')
+_handle_re = _re.compile(
+    "^Handle\\s+(.+),\\s+DMI\\s+type\\s+(\\d+),\\s+(\\d+)\\s+bytes$"
+)
+_in_block_re = _re.compile("^\\t\\t(.+)$")
+_record_re = _re.compile("\\t(.+):\\s+(.+)$")
+_record2_re = _re.compile("\\t(.+):$")
 
 _type2str = {
-    0: 'BIOS',
-    1: 'System',
-    2: 'Baseboard',
-    3: 'Chassis',
-    4: 'Processor',
-    5: 'Memory Controller',
-    6: 'Memory Module',
-    7: 'Cache',
-    8: 'Port Connector',
-    9: 'System Slots',
-    10: ' On Board Devices',
-    11: ' OEM Strings',
-    12: ' System Configuration Options',
-    13: ' BIOS Language',
-    14: ' Group Associations',
-    15: ' System Event Log',
-    16: ' Physical Memory Array',
-    17: ' Memory Device',
-    18: ' 32-bit Memory Error',
-    19: ' Memory Array Mapped Address',
-    20: ' Memory Device Mapped Address',
-    21: ' Built-in Pointing Device',
-    22: ' Portable Battery',
-    23: ' System Reset',
-    24: ' Hardware Security',
-    25: ' System Power Controls',
-    26: ' Voltage Probe',
-    27: ' Cooling Device',
-    28: ' Temperature Probe',
-    29: ' Electrical Current Probe',
-    30: ' Out-of-band Remote Access',
-    31: ' Boot Integrity Services',
-    32: ' System Boot',
-    33: ' 64-bit Memory Error',
-    34: ' Management Device',
-    35: ' Management Device Component',
-    36: ' Management Device Threshold Data',
-    37: ' Memory Channel',
-    38: ' IPMI Device',
-    39: ' Power Supply',
-    40: ' Additional Information',
-    41: ' Onboard Devices Extended Information',
-    42: ' Management Controller Host Interface'
+    0: "BIOS",
+    1: "System",
+    2: "Baseboard",
+    3: "Chassis",
+    4: "Processor",
+    5: "Memory Controller",
+    6: "Memory Module",
+    7: "Cache",
+    8: "Port Connector",
+    9: "System Slots",
+    10: " On Board Devices",
+    11: " OEM Strings",
+    12: " System Configuration Options",
+    13: " BIOS Language",
+    14: " Group Associations",
+    15: " System Event Log",
+    16: " Physical Memory Array",
+    17: " Memory Device",
+    18: " 32-bit Memory Error",
+    19: " Memory Array Mapped Address",
+    20: " Memory Device Mapped Address",
+    21: " Built-in Pointing Device",
+    22: " Portable Battery",
+    23: " System Reset",
+    24: " Hardware Security",
+    25: " System Power Controls",
+    26: " Voltage Probe",
+    27: " Cooling Device",
+    28: " Temperature Probe",
+    29: " Electrical Current Probe",
+    30: " Out-of-band Remote Access",
+    31: " Boot Integrity Services",
+    32: " System Boot",
+    33: " 64-bit Memory Error",
+    34: " Management Device",
+    35: " Management Device Component",
+    36: " Management Device Threshold Data",
+    37: " Memory Channel",
+    38: " IPMI Device",
+    39: " Power Supply",
+    40: " Additional Information",
+    41: " Onboard Devices Extended Information",
+    42: " Management Controller Host Interface",
 }
 _str2type = {}
 for type_id, type_str in _type2str.items():
@@ -70,7 +72,7 @@ def parse(output=None):
     else:
         buffer = _execute_cmd()
     if isinstance(buffer, bytes):
-        buffer = buffer.decode('utf-8')
+        buffer = buffer.decode("utf-8")
     _data = _parse(buffer)
     return _data
 
@@ -129,24 +131,31 @@ def get_by_type(data, type_id):
 
     result = []
     for entry in data.values():
-        if entry['DMIType'] == type_id:
+        if entry["DMIType"] == type_id:
             result.append(entry)
 
     return result
 
 
 def _execute_cmd():
-    if not is_tool('dmidecode'):
-        logging.error('Dmidecode does not seem to be present on your system. Add it your path or '
-                      'check the compatibility of this project with your distro.')
+    if not is_tool("dmidecode"):
+        logging.error(
+            "Dmidecode does not seem to be present on your system. Add it your path or "
+            "check the compatibility of this project with your distro."
+        )
         sys.exit(1)
-    return _subprocess.check_output(['dmidecode', ], stderr=_subprocess.PIPE)
+    return _subprocess.check_output(
+        [
+            "dmidecode",
+        ],
+        stderr=_subprocess.PIPE,
+    )
 
 
 def _parse(buffer):
     output_data = {}
     #  Each record is separated by double newlines
-    split_output = buffer.split('\n\n')
+    split_output = buffer.split("\n\n")
 
     for record in split_output:
         record_element = record.splitlines()
@@ -164,21 +173,21 @@ def _parse(buffer):
         dmi_handle = handle_data[0]
 
         output_data[dmi_handle] = {}
-        output_data[dmi_handle]['DMIType'] = int(handle_data[1])
-        output_data[dmi_handle]['DMISize'] = int(handle_data[2])
+        output_data[dmi_handle]["DMIType"] = int(handle_data[1])
+        output_data[dmi_handle]["DMISize"] = int(handle_data[2])
 
         #  Okay, we know 2nd line == name
-        output_data[dmi_handle]['DMIName'] = record_element[1]
+        output_data[dmi_handle]["DMIName"] = record_element[1]
 
-        in_block_elemet = ''
-        in_block_list = ''
+        in_block_elemet = ""
+        in_block_list = ""
 
         #  Loop over the rest of the record, gathering values
         for i in range(2, len(record_element), 1):
             if i >= len(record_element):
                 break
             #  Check whether we are inside a \t\t block
-            if in_block_elemet != '':
+            if in_block_elemet != "":
                 in_block_data = _in_block_re.findall(record_element[i])
 
                 if in_block_data:
@@ -192,7 +201,7 @@ def _parse(buffer):
                 else:
                     # We are out of the \t\t block; reset it again, and let
                     # the parsing continue
-                    in_block_elemet = ''
+                    in_block_elemet = ""
 
             record_data = _record_re.findall(record_element[i])
 
@@ -208,7 +217,7 @@ def _parse(buffer):
                 #  This is an array of data - let the loop know we are inside
                 #  an array block
                 in_block_elemet = record_data2[0]
-                in_block_list = ''
+                in_block_list = ""
 
                 continue
 

netbox_agent/drivers/file.py

@@ -2,7 +2,7 @@ import re
 
 
 def get(value, regex):
-    for line in open(value, 'r'):
+    for line in open(value, "r"):
         r = re.search(regex, line)
         if r and len(r.groups()) > 0:
             return r.groups()[0]

netbox_agent/ethtool.py

@@ -6,16 +6,16 @@ from shutil import which
 
 # mapping fields from ethtool output to simple names
 field_map = {
-    'Supported ports': 'ports',
-    'Supported link modes': 'sup_link_modes',
-    'Supports auto-negotiation': 'sup_autoneg',
-    'Advertised link modes': 'adv_link_modes',
-    'Advertised auto-negotiation': 'adv_autoneg',
-    'Speed': 'speed',
-    'Duplex': 'duplex',
-    'Port': 'port',
-    'Auto-negotiation': 'autoneg',
-    'Link detected': 'link',
+    "Supported ports": "ports",
+    "Supported link modes": "sup_link_modes",
+    "Supports auto-negotiation": "sup_autoneg",
+    "Advertised link modes": "adv_link_modes",
+    "Advertised auto-negotiation": "adv_autoneg",
+    "Speed": "speed",
+    "Duplex": "duplex",
+    "Port": "port",
+    "Auto-negotiation": "autoneg",
+    "Link detected": "link",
 }
 
 
@@ -25,7 +25,7 @@ def merge_two_dicts(x, y):
     return z
 
 
-class Ethtool():
+class Ethtool:
     """
     This class aims to parse ethtool output
     There is several bindings to have something proper, but it requires
@@ -40,39 +40,40 @@ class Ethtool():
         parse ethtool output
         """
 
-        output = subprocess.getoutput('ethtool {}'.format(self.interface))
+        output = subprocess.getoutput("ethtool {}".format(self.interface))
 
         fields = {}
-        field = ''
-        fields['speed'] = '-'
-        fields['link'] = '-'
-        fields['duplex'] = '-'
-        for line in output.split('\n')[1:]:
+        field = ""
+        fields["speed"] = "-"
+        fields["link"] = "-"
+        fields["duplex"] = "-"
+        for line in output.split("\n")[1:]:
             line = line.rstrip()
-            r = line.find(':')
+            r = line.find(":")
             if r > 0:
                 field = line[:r].strip()
                 if field not in field_map:
                     continue
                 field = field_map[field]
-                output = line[r + 1:].strip()
+                output = line[r + 1 :].strip()
                 fields[field] = output
             else:
-                if len(field) > 0 and \
-                   field in field_map:
-                    fields[field] += ' ' + line.strip()
+                if len(field) > 0 and field in field_map:
+                    fields[field] += " " + line.strip()
         return fields
 
     def _parse_ethtool_module_output(self):
-        status, output = subprocess.getstatusoutput('ethtool -m {}'.format(self.interface))
+        status, output = subprocess.getstatusoutput(
+            "ethtool -m {}".format(self.interface)
+        )
         if status == 0:
-            r = re.search(r'Identifier.*\((\w+)\)', output)
+            r = re.search(r"Identifier.*\((\w+)\)", output)
             if r and len(r.groups()) > 0:
-                return {'form_factor': r.groups()[0]}
+                return {"form_factor": r.groups()[0]}
         return {}
 
     def parse(self):
-        if which('ethtool') is None:
+        if which("ethtool") is None:
             return None
         output = self._parse_ethtool_output()
         output.update(self._parse_ethtool_module_output())

netbox_agent/inventory.py

@@ -1,3 +1,11 @@
+import json
+import logging
+import re
+import sys
+import traceback
+
+import pynetbox
+
 from netbox_agent.config import config
 from netbox_agent.config import netbox_instance as nb
 from netbox_agent.lshw import LSHW
@@ -5,25 +13,19 @@ from netbox_agent.misc import get_vendor, is_tool
 from netbox_agent.raid.hp import HPRaid
 from netbox_agent.raid.omreport import OmreportRaid
 from netbox_agent.raid.storcli import StorcliRaid
-import traceback
-import pynetbox
-import logging
-import json
-import re
-
 
 INVENTORY_TAG = {
-    'cpu': {'name': 'hw:cpu', 'slug': 'hw-cpu'},
-    'gpu': {'name': 'hw:gpu', 'slug': 'hw-gpu'},
-    'disk': {'name': 'hw:disk', 'slug': 'hw-disk'},
-    'interface': {'name': 'hw:interface', 'slug': 'hw-interface'},
-    'memory': {'name': 'hw:memory', 'slug': 'hw-memory'},
-    'motherboard': {'name': 'hw:motherboard', 'slug': 'hw-motherboard'},
-    'raid_card': {'name': 'hw:raid_card', 'slug': 'hw-raid-card'},
+    "cpu": {"name": "hw:cpu", "slug": "hw-cpu"},
+    "gpu": {"name": "hw:gpu", "slug": "hw-gpu"},
+    "disk": {"name": "hw:disk", "slug": "hw-disk"},
+    "interface": {"name": "hw:interface", "slug": "hw-interface"},
+    "memory": {"name": "hw:memory", "slug": "hw-memory"},
+    "motherboard": {"name": "hw:motherboard", "slug": "hw-motherboard"},
+    "raid_card": {"name": "hw:raid_card", "slug": "hw-raid-card"},
 }
 
 
-class Inventory():
+class Inventory:
     """
     Better Inventory items coming, see:
     - https://github.com/netbox-community/netbox/issues/3087
@@ -62,14 +64,12 @@ class Inventory():
     def create_netbox_tags(self):
         ret = []
         for key, tag in INVENTORY_TAG.items():
-            nb_tag = nb.extras.tags.get(
-                name=tag['name']
-            )
+            nb_tag = nb.extras.tags.get(name=tag["name"])
             if not nb_tag:
                 nb_tag = nb.extras.tags.create(
-                    name=tag['name'],
-                    slug=tag['slug'],
-                    comments=tag['name'],
+                    name=tag["name"],
+                    slug=tag["slug"],
+                    comments=tag["name"],
                 )
             ret.append(nb_tag)
         return ret
@@ -82,29 +82,28 @@ class Inventory():
             name=name,
         )
         if not manufacturer:
-            logging.info('Creating missing manufacturer {name}'.format(name=name))
+            logging.info("Creating missing manufacturer {name}".format(name=name))
             manufacturer = nb.dcim.manufacturers.create(
                 name=name,
-                slug=re.sub('[^A-Za-z0-9]+', '-', name).lower(),
+                slug=re.sub("[^A-Za-z0-9]+", "-", name).lower(),
             )
 
-            logging.info('Creating missing manufacturer {name}'.format(name=name))
+            logging.info("Creating missing manufacturer {name}".format(name=name))
 
         return manufacturer
 
     def get_netbox_inventory(self, device_id, tag):
         try:
-            items = nb.dcim.inventory_items.filter(
-                device_id=device_id,
-                tag=tag
-            )
+            items = nb.dcim.inventory_items.filter(device_id=device_id, tag=tag)
         except pynetbox.core.query.RequestError:
-            logging.info('Tag {tag} is missing, returning empty array.'.format(tag=tag))
+            logging.info("Tag {tag} is missing, returning empty array.".format(tag=tag))
             items = []
 
         return list(items)
 
-    def create_netbox_inventory_item(self, device_id, tags, vendor, name, serial, description):
+    def create_netbox_inventory_item(
+        self, device_id, tags, vendor, name, serial, description
+    ):
         manufacturer = self.find_or_create_manufacturer(vendor)
 
         _ = nb.dcim.inventory_items.create(
@@ -112,26 +111,25 @@ class Inventory():
             manufacturer=manufacturer.id,
             discovered=True,
             tags=tags,
-            name='{}'.format(name),
-            serial='{}'.format(serial),
-            description=description
+            name="{}".format(name),
+            serial="{}".format(serial),
+            description=description,
         )
 
-        logging.info('Creating inventory item {} {}/{} {} '.format(
-            vendor,
-            name,
-            serial,
-            description)
+        logging.info(
+            "Creating inventory item {} {}/{} {} ".format(
+                vendor, name, serial, description
+            )
         )
 
     def get_hw_motherboards(self):
         motherboards = []
 
         m = {}
-        m['serial'] = self.lshw.motherboard_serial
-        m['vendor'] = self.lshw.vendor
-        m['name'] = '{} {}'.format(self.lshw.vendor, self.lshw.motherboard)
-        m['description'] = '{} Motherboard'.format(self.lshw.motherboard)
+        m["serial"] = self.lshw.motherboard_serial
+        m["vendor"] = self.lshw.vendor
+        m["name"] = "{} {}".format(self.lshw.vendor, self.lshw.motherboard)
+        m["description"] = "{} Motherboard".format(self.lshw.motherboard)
 
         motherboards.append(m)
 
@@ -141,27 +139,29 @@ class Inventory():
 
         motherboards = self.get_hw_motherboards()
         nb_motherboards = self.get_netbox_inventory(
-            device_id=self.device_id,
-            tag=INVENTORY_TAG['motherboard']['slug'])
+            device_id=self.device_id, tag=INVENTORY_TAG["motherboard"]["slug"]
+        )
 
         for nb_motherboard in nb_motherboards:
-            if nb_motherboard.serial not in [x['serial'] for x in motherboards]:
-                logging.info('Deleting unknown motherboard {motherboard}/{serial}'.format(
+            if nb_motherboard.serial not in [x["serial"] for x in motherboards]:
+                logging.info(
+                    "Deleting unknown motherboard {motherboard}/{serial}".format(
                         motherboard=self.lshw.motherboard,
                         serial=nb_motherboard.serial,
-                ))
+                    )
+                )
                 nb_motherboard.delete()
 
         # create interfaces that are not in netbox
         for motherboard in motherboards:
-            if motherboard.get('serial') not in [x.serial for x in nb_motherboards]:
+            if motherboard.get("serial") not in [x.serial for x in nb_motherboards]:
                 self.create_netbox_inventory_item(
                     device_id=self.device_id,
-                    tags=[{'name': INVENTORY_TAG['motherboard']['name']}],
-                    vendor='{}'.format(motherboard.get('vendor', 'N/A')),
-                    serial='{}'.format(motherboard.get('serial', 'No SN')),
-                    name='{}'.format(motherboard.get('name')),
-                    description='{}'.format(motherboard.get('description'))
+                    tags=[{"name": INVENTORY_TAG["motherboard"]["name"]}],
+                    vendor="{}".format(motherboard.get("vendor", "N/A")),
+                    serial="{}".format(motherboard.get("serial", "No SN")),
+                    name="{}".format(motherboard.get("name")),
+                    description="{}".format(motherboard.get("description")),
                 )
 
     def create_netbox_interface(self, iface):
@@ -170,56 +170,57 @@ class Inventory():
             device=self.device_id,
             manufacturer=manufacturer.id,
             discovered=True,
-            tags=[{'name': INVENTORY_TAG['interface']['name']}],
-            name="{}".format(iface['product']),
-            serial='{}'.format(iface['serial']),
-            description='{} {}'.format(iface['description'], iface['name'])
+            tags=[{"name": INVENTORY_TAG["interface"]["name"]}],
+            name="{}".format(iface["product"]),
+            serial="{}".format(iface["serial"]),
+            description="{} {}".format(iface["description"], iface["name"]),
         )
 
     def do_netbox_interfaces(self):
         nb_interfaces = self.get_netbox_inventory(
-            device_id=self.device_id,
-            tag=INVENTORY_TAG['interface']['slug'])
+            device_id=self.device_id, tag=INVENTORY_TAG["interface"]["slug"]
+        )
         interfaces = self.lshw.interfaces
 
         # delete interfaces that are in netbox but not locally
         # use the serial_number has the comparison element
         for nb_interface in nb_interfaces:
-            if nb_interface.serial not in [x['serial'] for x in interfaces]:
-                logging.info('Deleting unknown interface {serial}'.format(
+            if nb_interface.serial not in [x["serial"] for x in interfaces]:
+                logging.info(
+                    "Deleting unknown interface {serial}".format(
                         serial=nb_interface.serial,
-                ))
+                    )
+                )
                 nb_interface.delete()
 
         # create interfaces that are not in netbox
         for iface in interfaces:
-            if iface.get('serial') not in [x.serial for x in nb_interfaces]:
+            if iface.get("serial") not in [x.serial for x in nb_interfaces]:
                 self.create_netbox_interface(iface)
 
     def create_netbox_cpus(self):
-        for cpu in self.lshw.get_hw_linux('cpu'):
+        for cpu in self.lshw.get_hw_linux("cpu"):
             manufacturer = self.find_or_create_manufacturer(cpu["vendor"])
             _ = nb.dcim.inventory_items.create(
                 device=self.device_id,
                 manufacturer=manufacturer.id,
                 discovered=True,
-                tags=[{'name': INVENTORY_TAG['cpu']['name']}],
-                name=cpu['product'],
-                description='CPU {}'.format(cpu['location']),
+                tags=[{"name": INVENTORY_TAG["cpu"]["name"]}],
+                name=cpu["product"],
+                description="CPU {}".format(cpu["location"]),
                 # asset_tag=cpu['location']
             )
 
-            logging.info('Creating CPU model {}'.format(cpu['product']))
+            logging.info("Creating CPU model {}".format(cpu["product"]))
 
     def do_netbox_cpus(self):
-        cpus = self.lshw.get_hw_linux('cpu')
+        cpus = self.lshw.get_hw_linux("cpu")
         nb_cpus = self.get_netbox_inventory(
             device_id=self.device_id,
-            tag=INVENTORY_TAG['cpu']['slug'],
+            tag=INVENTORY_TAG["cpu"]["slug"],
         )
 
-        if not len(nb_cpus) or \
-           len(nb_cpus) and len(cpus) != len(nb_cpus):
+        if not len(nb_cpus) or len(nb_cpus) and len(cpus) != len(nb_cpus):
             for x in nb_cpus:
                 x.delete()
 
@@ -227,13 +228,13 @@ class Inventory():
 
     def get_raid_cards(self, filter_cards=False):
         raid_class = None
-        if self.server.manufacturer in ('Dell', 'Huawei'):
-            if is_tool('omreport'):
+        if self.server.manufacturer in ("Dell", "Huawei"):
+            if is_tool("omreport"):
                 raid_class = OmreportRaid
-            if is_tool('storcli'):
+            if is_tool("storcli"):
                 raid_class = StorcliRaid
-        elif self.server.manufacturer in ('HP', 'HPE'):
-            if is_tool('ssacli'):
+        elif self.server.manufacturer in ("HP", "HPE"):
+            if is_tool("ssacli"):
                 raid_class = HPRaid
 
         if not raid_class:
@@ -241,19 +242,21 @@ class Inventory():
 
         self.raid = raid_class()
 
-        if filter_cards and config.expansion_as_device \
-                and self.server.own_expansion_slot():
+        if (
+            filter_cards
+            and config.expansion_as_device
+            and self.server.own_expansion_slot()
+        ):
             return [
-                c for c in self.raid.get_controllers()
+                c
+                for c in self.raid.get_controllers()
                 if c.is_external() is self.update_expansion
             ]
         else:
             return self.raid.get_controllers()
 
     def create_netbox_raid_card(self, raid_card):
-        manufacturer = self.find_or_create_manufacturer(
-            raid_card.get_manufacturer()
-        )
+        manufacturer = self.find_or_create_manufacturer(raid_card.get_manufacturer())
 
         name = raid_card.get_product_name()
         serial = raid_card.get_serial_number()
@@ -261,15 +264,17 @@ class Inventory():
             device=self.device_id,
             discovered=True,
             manufacturer=manufacturer.id if manufacturer else None,
-            tags=[{'name': INVENTORY_TAG['raid_card']['name']}],
-            name='{}'.format(name),
-            serial='{}'.format(serial),
-            description='RAID Card',
+            tags=[{"name": INVENTORY_TAG["raid_card"]["name"]}],
+            name="{}".format(name),
+            serial="{}".format(serial),
+            description="RAID Card",
         )
-        logging.info('Creating RAID Card {name} (SN: {serial})'.format(
+        logging.info(
+            "Creating RAID Card {name} (SN: {serial})".format(
                 name=name,
                 serial=serial,
-        ))
+            )
+        )
         return nb_raid_card
 
     def do_netbox_raid_cards(self):
@@ -284,8 +289,7 @@ class Inventory():
         """
 
         nb_raid_cards = self.get_netbox_inventory(
-            device_id=self.device_id,
-            tag=[INVENTORY_TAG['raid_card']['slug']]
+            device_id=self.device_id, tag=[INVENTORY_TAG["raid_card"]["slug"]]
         )
         raid_cards = self.get_raid_cards(filter_cards=True)
 
@@ -293,9 +297,11 @@ class Inventory():
         # use the serial_number has the comparison element
         for nb_raid_card in nb_raid_cards:
             if nb_raid_card.serial not in [x.get_serial_number() for x in raid_cards]:
-                logging.info('Deleting unknown locally RAID Card {serial}'.format(
+                logging.info(
+                    "Deleting unknown locally RAID Card {serial}".format(
                         serial=nb_raid_card.serial,
-                ))
+                    )
+                )
                 nb_raid_card.delete()
 
         # create card that are not in netbox
@@ -304,25 +310,32 @@ class Inventory():
                 self.create_netbox_raid_card(raid_card)
 
     def is_virtual_disk(self, disk, raid_devices):
-        disk_type = disk.get('type')
-        logicalname = disk.get('logicalname')
-        description = disk.get('description')
-        size = disk.get('size')
-        product = disk.get('product')
-        if logicalname in raid_devices or disk_type is None or product is None or description is None:
+        disk_type = disk.get("type")
+        logicalname = disk.get("logicalname")
+        description = disk.get("description")
+        size = disk.get("size")
+        product = disk.get("product")
+        if (
+            logicalname in raid_devices
+            or disk_type is None
+            or product is None
+            or description is None
+        ):
             return True
         non_raid_disks = [
-            'MR9361-8i',
+            "MR9361-8i",
         ]
 
-        if logicalname in raid_devices or \
-           product in non_raid_disks or \
-           'virtual' in product.lower() or \
-           'logical' in product.lower() or \
-           'volume' in description.lower() or \
-           'dvd-ram' in description.lower() or \
-           description == 'SCSI Enclosure' or \
-           (size is None and logicalname is None):
+        if (
+            logicalname in raid_devices
+            or product in non_raid_disks
+            or "virtual" in product.lower()
+            or "logical" in product.lower()
+            or "volume" in description.lower()
+            or "dvd-ram" in description.lower()
+            or description == "SCSI Enclosure"
+            or (size is None and logicalname is None)
+        ):
             return True
         return False
 
@@ -333,33 +346,33 @@ class Inventory():
             disks.extend(raid_card.get_physical_disks())
 
         raid_devices = [
-            d.get('custom_fields', {}).get('vd_device')
+            d.get("custom_fields", {}).get("vd_device")
             for d in disks
-            if d.get('custom_fields', {}).get('vd_device')
+            if d.get("custom_fields", {}).get("vd_device")
         ]
 
         for disk in self.lshw.get_hw_linux("storage"):
             if self.is_virtual_disk(disk, raid_devices):
                 continue
-            size = int(getattr(disk, "size", 0)) / 1073741824
+            size = round(int(disk.get("size", 0)) / 1073741824, 1)
             d = {
                 "name": "",
-                'Size': '{} GB'.format(size),
-                'logicalname': disk.get('logicalname'),
-                'description': disk.get('description'),
-                'SN': disk.get('serial'),
-                'Model': disk.get('product'),
-                'Type': disk.get('type'),
+                "Size": "{} GB".format(size),
+                "logicalname": disk.get("logicalname"),
+                "description": disk.get("description"),
+                "SN": disk.get("serial"),
+                "Model": disk.get("product"),
+                "Type": disk.get("type"),
             }
-            if disk.get('vendor'):
-                d['Vendor'] = disk['vendor']
+            if disk.get("vendor"):
+                d["Vendor"] = disk["vendor"]
             else:
-                d['Vendor'] = get_vendor(disk['product'])
+                d["Vendor"] = get_vendor(disk["product"])
             disks.append(d)
 
         # remove duplicate serials
         seen = set()
-        uniq = [x for x in disks if x['SN'] not in seen and not seen.add(x['SN'])]
+        uniq = [x for x in disks if x["SN"] not in seen and not seen.add(x["SN"])]
         return uniq
 
     def create_netbox_disk(self, disk):
@@ -367,53 +380,45 @@ class Inventory():
         if "Vendor" in disk:
             manufacturer = self.find_or_create_manufacturer(disk["Vendor"])
 
-        logicalname = disk.get('logicalname')
-        desc = disk.get('description')
-        name = '{} ({})'.format(disk['Model'], disk['Size'])
-        description = disk['Type']
-        sn = disk.get('SN', 'unknown')
+        name = "{} ({})".format(disk["Model"], disk["Size"])
+        description = disk["Type"]
+        sn = disk.get("SN", "unknown")
 
         parms = {
-            'device': self.device_id,
-            'discovered': True,
-            'tags': [{'name': INVENTORY_TAG['disk']['name']}],
-            'name': name,
-            'serial': sn,
-            'part_id': disk['Model'],
-            'description': description,
-            'manufacturer': getattr(manufacturer, "id", None),
+            "device": self.device_id,
+            "discovered": True,
+            "tags": [{"name": INVENTORY_TAG["disk"]["name"]}],
+            "name": name,
+            "serial": sn,
+            "part_id": disk["Model"],
+            "description": description,
+            "manufacturer": getattr(manufacturer, "id", None),
         }
         if config.process_virtual_drives:
-            parms['custom_fields'] = disk.get("custom_fields", {})
+            parms["custom_fields"] = disk.get("custom_fields", {})
 
         _ = nb.dcim.inventory_items.create(**parms)
 
-        logging.info('Creating Disk {model} {serial}'.format(
-            model=disk['Model'],
+        logging.info(
+            "Creating Disk {model} {serial}".format(
+                model=disk["Model"],
                 serial=sn,
-        ))
+            )
+        )
 
     def dump_disks_map(self, disks):
-        disk_map = [d['custom_fields'] for d in disks if 'custom_fields' in d]
+        disk_map = [d["custom_fields"] for d in disks if "custom_fields" in d]
         if config.dump_disks_map == "-":
             f = sys.stdout
         else:
             f = open(config.dump_disks_map, "w")
-        f.write(
-            json.dumps(
-                disk_map,
-                separators=(',', ':'),
-                indent=4,
-                sort_keys=True
-            )
-        )
+        f.write(json.dumps(disk_map, separators=(",", ":"), indent=4, sort_keys=True))
         if config.dump_disks_map != "-":
             f.close()
 
     def do_netbox_disks(self):
         nb_disks = self.get_netbox_inventory(
-            device_id=self.device_id,
-            tag=INVENTORY_TAG['disk']['slug']
+            device_id=self.device_id, tag=INVENTORY_TAG["disk"]["slug"]
         )
         disks = self.get_hw_disks()
         if config.dump_disks_map:
@@ -422,100 +427,108 @@ class Inventory():
             except Exception as e:
                 logging.error("Failed to dump disks map: {}".format(e))
                 logging.debug(traceback.format_exc())
-        disk_serials = [d['SN'] for d in disks if 'SN' in d]
+        disk_serials = [d["SN"] for d in disks if "SN" in d]
 
         # delete disks that are in netbox but not locally
         # use the serial_number has the comparison element
         for nb_disk in nb_disks:
-            if nb_disk.serial not in disk_serials or \
-                    config.force_disk_refresh:
-                logging.info('Deleting unknown locally Disk {serial}'.format(
+            if nb_disk.serial not in disk_serials or config.force_disk_refresh:
+                logging.info(
+                    "Deleting unknown locally Disk {serial}".format(
                         serial=nb_disk.serial,
-                ))
+                    )
+                )
                 nb_disk.delete()
 
         if config.force_disk_refresh:
             nb_disks = self.get_netbox_inventory(
-                device_id=self.device_id,
-                tag=INVENTORY_TAG['disk']['slug']
+                device_id=self.device_id, tag=INVENTORY_TAG["disk"]["slug"]
             )
 
         # create disks that are not in netbox
         for disk in disks:
-            if disk.get('SN') not in [d.serial for d in nb_disks]:
+            if disk.get("SN") not in [d.serial for d in nb_disks]:
                 self.create_netbox_disk(disk)
 
     def create_netbox_memory(self, memory):
-        manufacturer = self.find_or_create_manufacturer(memory['vendor'])
-        name = 'Slot {} ({}GB)'.format(memory['slot'], memory['size'])
+        manufacturer = self.find_or_create_manufacturer(memory["vendor"])
+        name = "Slot {} ({}GB)".format(memory["slot"], memory["size"])
         nb_memory = nb.dcim.inventory_items.create(
             device=self.device_id,
             discovered=True,
             manufacturer=manufacturer.id,
-            tags=[{'name': INVENTORY_TAG['memory']['name']}],
+            tags=[{"name": INVENTORY_TAG["memory"]["name"]}],
             name=name,
-            part_id=memory['product'],
-            serial=memory['serial'],
-            description=memory['description'],
+            part_id=memory["product"],
+            serial=memory["serial"],
+            description=memory["description"],
         )
 
-        logging.info('Creating Memory {location} {type} {size}GB'.format(
-            location=memory['slot'],
-            type=memory['product'],
-            size=memory['size'],
-        ))
+        logging.info(
+            "Creating Memory {location} {type} {size}GB".format(
+                location=memory["slot"],
+                type=memory["product"],
+                size=memory["size"],
+            )
+        )
 
         return nb_memory
 
     def do_netbox_memories(self):
         memories = self.lshw.memories
         nb_memories = self.get_netbox_inventory(
-            device_id=self.device_id,
-            tag=INVENTORY_TAG['memory']['slug']
+            device_id=self.device_id, tag=INVENTORY_TAG["memory"]["slug"]
         )
 
         for nb_memory in nb_memories:
-            if nb_memory.serial not in [x['serial'] for x in memories]:
-                logging.info('Deleting unknown locally Memory {serial}'.format(
+            if nb_memory.serial not in [x["serial"] for x in memories]:
+                logging.info(
+                    "Deleting unknown locally Memory {serial}".format(
                         serial=nb_memory.serial,
-                ))
+                    )
+                )
                 nb_memory.delete()
 
         for memory in memories:
-            if memory.get('serial') not in [x.serial for x in nb_memories]:
+            if memory.get("serial") not in [x.serial for x in nb_memories]:
                 self.create_netbox_memory(memory)
 
     def create_netbox_gpus(self, gpus):
         for gpu in gpus:
-            if 'product' in gpu and len(gpu['product']) > 50:
-                gpu['product'] = (gpu['product'][:48] + '..')
+            if "product" in gpu and len(gpu["product"]) > 50:
+                gpu["product"] = gpu["product"][:48] + ".."
 
             manufacturer = self.find_or_create_manufacturer(gpu["vendor"])
             _ = nb.dcim.inventory_items.create(
                 device=self.device_id,
                 manufacturer=manufacturer.id,
                 discovered=True,
-                tags=[{'name': INVENTORY_TAG['gpu']['name']}],
-                name=gpu['product'],
-                description=gpu['description'],
+                tags=[{"name": INVENTORY_TAG["gpu"]["name"]}],
+                name=gpu["product"],
+                description=gpu["description"],
             )
 
-            logging.info('Creating GPU model {}'.format(gpu['product']))
+            logging.info("Creating GPU model {}".format(gpu["product"]))
 
     def is_external_gpu(self, gpu):
-        is_3d_gpu = gpu['description'].startswith('3D')
-        return self.server.is_blade() and \
-            self.server.own_gpu_expansion_slot() and is_3d_gpu
+        is_3d_gpu = gpu["description"].startswith("3D")
+        return (
+            self.server.is_blade()
+            and self.server.own_gpu_expansion_slot()
+            and is_3d_gpu
+        )
 
     def do_netbox_gpus(self):
         gpus = []
         gpu_models = {}
-        for gpu in  self.lshw.get_hw_linux('gpu'):
+        for gpu in self.lshw.get_hw_linux("gpu"):
             # Filters GPU if an expansion bay is detected:
             # The internal (VGA) GPU only goes into the blade inventory,
             # the external (3D) GPU goes into the expansion blade.
-            if config.expansion_as_device and \
-                    self.update_expansion ^ self.is_external_gpu(gpu):
+            if (
+                config.expansion_as_device
+                and self.update_expansion ^ self.is_external_gpu(gpu)
+            ):
                 continue
             gpus.append(gpu)
             gpu_models.setdefault(gpu["product"], 0)
@@ -523,7 +536,7 @@ class Inventory():
 
         nb_gpus = self.get_netbox_inventory(
             device_id=self.device_id,
-            tag=INVENTORY_TAG['gpu']['slug'],
+            tag=INVENTORY_TAG["gpu"]["slug"],
         )
         nb_gpu_models = {}
         for gpu in nb_gpus:

netbox_agent/ipmi.py

@@ -4,7 +4,7 @@ import subprocess
 from netaddr import IPNetwork
 
 
-class IPMI():
+class IPMI:
     """
     Parse IPMI output
     ie:
@@ -37,9 +37,9 @@ class IPMI():
     """
 
     def __init__(self):
-        self.ret, self.output = subprocess.getstatusoutput('ipmitool lan print')
+        self.ret, self.output = subprocess.getstatusoutput("ipmitool lan print")
         if self.ret != 0:
-            logging.error('Cannot get ipmi info: {}'.format(self.output))
+            logging.error("Cannot get ipmi info: {}".format(self.output))
 
     def parse(self):
         _ipmi = {}
@@ -47,23 +47,31 @@ class IPMI():
             return _ipmi
 
         for line in self.output.splitlines():
-            key = line.split(':')[0].strip()
-            if key not in ['802.1q VLAN ID', 'IP Address', 'Subnet Mask', 'MAC Address']:
+            key = line.split(":")[0].strip()
+            if key not in [
+                "802.1q VLAN ID",
+                "IP Address",
+                "Subnet Mask",
+                "MAC Address",
+            ]:
                 continue
-            value = ':'.join(line.split(':')[1:]).strip()
+            value = ":".join(line.split(":")[1:]).strip()
             _ipmi[key] = value
 
         ret = {}
-        ret['name'] = 'IPMI'
+        ret["name"] = "IPMI"
         ret["mtu"] = 1500
-        ret['bonding'] = False
-        ret['mac'] = _ipmi['MAC Address']
-        ret['vlan'] = int(_ipmi['802.1q VLAN ID']) \
-            if _ipmi['802.1q VLAN ID'] != 'Disabled' else None
-        ip = _ipmi['IP Address']
-        netmask = _ipmi['Subnet Mask']
-        address = str(IPNetwork('{}/{}'.format(ip, netmask)))
+        ret["bonding"] = False
+        ret["mac"] = _ipmi["MAC Address"]
+        ret["vlan"] = (
+            int(_ipmi["802.1q VLAN ID"])
+            if _ipmi["802.1q VLAN ID"] != "Disabled"
+            else None
+        )
+        ip = _ipmi["IP Address"]
+        netmask = _ipmi["Subnet Mask"]
+        address = str(IPNetwork("{}/{}".format(ip, netmask)))
 
-        ret['ip'] = [address]
-        ret['ipmi'] = True
+        ret["ip"] = [address]
+        ret["ipmi"] = True
         return ret

netbox_agent/lldp.py

@@ -4,14 +4,14 @@ import subprocess
 from netbox_agent.misc import is_tool
 
 
-class LLDP():
+class LLDP:
     def __init__(self, output=None):
-        if not is_tool('lldpctl'):
-            logging.debug('lldpd package seems to be missing or daemon not running.')
+        if not is_tool("lldpctl"):
+            logging.debug("lldpd package seems to be missing or daemon not running.")
         if output:
             self.output = output
         else:
-            self.output = subprocess.getoutput('lldpctl -f keyvalue')
+            self.output = subprocess.getoutput("lldpctl -f keyvalue")
         self.data = self.parse()
 
     def parse(self):
@@ -19,7 +19,7 @@ class LLDP():
         vlans = {}
         vid = None
         for entry in self.output.splitlines():
-            if '=' not in entry:
+            if "=" not in entry:
                 continue
             path, value = entry.strip().split("=", 1)
             split_path = path.split(".")
@@ -34,38 +34,38 @@ class LLDP():
                 if not isinstance(current_dict.get(path_component), dict):
                     current_dict[path_component] = {}
                 current_dict = current_dict.get(path_component)
-                if 'vlan-id' in path:
+                if "vlan-id" in path:
                     vid = value
                     vlans[interface][value] = vlans[interface].get(vid, {})
-                elif path.endswith('vlan'):
-                    vid = value.replace('vlan-', '')
+                elif path.endswith("vlan"):
+                    vid = value.replace("vlan-", "")
                     vlans[interface][vid] = vlans[interface].get(vid, {})
-                elif 'pvid' in path:
-                    vlans[interface][vid]['pvid'] = True
-            if 'vlan' not in path:
+                elif "pvid" in path:
+                    vlans[interface][vid]["pvid"] = True
+            if "vlan" not in path:
                 current_dict[final] = value
         for interface, vlan in vlans.items():
-            output_dict['lldp'][interface]['vlan'] = vlan
+            output_dict["lldp"][interface]["vlan"] = vlan
         if not output_dict:
-            logging.debug('No LLDP output, please check your network config.')
+            logging.debug("No LLDP output, please check your network config.")
         return output_dict
 
     def get_switch_ip(self, interface):
         # lldp.eth0.chassis.mgmt-ip=100.66.7.222
-        if self.data['lldp'].get(interface) is None:
+        if self.data["lldp"].get(interface) is None:
             return None
-        return self.data['lldp'][interface]['chassis'].get('mgmt-ip')
+        return self.data["lldp"][interface]["chassis"].get("mgmt-ip")
 
     def get_switch_port(self, interface):
         # lldp.eth0.port.descr=GigabitEthernet1/0/1
-        if self.data['lldp'].get(interface) is None:
+        if self.data["lldp"].get(interface) is None:
             return None
-        if self.data['lldp'][interface]['port'].get('ifname'):
-            return self.data['lldp'][interface]['port']['ifname']
-        return self.data['lldp'][interface]['port']['descr']
+        if self.data["lldp"][interface]["port"].get("ifname"):
+            return self.data["lldp"][interface]["port"]["ifname"]
+        return self.data["lldp"][interface]["port"]["descr"]
 
     def get_switch_vlan(self, interface):
         # lldp.eth0.vlan.vlan-id=296
-        if self.data['lldp'].get(interface) is None:
+        if self.data["lldp"].get(interface) is None:
             return None
-        return self.data['lldp'][interface]['vlan']
+        return self.data["lldp"][interface]["vlan"]

netbox_agent/location.py

@@ -4,7 +4,7 @@ import importlib.machinery
 from netbox_agent.config import config
 
 
-class LocationBase():
+class LocationBase:
     """
     This class is used to guess the location in order to push the information
     in Netbox for a `Device`
@@ -27,15 +27,19 @@ class LocationBase():
         if self.driver_file:
             try:
                 # FIXME: Works with Python 3.3+, support older version?
-                loader = importlib.machinery.SourceFileLoader('driver_file', self.driver_file)
+                loader = importlib.machinery.SourceFileLoader(
+                    "driver_file", self.driver_file
+                )
                 self.driver = loader.load_module()
             except ImportError:
-                raise ImportError("Couldn't import {} as a module".format(self.driver_file))
+                raise ImportError(
+                    "Couldn't import {} as a module".format(self.driver_file)
+                )
         else:
             if self.driver:
                 try:
                     self.driver = importlib.import_module(
-                        'netbox_agent.drivers.{}'.format(self.driver)
+                        "netbox_agent.drivers.{}".format(self.driver)
                     )
                 except ImportError:
                     raise ImportError("Driver {} doesn't exists".format(self.driver))
@@ -43,19 +47,23 @@ class LocationBase():
     def get(self):
         if self.driver is None:
             return None
-        if not hasattr(self.driver, 'get'):
+        if not hasattr(self.driver, "get"):
             raise Exception(
-                "Your driver {} doesn't have a get() function, please fix it".format(self.driver)
+                "Your driver {} doesn't have a get() function, please fix it".format(
+                    self.driver
                 )
-        return getattr(self.driver, 'get')(self.driver_value, self.regex)
+            )
+        return getattr(self.driver, "get")(self.driver_value, self.regex)
 
 
 class Tenant(LocationBase):
     def __init__(self):
-        driver = config.tenant.driver.split(':')[0] if \
-            config.tenant.driver else None
-        driver_value = ':'.join(config.tenant.driver.split(':')[1:]) if \
-            config.tenant.driver else None
+        driver = config.tenant.driver.split(":")[0] if config.tenant.driver else None
+        driver_value = (
+            ":".join(config.tenant.driver.split(":")[1:])
+            if config.tenant.driver
+            else None
+        )
         driver_file = config.tenant.driver_file
         regex = config.tenant.regex
         super().__init__(driver, driver_value, driver_file, regex)
@@ -63,10 +71,16 @@ class Tenant(LocationBase):
 
 class Datacenter(LocationBase):
     def __init__(self):
-        driver = config.datacenter_location.driver.split(':')[0] if \
-            config.datacenter_location.driver else None
-        driver_value = ':'.join(config.datacenter_location.driver.split(':')[1:]) if \
-            config.datacenter_location.driver else None
+        driver = (
+            config.datacenter_location.driver.split(":")[0]
+            if config.datacenter_location.driver
+            else None
+        )
+        driver_value = (
+            ":".join(config.datacenter_location.driver.split(":")[1:])
+            if config.datacenter_location.driver
+            else None
+        )
         driver_file = config.datacenter_location.driver_file
         regex = config.datacenter_location.regex
         super().__init__(driver, driver_value, driver_file, regex)
@@ -74,10 +88,16 @@ class Datacenter(LocationBase):
 
 class Rack(LocationBase):
     def __init__(self):
-        driver = config.rack_location.driver.split(':')[0] if \
-            config.rack_location.driver else None
-        driver_value = ':'.join(config.rack_location.driver.split(':')[1:]) if \
-            config.rack_location.driver else None
+        driver = (
+            config.rack_location.driver.split(":")[0]
+            if config.rack_location.driver
+            else None
+        )
+        driver_value = (
+            ":".join(config.rack_location.driver.split(":")[1:])
+            if config.rack_location.driver
+            else None
+        )
         driver_file = config.rack_location.driver_file
         regex = config.rack_location.regex
         super().__init__(driver, driver_value, driver_file, regex)
@@ -85,10 +105,16 @@ class Rack(LocationBase):
 
 class Slot(LocationBase):
     def __init__(self):
-        driver = config.slot_location.driver.split(':')[0] if \
-            config.slot_location.driver else None
-        driver_value = ':'.join(config.slot_location.driver.split(':')[1:]) if \
-            config.slot_location.driver else None
+        driver = (
+            config.slot_location.driver.split(":")[0]
+            if config.slot_location.driver
+            else None
+        )
+        driver_value = (
+            ":".join(config.slot_location.driver.split(":")[1:])
+            if config.slot_location.driver
+            else None
+        )
         driver_file = config.slot_location.driver_file
         regex = config.slot_location.regex
         super().__init__(driver, driver_value, driver_file, regex)

netbox_agent/logging.py

@@ -3,7 +3,7 @@ import logging
 from netbox_agent.config import config
 
 logger = logging.getLogger()
-if config.log_level.lower() == 'debug':
+if config.log_level.lower() == "debug":
     logger.setLevel(logging.DEBUG)
 else:
     logger.setLevel(logging.INFO)

netbox_agent/lshw.py

@@ -1,19 +1,18 @@
-from netbox_agent.misc import is_tool
-import subprocess
-import logging
 import json
+import logging
+import subprocess
 import sys
 
+from netbox_agent.misc import is_tool
 
-class LSHW():
+
+class LSHW:
     def __init__(self):
-        if not is_tool('lshw'):
-            logging.error('lshw does not seem to be installed')
+        if not is_tool("lshw"):
+            logging.error("lshw does not seem to be installed")
             sys.exit(1)
 
-        data = subprocess.getoutput(
-            'lshw -quiet -json'
-        )
+        data = subprocess.getoutput("lshw -quiet -json")
         json_data = json.loads(data)
         # Starting from version 02.18, `lshw -json` wraps its result in a list
         # rather than returning directly a dictionary
@@ -63,9 +62,9 @@ class LSHW():
             return self.gpus
         if hwclass == "network":
             return self.interfaces
-        if hwclass == 'storage':
+        if hwclass == "storage":
             return self.disks
-        if hwclass == 'memory':
+        if hwclass == "memory":
             return self.memories
 
     def find_network(self, obj):
@@ -77,26 +76,29 @@ class LSHW():
             # newer versions of lshw can return a list of names, see issue #227
             if not isinstance(i["name"], list):
                 if i["name"].startswith("unknown"):
-                    unkn_intfs.push(i)
+                    unkn_intfs.append(i)
             else:
                 for j in i["name"]:
                     if j.startswith("unknown"):
-                        unkn_intfs.push(j)
+                        unkn_intfs.append(j)
 
         unkn_name = "unknown{}".format(len(unkn_intfs))
-        self.interfaces.append({
+        self.interfaces.append(
+            {
                 "name": obj.get("logicalname", unkn_name),
                 "macaddress": obj.get("serial", ""),
                 "serial": obj.get("serial", ""),
-            "product": obj["product"],
-            "vendor": obj["vendor"],
-            "description": obj["description"],
-        })
+                "product": obj.get("product", "Unknown NIC"),
+                "vendor": obj.get("vendor", "Unknown"),
+                "description": obj.get("description", ""),
+            }
+        )
 
     def find_storage(self, obj):
         if "children" in obj:
             for device in obj["children"]:
-                self.disks.append({
+                self.disks.append(
+                    {
                         "logicalname": device.get("logicalname"),
                         "product": device.get("product"),
                         "serial": device.get("serial"),
@@ -104,42 +106,45 @@ class LSHW():
                         "size": device.get("size"),
                         "description": device.get("description"),
                         "type": device.get("description"),
-                })
+                    }
+                )
         elif "nvme" in obj["configuration"]["driver"]:
-            if not is_tool('nvme'):
-                logging.error('nvme-cli >= 1.0 does not seem to be installed')
+            if not is_tool("nvme"):
+                logging.error("nvme-cli >= 1.0 does not seem to be installed")
                 return
             try:
                 nvme = json.loads(
                     subprocess.check_output(
-                        ["nvme", '-list', '-o', 'json'],
-                        encoding='utf8')
+                        ["nvme", "-list", "-o", "json"], encoding="utf8"
+                    )
                 )
                 for device in nvme["Devices"]:
                     d = {
-                        'logicalname': device["DevicePath"],
-                        'product': device["ModelNumber"],
-                        'serial': device["SerialNumber"],
+                        "logicalname": device["DevicePath"],
+                        "product": device["ModelNumber"],
+                        "serial": device["SerialNumber"],
                         "version": device["Firmware"],
-                        'description': "NVME",
-                        'type': "NVME",
+                        "description": "NVME",
+                        "type": "NVME",
                     }
                     if "UsedSize" in device:
-                        d['size'] = device["UsedSize"]
+                        d["size"] = device["UsedSize"]
                     if "UsedBytes" in device:
-                        d['size'] = device["UsedBytes"]
+                        d["size"] = device["UsedBytes"]
                     self.disks.append(d)
             except Exception:
                 pass
 
     def find_cpus(self, obj):
         if "product" in obj:
-            self.cpus.append({
-                "product": obj["product"],
-                "vendor": obj["vendor"],
-                "description": obj["description"],
-                "location": obj["slot"],
-            })
+            self.cpus.append(
+                {
+                    "product": obj.get("product", "Unknown CPU"),
+                    "vendor": obj.get("vendor", "Unknown vendor"),
+                    "description": obj.get("description", ""),
+                    "location": obj.get("slot", ""),
+                }
+            )
 
     def find_memories(self, obj):
         if "children" not in obj:
@@ -150,23 +155,26 @@ class LSHW():
             if "empty" in dimm["description"]:
                 continue
 
-            self.memories.append({
+            self.memories.append(
+                {
                     "slot": dimm.get("slot"),
                     "description": dimm.get("description"),
                     "id": dimm.get("id"),
-                "serial": dimm.get("serial", 'N/A'),
-                "vendor": dimm.get("vendor", 'N/A'),
-                "product": dimm.get("product", 'N/A'),
-                "size": dimm.get("size", 0) / 2 ** 20 / 1024,
-            })
+                    "serial": dimm.get("serial", "N/A"),
+                    "vendor": dimm.get("vendor", "N/A"),
+                    "product": dimm.get("product", "N/A"),
+                    "size": dimm.get("size", 0) / 2**20 / 1024,
+                }
+            )
 
     def find_gpus(self, obj):
         if "product" in obj:
-            self.gpus.append({
-                "product": obj["product"],
-                "vendor": obj["vendor"],
-                "description": obj["description"],
-            })
+            infos = {
+                "product": obj.get("product", "Unknown GPU"),
+                "vendor": obj.get("vendor", "Unknown"),
+                "description": obj.get("description", ""),
+            }
+            self.gpus.append(infos)
 
     def walk_bridge(self, obj):
         if "children" not in obj:

netbox_agent/misc.py

@@ -1,29 +1,27 @@
-from netbox_agent.config import netbox_instance as nb
-from slugify import slugify
-from shutil import which
-import subprocess
-import socket
 import re
+import socket
+import subprocess
+from shutil import which
+
+from slugify import slugify
+
+from netbox_agent.config import netbox_instance as nb
 
 
 def is_tool(name):
-    '''Check whether `name` is on PATH and marked as executable.'''
+    """Check whether `name` is on PATH and marked as executable."""
     return which(name) is not None
 
 
 def get_device_role(role):
-    device_role = nb.dcim.device_roles.get(
-        name=role
-    )
+    device_role = nb.dcim.device_roles.get(name=role)
     if device_role is None:
         raise Exception('DeviceRole "{}" does not exist, please create it'.format(role))
     return device_role
 
 
 def get_device_type(type):
-    device_type = nb.dcim.device_types.get(
-        model=type
-    )
+    device_type = nb.dcim.device_types.get(model=type)
     if device_type is None:
         raise Exception('DeviceType "{}" does not exist, please create it'.format(type))
     return device_type
@@ -35,9 +33,11 @@ def get_device_platform(device_platform):
             # Python 3.8+ moved linux_distribution() to distro
             try:
                 import distro
+
                 linux_distribution = " ".join(distro.linux_distribution())
             except ImportError:
                 import platform
+
                 linux_distribution = " ".join(platform.linux_distribution())
 
             if not linux_distribution:
@@ -54,24 +54,25 @@ def get_device_platform(device_platform):
         )
     return device_platform
 
+
 def get_vendor(name):
     vendors = {
-        'PERC': 'Dell',
-        'SANDISK': 'SanDisk',
-        'DELL': 'Dell',
-        'ST': 'Seagate',
-        'CRUCIAL': 'Crucial',
-        'MICRON': 'Micron',
-        'INTEL': 'Intel',
-        'SAMSUNG': 'Samsung',
-        'EH0': 'HP',
-        'HGST': 'HGST',
-        'HUH': 'HGST',
-        'MB': 'Toshiba',
-        'MC': 'Toshiba',
-        'MD': 'Toshiba',
-        'MG': 'Toshiba',
-        'WD': 'WDC'
+        "PERC": "Dell",
+        "SANDISK": "SanDisk",
+        "DELL": "Dell",
+        "ST": "Seagate",
+        "CRUCIAL": "Crucial",
+        "MICRON": "Micron",
+        "INTEL": "Intel",
+        "SAMSUNG": "Samsung",
+        "EH0": "HP",
+        "HGST": "HGST",
+        "HUH": "HGST",
+        "MB": "Toshiba",
+        "MC": "Toshiba",
+        "MD": "Toshiba",
+        "MG": "Toshiba",
+        "WD": "WDC",
     }
     for key, value in vendors.items():
         if name.upper().startswith(key):
@@ -81,16 +82,14 @@ def get_vendor(name):
 
 def get_hostname(config):
     if config.hostname_cmd is None:
-        return '{}'.format(socket.gethostname())
+        return "{}".format(socket.gethostname())
     return subprocess.getoutput(config.hostname_cmd)
 
 
 def create_netbox_tags(tags):
     ret = []
     for tag in tags:
-        nb_tag = nb.extras.tags.get(
-            name=tag
-        )
+        nb_tag = nb.extras.tags.get(name=tag)
         if not nb_tag:
             nb_tag = nb.extras.tags.create(
                 name=tag,
@@ -102,15 +101,13 @@ def create_netbox_tags(tags):
 
 def get_mount_points():
     mount_points = {}
-    output = subprocess.getoutput('mount')
+    output = subprocess.getoutput("mount")
     for r in output.split("\n"):
         if not r.startswith("/dev/"):
             continue
         mount_info = r.split()
         device = mount_info[0]
-        device = re.sub(r'\d+$', '', device)
+        device = re.sub(r"\d+$", "", device)
         mp = mount_info[2]
         mount_points.setdefault(device, []).append(mp)
     return mount_points
-
-

netbox_agent/network.py

@@ -1,7 +1,7 @@
 import logging
 import os
 import re
-from itertools import chain
+from itertools import chain, islice
 
 import netifaces
 from netaddr import IPAddress
@@ -26,42 +26,45 @@ class Network(object):
         self.dcim_choices = {}
         dcim_c = nb.dcim.interfaces.choices()
         for _choice_type in dcim_c:
-            key = 'interface:{}'.format(_choice_type)
+            key = "interface:{}".format(_choice_type)
             self.dcim_choices[key] = {}
             for choice in dcim_c[_choice_type]:
-                self.dcim_choices[key][choice['display_name']] = choice['value']
+                self.dcim_choices[key][choice["display_name"]] = choice["value"]
 
         self.ipam_choices = {}
         ipam_c = nb.ipam.ip_addresses.choices()
         for _choice_type in ipam_c:
-            key = 'ip-address:{}'.format(_choice_type)
+            key = "ip-address:{}".format(_choice_type)
             self.ipam_choices[key] = {}
             for choice in ipam_c[_choice_type]:
-                self.ipam_choices[key][choice['display_name']] = choice['value']
+                self.ipam_choices[key][choice["display_name"]] = choice["value"]
 
     def get_network_type():
         return NotImplementedError
 
     def scan(self):
         nics = []
-        for interface in os.listdir('/sys/class/net/'):
+        for interface in os.listdir("/sys/class/net/"):
             # ignore if it's not a link (ie: bonding_masters etc)
-            if not os.path.islink('/sys/class/net/{}'.format(interface)):
+            if not os.path.islink("/sys/class/net/{}".format(interface)):
                 continue
 
-            if config.network.ignore_interfaces and \
-               re.match(config.network.ignore_interfaces, interface):
-                logging.debug('Ignore interface {interface}'.format(interface=interface))
+            if config.network.ignore_interfaces and re.match(
+                config.network.ignore_interfaces, interface
+            ):
+                logging.debug(
+                    "Ignore interface {interface}".format(interface=interface)
+                )
                 continue
 
             ip_addr = netifaces.ifaddresses(interface).get(netifaces.AF_INET, [])
             ip6_addr = netifaces.ifaddresses(interface).get(netifaces.AF_INET6, [])
             if config.network.ignore_ips:
                 for i, ip in enumerate(ip_addr):
-                    if re.match(config.network.ignore_ips, ip['addr']):
+                    if re.match(config.network.ignore_ips, ip["addr"]):
                         ip_addr.pop(i)
                 for i, ip in enumerate(ip6_addr):
-                    if re.match(config.network.ignore_ips, ip['addr']):
+                    if re.match(config.network.ignore_ips, ip["addr"]):
                         ip6_addr.pop(i)
 
             # netifaces returns a ipv6 netmask that netaddr does not understand.
@@ -80,63 +83,72 @@ class Network(object):
             #   }
             #
             for addr in ip6_addr:
-                addr["addr"] = addr["addr"].replace('%{}'.format(interface), '')
-                addr["netmask"] = addr["netmask"].split('/')[0]
+                addr["addr"] = addr["addr"].replace("%{}".format(interface), "")
+                addr["mask"] = addr["mask"].split("/")[0]
                 ip_addr.append(addr)
 
-            mac = open('/sys/class/net/{}/address'.format(interface), 'r').read().strip()
-            mtu = int(open('/sys/class/net/{}/mtu'.format(interface), 'r').read().strip())
+            mac = (
+                open("/sys/class/net/{}/address".format(interface), "r").read().strip()
+            )
+            mtu = int(
+                open("/sys/class/net/{}/mtu".format(interface), "r").read().strip()
+            )
             vlan = None
-            if len(interface.split('.')) > 1:
-                vlan = int(interface.split('.')[1])
+            if len(interface.split(".")) > 1:
+                vlan = int(interface.split(".")[1])
 
             bonding = False
             bonding_slaves = []
-            if os.path.isdir('/sys/class/net/{}/bonding'.format(interface)):
+            if os.path.isdir("/sys/class/net/{}/bonding".format(interface)):
                 bonding = True
-                bonding_slaves = open(
-                    '/sys/class/net/{}/bonding/slaves'.format(interface)
-                ).read().split()
-
-            # Tun and TAP support
-            virtual = os.path.isfile(
-                '/sys/class/net/{}/tun_flags'.format(interface)
+                bonding_slaves = (
+                    open("/sys/class/net/{}/bonding/slaves".format(interface))
+                    .read()
+                    .split()
                 )
 
+            # Tun and TAP support
+            virtual = os.path.isfile("/sys/class/net/{}/tun_flags".format(interface))
+
             nic = {
-                'name': interface,
-                'mac': mac if mac != '00:00:00:00:00:00' else None,
-                'ip': [
-                    '{}/{}'.format(
-                        x['addr'],
-                        IPAddress(x['netmask']).netmask_bits()
-                    ) for x in ip_addr
-                ] if ip_addr else None,  # FIXME: handle IPv6 addresses
-                'ethtool': Ethtool(interface).parse(),
-                'virtual': virtual,
-                'vlan': vlan,
-                'mtu': mtu,
-                'bonding': bonding,
-                'bonding_slaves': bonding_slaves,
+                "name": interface,
+                "mac": mac if mac != "00:00:00:00:00:00" else None,
+                "ip": (
+                    [
+                        "{}/{}".format(x["addr"], IPAddress(x["mask"]).netmask_bits())
+                        for x in ip_addr
+                    ]
+                    if ip_addr
+                    else None
+                ),  # FIXME: handle IPv6 addresses
+                "ethtool": Ethtool(interface).parse(),
+                "virtual": virtual,
+                "vlan": vlan,
+                "mtu": mtu,
+                "bonding": bonding,
+                "bonding_slaves": bonding_slaves,
             }
             nics.append(nic)
         return nics
 
     def _set_bonding_interfaces(self):
-        bonding_nics = (x for x in self.nics if x['bonding'])
+        bonding_nics = (x for x in self.nics if x["bonding"])
         for nic in bonding_nics:
             bond_int = self.get_netbox_network_card(nic)
-            logging.debug('Setting slave interface for {name}'.format(
-                name=bond_int.name
-            ))
+            logging.debug(
+                "Setting slave interface for {name}".format(name=bond_int.name)
+            )
             for slave_int in (
                 self.get_netbox_network_card(slave_nic)
                 for slave_nic in self.nics
-                    if slave_nic['name'] in nic['bonding_slaves']):
+                if slave_nic["name"] in nic["bonding_slaves"]
+            ):
                 if slave_int.lag is None or slave_int.lag.id != bond_int.id:
-                    logging.debug('Settting interface {name} as slave of {master}'.format(
+                    logging.debug(
+                        "Settting interface {name} as slave of {master}".format(
                             name=slave_int.name, master=bond_int.name
-                    ))
+                        )
+                    )
                     slave_int.lag = bond_int
                     slave_int.save()
         else:
@@ -147,55 +159,50 @@ class Network(object):
         return self.nics
 
     def get_netbox_network_card(self, nic):
-        if nic['mac'] is None:
+        if nic["mac"] is None:
             interface = self.nb_net.interfaces.get(
-                name=nic['name'],
-                **self.custom_arg_id
+                name=nic["name"], **self.custom_arg_id
             )
         else:
             interface = self.nb_net.interfaces.get(
-                mac_address=nic['mac'],
-                name=nic['name'],
-                **self.custom_arg_id
+                mac_address=nic["mac"], name=nic["name"], **self.custom_arg_id
             )
         return interface
 
     def get_netbox_network_cards(self):
-        return self.nb_net.interfaces.filter(
-            **self.custom_arg_id
-        )
+        return self.nb_net.interfaces.filter(**self.custom_arg_id)
 
     def get_netbox_type_for_nic(self, nic):
-        if self.get_network_type() == 'virtual':
-            return self.dcim_choices['interface:type']['Virtual']
+        if self.get_network_type() == "virtual":
+            return self.dcim_choices["interface:type"]["Virtual"]
 
-        if nic.get('bonding'):
-            return self.dcim_choices['interface:type']['Link Aggregation Group (LAG)']
+        if nic.get("bonding"):
+            return self.dcim_choices["interface:type"]["Link Aggregation Group (LAG)"]
 
-        if nic.get('bonding'):
-            return self.dcim_choices['interface:type']['Link Aggregation Group (LAG)']
+        if nic.get("bonding"):
+            return self.dcim_choices["interface:type"]["Link Aggregation Group (LAG)"]
 
-        if nic.get('virtual'):
-            return self.dcim_choices['interface:type']['Virtual']
+        if nic.get("virtual"):
+            return self.dcim_choices["interface:type"]["Virtual"]
 
-        if nic.get('ethtool') is None:
-            return self.dcim_choices['interface:type']['Other']
+        if nic.get("ethtool") is None:
+            return self.dcim_choices["interface:type"]["Other"]
 
-        if nic['ethtool']['speed'] == '10000Mb/s':
-            if nic['ethtool']['port'] in ('FIBRE', 'Direct Attach Copper'):
-                return self.dcim_choices['interface:type']['SFP+ (10GE)']
-            return self.dcim_choices['interface:type']['10GBASE-T (10GE)']
+        if nic["ethtool"]["speed"] == "10000Mb/s":
+            if nic["ethtool"]["port"] in ("FIBRE", "Direct Attach Copper"):
+                return self.dcim_choices["interface:type"]["SFP+ (10GE)"]
+            return self.dcim_choices["interface:type"]["10GBASE-T (10GE)"]
 
-        elif nic['ethtool']['speed'] == '25000Mb/s':
-            if nic['ethtool']['port'] in ('FIBRE', 'Direct Attach Copper'):
-                return self.dcim_choices['interface:type']['SFP28 (25GE)']
+        elif nic["ethtool"]["speed"] == "25000Mb/s":
+            if nic["ethtool"]["port"] in ("FIBRE", "Direct Attach Copper"):
+                return self.dcim_choices["interface:type"]["SFP28 (25GE)"]
 
-        elif nic['ethtool']['speed'] == '1000Mb/s':
-            if nic['ethtool']['port'] in ('FIBRE', 'Direct Attach Copper'):
-                return self.dcim_choices['interface:type']['SFP (1GE)']
-            return self.dcim_choices['interface:type']['1000BASE-T (1GE)']
+        elif nic["ethtool"]["speed"] == "1000Mb/s":
+            if nic["ethtool"]["port"] in ("FIBRE", "Direct Attach Copper"):
+                return self.dcim_choices["interface:type"]["SFP (1GE)"]
+            return self.dcim_choices["interface:type"]["1000BASE-T (1GE)"]
 
-        return self.dcim_choices['interface:type']['Other']
+        return self.dcim_choices["interface:type"]["Other"]
 
     def get_or_create_vlan(self, vlan_id):
         # FIXME: we may need to specify the datacenter
@@ -205,25 +212,35 @@ class Network(object):
         )
         if vlan is None:
             vlan = nb.ipam.vlans.create(
-                name='VLAN {}'.format(vlan_id),
+                name="VLAN {}".format(vlan_id),
                 vid=vlan_id,
             )
         return vlan
 
     def reset_vlan_on_interface(self, nic, interface):
         update = False
-        vlan_id = nic['vlan']
-        lldp_vlan = self.lldp.get_switch_vlan(nic['name']) if config.network.lldp else None
+        vlan_id = nic["vlan"]
+        lldp_vlan = (
+            self.lldp.get_switch_vlan(nic["name"])
+            if config.network.lldp and isinstance(self, ServerNetwork)
+            else None
+        )
         # For strange reason, we need to get the object from scratch
         # The object returned by pynetbox's save isn't always working (since pynetbox 6)
         interface = self.nb_net.interfaces.get(id=interface.id)
 
         # Handle the case were the local interface isn't an interface vlan as reported by Netbox
         # and that LLDP doesn't report a vlan-id
-        if vlan_id is None and lldp_vlan is None and \
-           (interface.mode is not None or len(interface.tagged_vlans) > 0):
-            logging.info('Interface {interface} is not tagged, reseting mode'.format(
-                interface=interface))
+        if (
+            vlan_id is None
+            and lldp_vlan is None
+            and (interface.mode is not None or len(interface.tagged_vlans) > 0)
+        ):
+            logging.info(
+                "Interface {interface} is not tagged, reseting mode".format(
+                    interface=interface
+                )
+            )
             update = True
             interface.mode = None
             interface.tagged_vlans = []
@@ -232,76 +249,92 @@ class Network(object):
         # if mode is either not set or not correctly configured or vlan are not
         # correctly configured, we reset the vlan
         elif vlan_id and (
-                interface.mode is None or
-                type(interface.mode) is not int and (
-                    hasattr(interface.mode, 'value') and
-                    interface.mode.value == self.dcim_choices['interface:mode']['Access'] or
-                    len(interface.tagged_vlans) != 1 or
-                    int(interface.tagged_vlans[0].vid) != int(vlan_id))):
-            logging.info('Resetting tagged VLAN(s) on interface {interface}'.format(
-                interface=interface))
+            interface.mode is None
+            or not isinstance(interface.mode, int)
+            and (
+                hasattr(interface.mode, "value")
+                and interface.mode.value
+                == self.dcim_choices["interface:mode"]["Access"]
+                or len(interface.tagged_vlans) != 1
+                or int(interface.tagged_vlans[0].vid) != int(vlan_id)
+            )
+        ):
+            logging.info(
+                "Resetting tagged VLAN(s) on interface {interface}".format(
+                    interface=interface
+                )
+            )
             update = True
             nb_vlan = self.get_or_create_vlan(vlan_id)
-            interface.mode = self.dcim_choices['interface:mode']['Tagged']
+            interface.mode = self.dcim_choices["interface:mode"]["Tagged"]
             interface.tagged_vlans = [nb_vlan] if nb_vlan else []
             interface.untagged_vlan = None
         # Finally if LLDP reports a vlan-id with the pvid attribute
         elif lldp_vlan:
-            pvid_vlan = [key for (key, value) in lldp_vlan.items() if value['pvid']]
+            pvid_vlan = [key for (key, value) in lldp_vlan.items() if value["pvid"]]
             if len(pvid_vlan) > 0 and (
-                    interface.mode is None or
-                    interface.mode.value != self.dcim_choices['interface:mode']['Access'] or
-                    interface.untagged_vlan is None or
-                    interface.untagged_vlan.vid != int(pvid_vlan[0])):
-                logging.info('Resetting access VLAN on interface {interface}'.format(
-                    interface=interface))
+                interface.mode is None
+                or interface.mode.value != self.dcim_choices["interface:mode"]["Access"]
+                or interface.untagged_vlan is None
+                or interface.untagged_vlan.vid != int(pvid_vlan[0])
+            ):
+                logging.info(
+                    "Resetting access VLAN on interface {interface}".format(
+                        interface=interface
+                    )
+                )
                 update = True
                 nb_vlan = self.get_or_create_vlan(pvid_vlan[0])
-                interface.mode = self.dcim_choices['interface:mode']['Access']
+                interface.mode = self.dcim_choices["interface:mode"]["Access"]
                 interface.untagged_vlan = nb_vlan.id
         return update, interface
 
     def create_netbox_nic(self, nic, mgmt=False):
         # TODO: add Optic Vendor, PN and Serial
         nic_type = self.get_netbox_type_for_nic(nic)
-        logging.info('Creating NIC {name} ({mac}) on {device}'.format(
-            name=nic['name'], mac=nic['mac'], device=self.device.name))
+        logging.info(
+            "Creating NIC {name} ({mac}) on {device}".format(
+                name=nic["name"], mac=nic["mac"], device=self.device.name
+            )
+        )
 
         nb_vlan = None
 
         params = dict(self.custom_arg)
-        params.update({
-            'name': nic['name'],
-            'type': nic_type,
-            'mgmt_only': mgmt,
-        })
-        if nic['mac']:
-            params['mac_address'] = nic['mac']
+        params.update(
+            {
+                "name": nic["name"],
+                "type": nic_type,
+                "mgmt_only": mgmt,
+            }
+        )
+        if nic["mac"]:
+            params["mac_address"] = nic["mac"]
 
-        if nic['mtu']:
-            params['mtu'] = nic['mtu']
+        if nic["mtu"]:
+            params["mtu"] = nic["mtu"]
 
         interface = self.nb_net.interfaces.create(**params)
 
-        if nic['vlan']:
-            nb_vlan = self.get_or_create_vlan(nic['vlan'])
-            interface.mode = self.dcim_choices['interface:mode']['Tagged']
+        if nic["vlan"]:
+            nb_vlan = self.get_or_create_vlan(nic["vlan"])
+            interface.mode = self.dcim_choices["interface:mode"]["Tagged"]
             interface.tagged_vlans = [nb_vlan.id]
             interface.save()
-        elif config.network.lldp and self.lldp.get_switch_vlan(nic['name']) is not None:
+        elif config.network.lldp and self.lldp.get_switch_vlan(nic["name"]) is not None:
             # if lldp reports a vlan on an interface, tag the interface in access and set the vlan
             # report only the interface which has `pvid=yes` (ie: lldp.eth3.vlan.pvid=yes)
             # if pvid is not present, it'll be processed as a vlan tagged interface
-            vlans = self.lldp.get_switch_vlan(nic['name'])
+            vlans = self.lldp.get_switch_vlan(nic["name"])
             for vid, vlan_infos in vlans.items():
                 nb_vlan = self.get_or_create_vlan(vid)
-                if vlan_infos.get('vid'):
-                    interface.mode = self.dcim_choices['interface:mode']['Access']
+                if vlan_infos.get("vid"):
+                    interface.mode = self.dcim_choices["interface:mode"]["Access"]
                     interface.untagged_vlan = nb_vlan.id
             interface.save()
 
         # cable the interface
-        if config.network.lldp:
+        if config.network.lldp and isinstance(self, ServerNetwork):
             switch_ip = self.lldp.get_switch_ip(interface.name)
             switch_interface = self.lldp.get_switch_port(interface.name)
 
@@ -314,7 +347,7 @@ class Network(object):
         return interface
 
     def create_or_update_netbox_ip_on_interface(self, ip, interface):
-        '''
+        """
         Two behaviors:
         - Anycast IP
         * If IP exists and is in Anycast, create a new Anycast one
@@ -325,69 +358,78 @@ class Network(object):
         * If IP doesn't exist, create it
         * If IP exists and isn't assigned, take it
         * If IP exists and interface is wrong, change interface
-        '''
+        """
         netbox_ips = nb.ipam.ip_addresses.filter(
             address=ip,
         )
         if not netbox_ips:
-            logging.info('Create new IP {ip} on {interface}'.format(
-                ip=ip, interface=interface))
+            logging.info(
+                "Create new IP {ip} on {interface}".format(ip=ip, interface=interface)
+            )
             query_params = {
-                'address': ip,
-                'status': "active",
-                'assigned_object_type': self.assigned_object_type,
-                'assigned_object_id': interface.id
+                "address": ip,
+                "status": "active",
+                "assigned_object_type": self.assigned_object_type,
+                "assigned_object_id": interface.id,
             }
 
-            netbox_ip = nb.ipam.ip_addresses.create(
-                **query_params
-            )
+            netbox_ip = nb.ipam.ip_addresses.create(**query_params)
             return netbox_ip
 
         netbox_ip = list(netbox_ips)[0]
         # If IP exists in anycast
-        if netbox_ip.role and netbox_ip.role.label == 'Anycast':
-            logging.debug('IP {} is Anycast..'.format(ip))
+        if netbox_ip.role and netbox_ip.role.label == "Anycast":
+            logging.debug("IP {} is Anycast..".format(ip))
             unassigned_anycast_ip = [x for x in netbox_ips if x.interface is None]
-            assigned_anycast_ip = [x for x in netbox_ips if
-                                   x.interface and x.interface.id == interface.id]
+            assigned_anycast_ip = [
+                x for x in netbox_ips if x.interface and x.interface.id == interface.id
+            ]
             # use the first available anycast ip
             if len(unassigned_anycast_ip):
-                logging.info('Assigning existing Anycast IP {} to interface'.format(ip))
+                logging.info("Assigning existing Anycast IP {} to interface".format(ip))
                 netbox_ip = unassigned_anycast_ip[0]
                 netbox_ip.interface = interface
                 netbox_ip.save()
             # or if everything is assigned to other servers
             elif not len(assigned_anycast_ip):
-                logging.info('Creating Anycast IP {} and assigning it to interface'.format(ip))
+                logging.info(
+                    "Creating Anycast IP {} and assigning it to interface".format(ip)
+                )
                 query_params = {
                     "address": ip,
                     "status": "active",
-                    "role": self.ipam_choices['ip-address:role']['Anycast'],
+                    "role": self.ipam_choices["ip-address:role"]["Anycast"],
                     "tenant": self.tenant.id if self.tenant else None,
                     "assigned_object_type": self.assigned_object_type,
-                    "assigned_object_id": interface.id
+                    "assigned_object_id": interface.id,
                 }
                 netbox_ip = nb.ipam.ip_addresses.create(**query_params)
             return netbox_ip
         else:
-            ip_interface = getattr(netbox_ip, 'interface', None)
-            assigned_object = getattr(netbox_ip, 'assigned_object', None)
+            ip_interface = getattr(netbox_ip, "interface", None)
+            assigned_object = getattr(netbox_ip, "assigned_object", None)
             if not ip_interface or not assigned_object:
-                logging.info('Assigning existing IP {ip} to {interface}'.format(
-                    ip=ip, interface=interface))
-            elif (ip_interface and ip_interface.id != interface.id) or \
-                 (assigned_object and assigned_object_id != interface.id):
+                logging.info(
+                    "Assigning existing IP {ip} to {interface}".format(
+                        ip=ip, interface=interface
+                    )
+                )
+            elif (ip_interface and ip_interface.id != interface.id) or (
+                assigned_object and assigned_object.id != interface.id
+            ):
 
                 old_interface = getattr(netbox_ip, "assigned_object", "n/a")
                 logging.info(
-                    'Detected interface change for ip {ip}: old interface is '
-                    '{old_interface} (id: {old_id}), new interface is {new_interface} '
-                    ' (id: {new_id})'
-                    .format(
-                        old_interface=old_interface, new_interface=interface,
-                        old_id=netbox_ip.id, new_id=interface.id, ip=netbox_ip.address
-                    ))
+                    "Detected interface change for ip {ip}: old interface is "
+                    "{old_interface} (id: {old_id}), new interface is {new_interface} "
+                    " (id: {new_id})".format(
+                        old_interface=old_interface,
+                        new_interface=interface,
+                        old_id=netbox_ip.id,
+                        new_id=interface.id,
+                        ip=netbox_ip.address,
+                    )
+                )
             else:
                 return netbox_ip
 
@@ -398,82 +440,98 @@ class Network(object):
     def create_or_update_netbox_network_cards(self):
         if config.update_all is None or config.update_network is None:
             return None
-        logging.debug('Creating/Updating NIC...')
+        logging.debug("Creating/Updating NIC...")
 
         # delete unknown interface
         nb_nics = list(self.get_netbox_network_cards())
-        local_nics = [x['name'] for x in self.nics]
+        local_nics = [x["name"] for x in self.nics]
         for nic in nb_nics:
             if nic.name not in local_nics:
-                logging.info('Deleting netbox interface {name} because not present locally'.format(
+                logging.info(
+                    "Deleting netbox interface {name} because not present locally".format(
                         name=nic.name
-                ))
+                    )
+                )
                 nb_nics.remove(nic)
                 nic.delete()
 
         # delete IP on netbox that are not known on this server
         if len(nb_nics):
-            netbox_ips = nb.ipam.ip_addresses.filter(
-                **{self.intf_type: [x.id for x in nb_nics]}
-            )
 
-            netbox_ips = list(netbox_ips)
-            all_local_ips = list(chain.from_iterable([
-                x['ip'] for x in self.nics if x['ip'] is not None
-            ]))
+            def batched(it, n):
+                while batch := tuple(islice(it, n)):
+                    yield batch
+
+            netbox_ips = []
+            for ids in batched((x.id for x in nb_nics), 25):
+                netbox_ips += list(nb.ipam.ip_addresses.filter(**{self.intf_type: ids}))
+
+            all_local_ips = list(
+                chain.from_iterable([x["ip"] for x in self.nics if x["ip"] is not None])
+            )
             for netbox_ip in netbox_ips:
                 if netbox_ip.address not in all_local_ips:
-                    logging.info('Unassigning IP {ip} from {interface}'.format(
-                        ip=netbox_ip.address, interface=netbox_ip.assigned_object))
+                    logging.info(
+                        "Unassigning IP {ip} from {interface}".format(
+                            ip=netbox_ip.address, interface=netbox_ip.assigned_object
+                        )
+                    )
                     netbox_ip.assigned_object_type = None
-                    netbox_ip.assigned_object_id = None
+                    netbox_ip.ssigned_object_id = None
                     netbox_ip.save()
 
         # update each nic
         for nic in self.nics:
             interface = self.get_netbox_network_card(nic)
             if not interface:
-                logging.info('Interface {mac_address} not found, creating..'.format(
-                    mac_address=nic['mac'])
+                logging.info(
+                    "Interface {mac_address} not found, creating..".format(
+                        mac_address=nic["mac"]
+                    )
                 )
                 interface = self.create_netbox_nic(nic)
 
             nic_update = 0
-            if nic['name'] != interface.name:
-                logging.info('Updating interface {interface} name to: {name}'.format(
-                    interface=interface, name=nic['name']))
-                interface.name = nic['name']
+            if nic["name"] != interface.name:
+                logging.info(
+                    "Updating interface {interface} name to: {name}".format(
+                        interface=interface, name=nic["name"]
+                    )
+                )
+                interface.name = nic["name"]
                 nic_update += 1
 
             ret, interface = self.reset_vlan_on_interface(nic, interface)
             nic_update += ret
 
-            if hasattr(interface, 'mtu'):
-                if nic['mtu'] != interface.mtu:
-                    logging.info('Interface mtu is wrong, updating to: {mtu}'.format(
-                        mtu=nic['mtu']))
-                    interface.mtu = nic['mtu']
+            if hasattr(interface, "mtu"):
+                if nic["mtu"] != interface.mtu:
+                    logging.info(
+                        "Interface mtu is wrong, updating to: {mtu}".format(
+                            mtu=nic["mtu"]
+                        )
+                    )
+                    interface.mtu = nic["mtu"]
                     nic_update += 1
 
-            if hasattr(interface, 'type'):
+            if hasattr(interface, "type"):
                 _type = self.get_netbox_type_for_nic(nic)
-                if not interface.type or \
-                   _type != interface.type.value:
-                    logging.info('Interface type is wrong, resetting')
+                if not interface.type or _type != interface.type.value:
+                    logging.info("Interface type is wrong, resetting")
                     interface.type = _type
                     nic_update += 1
 
-            if hasattr(interface, 'lag') and interface.lag is not None:
+            if hasattr(interface, "lag") and interface.lag is not None:
                 local_lag_int = next(
-                    item for item in self.nics if item['name'] == interface.lag.name
+                    item for item in self.nics if item["name"] == interface.lag.name
                 )
-                if nic['name'] not in local_lag_int['bonding_slaves']:
-                    logging.info('Interface has no LAG, resetting')
+                if nic["name"] not in local_lag_int["bonding_slaves"]:
+                    logging.info("Interface has no LAG, resetting")
                     nic_update += 1
                     interface.lag = None
 
             # cable the interface
-            if config.network.lldp:
+            if config.network.lldp and isinstance(self, ServerNetwork):
                 switch_ip = self.lldp.get_switch_ip(interface.name)
                 switch_interface = self.lldp.get_switch_port(interface.name)
                 if switch_ip and switch_interface:
@@ -482,15 +540,15 @@ class Network(object):
                     )
                     nic_update += ret
 
-            if nic['ip']:
+            if nic["ip"]:
                 # sync local IPs
-                for ip in nic['ip']:
+                for ip in nic["ip"]:
                     self.create_or_update_netbox_ip_on_interface(ip, interface)
             if nic_update > 0:
                 interface.save()
 
         self._set_bonding_interfaces()
-        logging.debug('Finished updating NIC!')
+        logging.debug("Finished updating NIC!")
 
 
 class ServerNetwork(Network):
@@ -505,38 +563,43 @@ class ServerNetwork(Network):
         self.server = server
         self.device = self.server.get_netbox_server()
         self.nb_net = nb.dcim
-        self.custom_arg = {'device': getattr(self.device, "id", None)}
-        self.custom_arg_id = {'device_id': getattr(self.device, "id", None)}
+        self.custom_arg = {"device": getattr(self.device, "id", None)}
+        self.custom_arg_id = {"device_id": getattr(self.device, "id", None)}
         self.intf_type = "interface_id"
         self.assigned_object_type = "dcim.interface"
 
     def get_network_type(self):
-        return 'server'
+        return "server"
 
     def get_ipmi(self):
         ipmi = IPMI().parse()
         return ipmi
 
-    def connect_interface_to_switch(self, switch_ip, switch_interface, nb_server_interface):
-        logging.info('Interface {} is not connected to switch, trying to connect..'.format(
+    def connect_interface_to_switch(
+        self, switch_ip, switch_interface, nb_server_interface
+    ):
+        logging.info(
+            "Interface {} is not connected to switch, trying to connect..".format(
                 nb_server_interface.name
-        ))
+            )
+        )
         nb_mgmt_ip = nb.ipam.ip_addresses.get(
             address=switch_ip,
         )
         if not nb_mgmt_ip:
-            logging.error('Switch IP {} cannot be found in Netbox'.format(switch_ip))
+            logging.error("Switch IP {} cannot be found in Netbox".format(switch_ip))
             return nb_server_interface
 
         try:
             nb_switch = nb_mgmt_ip.assigned_object.device
-            logging.info('Found a switch in Netbox based on LLDP infos: {} (id: {})'.format(
-                switch_ip,
-                nb_switch.id
-            ))
+            logging.info(
+                "Found a switch in Netbox based on LLDP infos: {} (id: {})".format(
+                    switch_ip, nb_switch.id
+                )
+            )
         except KeyError:
             logging.error(
-                'Switch IP {} is found but not associated to a Netbox Switch Device'.format(
+                "Switch IP {} is found but not associated to a Netbox Switch Device".format(
                     switch_ip
                 )
             )
@@ -548,13 +611,17 @@ class ServerNetwork(Network):
             name=switch_interface,
         )
         if nb_switch_interface is None:
-            logging.error('Switch interface {} cannot be found'.format(switch_interface))
+            logging.error(
+                "Switch interface {} cannot be found".format(switch_interface)
+            )
             return nb_server_interface
 
-        logging.info('Found interface {} on switch {}'.format(
+        logging.info(
+            "Found interface {} on switch {}".format(
                 switch_interface,
                 switch_ip,
-        ))
+            )
+        )
         cable = nb.dcim.cables.create(
             a_terminations=[
                 {"object_type": "dcim.interface", "object_id": nb_server_interface.id},
@@ -565,7 +632,7 @@ class ServerNetwork(Network):
         )
         nb_server_interface.cable = cable
         logging.info(
-            'Connected interface {interface} with {switch_interface} of {switch_ip}'.format(
+            "Connected interface {interface} with {switch_interface} of {switch_ip}".format(
                 interface=nb_server_interface.name,
                 switch_interface=switch_interface,
                 switch_ip=switch_ip,
@@ -583,38 +650,30 @@ class ServerNetwork(Network):
         else:
             nb_sw_int = nb_server_interface.cable.b_terminations[0]
             nb_sw = nb_sw_int.device
-            nb_mgmt_int = nb.dcim.interfaces.get(
-                device_id=nb_sw.id,
-                mgmt_only=True
-            )
-            nb_mgmt_ip = nb.ipam.ip_addresses.get(
-                interface_id=nb_mgmt_int.id
-            )
+            nb_mgmt_int = nb.dcim.interfaces.get(device_id=nb_sw.id, mgmt_only=True)
+            nb_mgmt_ip = nb.ipam.ip_addresses.get(interface_id=nb_mgmt_int.id)
             if nb_mgmt_ip is None:
                 logging.error(
-                    'Switch {switch_ip} does not have IP on its management interface'.format(
+                    "Switch {switch_ip} does not have IP on its management interface".format(
                         switch_ip=switch_ip,
                     )
                 )
                 return update, nb_server_interface
 
             # Netbox IP is always IP/Netmask
-            nb_mgmt_ip = nb_mgmt_ip.address.split('/')[0]
-            if nb_mgmt_ip != switch_ip or \
-               nb_sw_int.name != switch_interface:
-                logging.info('Netbox cable is not connected to correct ports, fixing..')
+            nb_mgmt_ip = nb_mgmt_ip.address.split("/")[0]
+            if nb_mgmt_ip != switch_ip or nb_sw_int.name != switch_interface:
+                logging.info("Netbox cable is not connected to correct ports, fixing..")
                 logging.info(
-                    'Deleting cable {cable_id} from {interface} to {switch_interface} of '
-                    '{switch_ip}'.format(
+                    "Deleting cable {cable_id} from {interface} to {switch_interface} of "
+                    "{switch_ip}".format(
                         cable_id=nb_server_interface.cable.id,
                         interface=nb_server_interface.name,
                         switch_interface=nb_sw_int.name,
                         switch_ip=nb_mgmt_ip,
                     )
                 )
-                cable = nb.dcim.cables.get(
-                    nb_server_interface.cable.id
-                )
+                cable = nb.dcim.cables.get(nb_server_interface.cable.id)
                 cable.delete()
                 update = True
                 nb_server_interface = self.connect_interface_to_switch(
@@ -629,17 +688,17 @@ class VirtualNetwork(Network):
         self.server = server
         self.device = self.server.get_netbox_vm()
         self.nb_net = nb.virtualization
-        self.custom_arg = {'virtual_machine': getattr(self.device, "id", None)}
-        self.custom_arg_id = {'virtual_machine_id': getattr(self.device, "id", None)}
+        self.custom_arg = {"virtual_machine": getattr(self.device, "id", None)}
+        self.custom_arg_id = {"virtual_machine_id": getattr(self.device, "id", None)}
         self.intf_type = "vminterface_id"
         self.assigned_object_type = "virtualization.vminterface"
 
         dcim_c = nb.virtualization.interfaces.choices()
         for _choice_type in dcim_c:
-            key = 'interface:{}'.format(_choice_type)
+            key = "interface:{}".format(_choice_type)
             self.dcim_choices[key] = {}
             for choice in dcim_c[_choice_type]:
-                self.dcim_choices[key][choice['display_name']] = choice['value']
+                self.dcim_choices[key][choice["display_name"]] = choice["value"]
 
     def get_network_type(self):
-        return 'virtual'
+        return "virtual"

netbox_agent/power.py

@@ -6,49 +6,51 @@ from netbox_agent.config import netbox_instance as nb
 PSU_DMI_TYPE = 39
 
 
-class PowerSupply():
+class PowerSupply:
     def __init__(self, server=None):
         self.server = server
         self.netbox_server = self.server.get_netbox_server()
         if self.server.is_blade():
-            self.device_id = self.netbox_server.parent_device.id if self.netbox_server else None
+            self.device_id = (
+                self.netbox_server.parent_device.id if self.netbox_server else None
+            )
         else:
             self.device_id = self.netbox_server.id if self.netbox_server else None
 
     def get_power_supply(self):
         power_supply = []
         for psu in dmidecode.get_by_type(self.server.dmi, PSU_DMI_TYPE):
-            if 'Present' not in psu['Status'] or psu['Status'] == 'Not Present':
+            if "Present" not in psu["Status"] or psu["Status"] == "Not Present":
                 continue
 
             try:
-                max_power = int(psu.get('Max Power Capacity').split()[0])
+                max_power = int(psu.get("Max Power Capacity").split()[0])
             except ValueError:
                 max_power = None
-            desc = '{} - {}'.format(
-                psu.get('Manufacturer', 'No Manufacturer').strip(),
-                psu.get('Name', 'No name').strip(),
+            desc = "{} - {}".format(
+                psu.get("Manufacturer", "No Manufacturer").strip(),
+                psu.get("Name", "No name").strip(),
             )
 
-            sn = psu.get('Serial Number', '').strip()
+            sn = psu.get("Serial Number", "").strip()
             # Let's assume that if no serial and no power reported we skip it
-            if sn == '' and max_power is None:
+            if sn == "" and max_power is None:
                 continue
-            if sn == '':
-                sn = 'N/A'
-            power_supply.append({
-                'name': sn,
-                'description': desc,
-                'allocated_draw': None,
-                'maximum_draw': max_power,
-                'device': self.device_id,
-            })
+            if sn == "":
+                sn = "N/A"
+            power_supply.append(
+                {
+                    "name": sn,
+                    "description": desc,
+                    "allocated_draw": None,
+                    "maximum_draw": max_power,
+                    "device": self.device_id,
+                }
+            )
         return power_supply
 
     def get_netbox_power_supply(self):
-        return nb.dcim.power_ports.filter(
-            device_id=self.device_id
-        )
+        return nb.dcim.power_ports.filter(device_id=self.device_id)
 
     def create_or_update_power_supply(self):
         nb_psus = list(self.get_netbox_power_supply())
@@ -57,10 +59,10 @@ class PowerSupply():
         # Delete unknown PSU
         delete = False
         for nb_psu in nb_psus:
-            if nb_psu.name not in [x['name'] for x in psus]:
-                logging.info('Deleting unknown locally PSU {name}'.format(
-                    name=nb_psu.name
-                ))
+            if nb_psu.name not in [x["name"] for x in psus]:
+                logging.info(
+                    "Deleting unknown locally PSU {name}".format(name=nb_psu.name)
+                )
                 nb_psu.delete()
                 delete = True
 
@@ -69,27 +71,23 @@ class PowerSupply():
 
         # sync existing Netbox PSU with local infos
         for nb_psu in nb_psus:
-            local_psu = next(
-                item for item in psus if item['name'] == nb_psu.name
-            )
+            local_psu = next(item for item in psus if item["name"] == nb_psu.name)
             update = False
-            if nb_psu.description != local_psu['description']:
+            if nb_psu.description != local_psu["description"]:
                 update = True
-                nb_psu.description = local_psu['description']
-            if nb_psu.maximum_draw != local_psu['maximum_draw']:
+                nb_psu.description = local_psu["description"]
+            if nb_psu.maximum_draw != local_psu["maximum_draw"]:
                 update = True
-                nb_psu.maximum_draw = local_psu['maximum_draw']
+                nb_psu.maximum_draw = local_psu["maximum_draw"]
             if update:
                 nb_psu.save()
 
         for psu in psus:
-            if psu['name'] not in [x.name for x in nb_psus]:
-                logging.info('Creating PSU {name} ({description}), {maximum_draw}W'.format(
-                    **psu
-                ))
-                nb_psu = nb.dcim.power_ports.create(
-                    **psu
+            if psu["name"] not in [x.name for x in nb_psus]:
+                logging.info(
+                    "Creating PSU {name} ({description}), {maximum_draw}W".format(**psu)
                 )
+                nb_psu = nb.dcim.power_ports.create(**psu)
 
         return True
 
@@ -97,7 +95,7 @@ class PowerSupply():
         try:
             psu_cons = self.server.get_power_consumption()
         except NotImplementedError:
-            logging.error('Cannot report power consumption for this vendor')
+            logging.error("Cannot report power consumption for this vendor")
             return False
         nb_psus = self.get_netbox_power_supply()
 
@@ -107,25 +105,25 @@ class PowerSupply():
         # find power feeds for rack or dc
         pwr_feeds = None
         if self.netbox_server.rack:
-            pwr_feeds = nb.dcim.power_feeds.filter(
-                rack=self.netbox_server.rack.id
-            )
+            pwr_feeds = nb.dcim.power_feeds.filter(rack=self.netbox_server.rack.id)
 
         if pwr_feeds:
-            voltage = [p['voltage'] for p in pwr_feeds]
+            voltage = [p["voltage"] for p in pwr_feeds]
         else:
-            logging.info('Could not find power feeds for Rack, defaulting value to 230')
+            logging.info("Could not find power feeds for Rack, defaulting value to 230")
             voltage = [230 for _ in nb_psus]
 
         for i, nb_psu in enumerate(nb_psus):
             nb_psu.allocated_draw = int(float(psu_cons[i]) * voltage[i])
             if nb_psu.allocated_draw < 1:
-                logging.info('PSU is not connected or in standby mode')
+                logging.info("PSU is not connected or in standby mode")
                 continue
             nb_psu.save()
-            logging.info('Updated power consumption for PSU {}: {}W'.format(
+            logging.info(
+                "Updated power consumption for PSU {}: {}W".format(
                     nb_psu.name,
                     nb_psu.allocated_draw,
-            ))
+                )
+            )
 
         return True

netbox_agent/raid/base.py

@@ -1,4 +1,4 @@
-class RaidController():
+class RaidController:
 
     def get_product_name(self):
         raise NotImplementedError
@@ -19,6 +19,6 @@ class RaidController():
         return False
 
 
-class Raid():
+class Raid:
     def get_controllers(self):
         raise NotImplementedError

netbox_agent/raid/hp.py

@@ -1,45 +1,50 @@
-from netbox_agent.raid.base import Raid, RaidController
-from netbox_agent.misc import get_vendor
-from netbox_agent.config import config
-import subprocess
 import logging
 import re
+import subprocess
+
+from netbox_agent.misc import get_vendor
+from netbox_agent.raid.base import Raid, RaidController
+
+REGEXP_CONTROLLER_HP = re.compile(r"Smart Array ([a-zA-Z0-9- ]+) in Slot ([0-9]+)")
 
-REGEXP_CONTROLLER_HP = re.compile(r'Smart Array ([a-zA-Z0-9- ]+) in Slot ([0-9]+)')
 
 class HPRaidControllerError(Exception):
     pass
 
+
 def ssacli(sub_command):
     command = ["ssacli"]
     command.extend(sub_command.split())
-    p = subprocess.Popen(
-        command,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT
-    )
+    p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
     stdout, stderr = p.communicate()
     stdout = stdout.decode("utf-8")
     if p.returncode != 0:
-        mesg = "Failed to execute command '{}':\n{}".format(
-            " ".join(command), stdout
-        )
+        mesg = "Failed to execute command '{}':\n{}".format(" ".join(command), stdout)
         raise HPRaidControllerError(mesg)
 
-    if 'does not have any physical' in stdout:
+    if "does not have any physical" in stdout:
         return list()
     else:
-        lines = stdout.split('\n')
+        lines = stdout.split("\n")
         lines = list(filter(None, lines))
         return lines
 
+
 def _test_if_valid_line(line):
-    ignore_patterns = ['Note:', 'Error:', 'is not loaded', 'README', ' failure', ' cache']
+    ignore_patterns = [
+        "Note:",
+        "Error:",
+        "is not loaded",
+        "README",
+        " failure",
+        " cache",
+    ]
     for pattern in ignore_patterns:
         if not line or pattern in line:
             return None
     return line
 
+
 def _parse_ctrl_output(lines):
     controllers = {}
     current_ctrl = None
@@ -53,14 +58,14 @@ def _parse_ctrl_output(lines):
         if ctrl is not None:
             slot = ctrl.group(2)
             current_ctrl = "{} - Slot {}".format(ctrl.group(1), slot)
-            controllers[current_ctrl] = {'Slot': slot}
-            if 'Embedded' not in line:
-                controllers[current_ctrl]['External'] = True
+            controllers[current_ctrl] = {"Slot": slot}
+            if "Embedded" not in line:
+                controllers[current_ctrl]["External"] = True
                 continue
-        if ': ' not in line:
+        if ": " not in line:
             continue
 
-        attr, val = line.split(': ', 1)
+        attr, val = line.split(": ", 1)
         attr = attr.strip()
         val = val.strip()
         controllers[current_ctrl][attr] = val
@@ -78,25 +83,24 @@ def _parse_pd_output(lines):
         if line is None:
             continue
         # Parses the Array the drives are in
-        if line.startswith('Array'):
+        if line.startswith("Array"):
             current_array = line.split(None, 1)[1]
         # Detects new physical drive
-        if line.startswith('physicaldrive'):
+        if line.startswith("physicaldrive"):
             current_drv = line.split(None, 1)[1]
             drives[current_drv] = {}
             if current_array is not None:
-                drives[current_drv]['Array'] = current_array
+                drives[current_drv]["Array"] = current_array
             continue
-        if ': ' not in line:
+        if ": " not in line:
             continue
-        attr, val = line.split(': ', 1)
+        attr, val = line.split(": ", 1)
         attr = attr.strip()
         val = val.strip()
         drives.setdefault(current_drv, {})[attr] = val
     return drives
 
 
-
 def _parse_ld_output(lines):
     drives = {}
     current_array = None
@@ -108,17 +112,17 @@ def _parse_ld_output(lines):
         if line is None:
             continue
         # Parses the Array the drives are in
-        if line.startswith('Array'):
+        if line.startswith("Array"):
             current_array = line.split(None, 1)[1]
             drives[current_array] = {}
         # Detects new physical drive
-        if line.startswith('Logical Drive'):
-            current_drv = line.split(': ', 1)[1]
-            drives.setdefault(current_array, {})['LogicalDrive'] = current_drv
+        if line.startswith("Logical Drive"):
+            current_drv = line.split(": ", 1)[1]
+            drives.setdefault(current_array, {})["LogicalDrive"] = current_drv
             continue
-        if ': ' not in line:
+        if ": " not in line:
             continue
-        attr, val = line.split(': ', 1)
+        attr, val = line.split(": ", 1)
         drives.setdefault(current_array, {})[attr] = val
     return drives
 
@@ -128,7 +132,7 @@ class HPRaidController(RaidController):
         self.controller_name = controller_name
         self.data = data
         self.pdrives = self._get_physical_disks()
-        arrays = [d['Array'] for d in self.pdrives.values() if d.get('Array')]
+        arrays = [d["Array"] for d in self.pdrives.values() if d.get("Array")]
         if arrays:
             self.ldrives = self._get_logical_drives()
             self._get_virtual_drives_map()
@@ -137,64 +141,69 @@ class HPRaidController(RaidController):
         return self.controller_name
 
     def get_manufacturer(self):
-        return 'HP'
+        return "HP"
 
     def get_serial_number(self):
-        return self.data['Serial Number']
+        return self.data["Serial Number"]
 
     def get_firmware_version(self):
-        return self.data['Firmware Version']
+        return self.data["Firmware Version"]
 
     def is_external(self):
-        return self.data.get('External', False)
+        return self.data.get("External", False)
 
     def _get_physical_disks(self):
-        lines = ssacli('ctrl slot={} pd all show detail'.format(self.data['Slot']))
+        lines = ssacli("ctrl slot={} pd all show detail".format(self.data["Slot"]))
         pdrives = _parse_pd_output(lines)
         ret = {}
 
         for name, attrs in pdrives.items():
-            array = attrs.get('Array', '')
-            model = attrs.get('Model', '').strip()
+            array = attrs.get("Array", "")
+            model = attrs.get("Model", "").strip()
             vendor = None
-            if model.startswith('HP'):
-                vendor = 'HP'
+            if model.startswith("HP"):
+                vendor = "HP"
             elif len(model.split()) > 1:
                 vendor = get_vendor(model.split()[1])
             else:
                 vendor = get_vendor(model)
 
             ret[name] = {
-                'Array': array,
-                'Model': model,
-                'Vendor': vendor,
-                'SN': attrs.get('Serial Number', '').strip(),
-                'Size': attrs.get('Size', '').strip(),
-                'Type': 'SSD' if attrs.get('Interface Type') == 'Solid State SATA'
-                else 'HDD',
-                '_src': self.__class__.__name__,
-                'custom_fields':  {
-                    'pd_identifier': name,
-                    'mount_point': attrs.get('Mount Points', '').strip(),
-                    'vd_device': attrs.get('Disk Name', '').strip(),
-                    'vd_size': attrs.get('Size', '').strip(),
-                }
+                "Array": array,
+                "Model": model,
+                "Vendor": vendor,
+                "SN": attrs.get("Serial Number", "").strip(),
+                "Size": attrs.get("Size", "").strip(),
+                "Type": (
+                    "SSD"
+                    if attrs.get("Interface Type") == "Solid State SATA"
+                    else "HDD"
+                ),
+                "_src": self.__class__.__name__,
+                "custom_fields": {
+                    "pd_identifier": name,
+                    "mount_point": attrs.get("Mount Points", "").strip(),
+                    "vd_device": attrs.get("Disk Name", "").strip(),
+                    "vd_size": attrs.get("Size", "").strip(),
+                },
             }
         return ret
 
     def _get_logical_drives(self):
-        lines = ssacli('ctrl slot={} ld all show detail'.format(self.data['Slot']))
+        lines = ssacli("ctrl slot={} ld all show detail".format(self.data["Slot"]))
         ldrives = _parse_ld_output(lines)
         ret = {}
 
         for array, attrs in ldrives.items():
             ret[array] = {
-                'vd_array': array,
-                'vd_size': attrs.get('Size', '').strip(),
-                'vd_consistency': attrs.get('Status', '').strip(),
-                'vd_raid_type': 'RAID {}'.format(attrs.get('Fault Tolerance', 'N/A').strip()),
-                'vd_device': attrs.get('LogicalDrive', '').strip(),
-                'mount_point': attrs.get('Mount Points', '').strip()
+                "vd_array": array,
+                "vd_size": attrs.get("Size", "").strip(),
+                "vd_consistency": attrs.get("Status", "").strip(),
+                "vd_raid_type": "RAID {}".format(
+                    attrs.get("Fault Tolerance", "N/A").strip()
+                ),
+                "vd_device": attrs.get("LogicalDrive", "").strip(),
+                "mount_point": attrs.get("Mount Points", "").strip(),
             }
         return ret
 
@@ -208,7 +217,7 @@ class HPRaidController(RaidController):
                     " Ignoring.".format(name)
                 )
                 continue
-            attrs['custom_fields'].update(ld)
+            attrs["custom_fields"].update(ld)
 
     def get_physical_disks(self):
         return list(self.pdrives.values())
@@ -216,18 +225,16 @@ class HPRaidController(RaidController):
 
 class HPRaid(Raid):
     def __init__(self):
-        self.output = subprocess.getoutput('ssacli ctrl all show detail')
+        self.output = subprocess.getoutput("ssacli ctrl all show detail")
         self.controllers = []
         self.convert_to_dict()
 
     def convert_to_dict(self):
-        lines = self.output.split('\n')
+        lines = self.output.split("\n")
         lines = list(filter(None, lines))
         controllers = _parse_ctrl_output(lines)
         for controller, attrs in controllers.items():
-            self.controllers.append(
-                HPRaidController(controller, attrs)
-            )
+            self.controllers.append(HPRaidController(controller, attrs))
 
     def get_controllers(self):
         return self.controllers

netbox_agent/raid/omreport.py

@@ -1,9 +1,9 @@
-from netbox_agent.raid.base import Raid, RaidController
-from netbox_agent.misc import get_vendor, get_mount_points
-from netbox_agent.config import config
-import subprocess
 import logging
 import re
+import subprocess
+
+from netbox_agent.misc import get_mount_points, get_vendor
+from netbox_agent.raid.base import Raid, RaidController
 
 
 class OmreportControllerError(Exception):
@@ -13,30 +13,24 @@ class OmreportControllerError(Exception):
 def omreport(sub_command):
     command = ["omreport"]
     command.extend(sub_command.split())
-    p = subprocess.Popen(
-        command,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT
-    )
+    p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
     p.wait()
     stdout = p.stdout.read().decode("utf-8")
     if p.returncode != 0:
-        mesg = "Failed to execute command '{}':\n{}".format(
-            " ".join(command), stdout
-        )
+        mesg = "Failed to execute command '{}':\n{}".format(" ".join(command), stdout)
         raise OmreportControllerError(mesg)
 
     res = {}
-    section_re = re.compile('^[A-Z]')
+    section_re = re.compile("^[A-Z]")
     current_section = None
     current_obj = None
 
-    for line in stdout.split('\n'):
-        if ': ' in line:
-            attr, value = line.split(': ', 1)
+    for line in stdout.split("\n"):
+        if ": " in line:
+            attr, value = line.split(": ", 1)
             attr = attr.strip()
             value = value.strip()
-            if attr == 'ID':
+            if attr == "ID":
                 obj = {}
                 res.setdefault(current_section, []).append(obj)
                 current_obj = obj
@@ -52,60 +46,57 @@ class OmreportController(RaidController):
         self.controller_index = controller_index
 
     def get_product_name(self):
-        return self.data['Name']
+        return self.data["Name"]
 
     def get_manufacturer(self):
         return None
 
     def get_serial_number(self):
-        return self.data.get('DeviceSerialNumber')
+        return self.data.get("DeviceSerialNumber")
 
     def get_firmware_version(self):
-        return self.data.get('Firmware Version')
+        return self.data.get("Firmware Version")
 
     def _get_physical_disks(self):
         pds = {}
-        res = omreport('storage pdisk controller={}'.format(
-            self.controller_index
-        ))
+        res = omreport("storage pdisk controller={}".format(self.controller_index))
         for pdisk in [d for d in list(res.values())[0]]:
-            disk_id = pdisk['ID']
-            size = re.sub('B .*$', 'B', pdisk['Capacity'])
+            disk_id = pdisk["ID"]
+            size = re.sub("B .*$", "B", pdisk["Capacity"])
             pds[disk_id] = {
-                'Vendor': get_vendor(pdisk['Vendor ID']),
-                'Model': pdisk['Product ID'],
-                'SN': pdisk['Serial No.'],
-                'Size': size,
-                'Type': pdisk['Media'],
-                '_src': self.__class__.__name__,
+                "Vendor": get_vendor(pdisk["Vendor ID"]),
+                "Model": pdisk["Product ID"],
+                "SN": pdisk["Serial No."],
+                "Size": size,
+                "Type": pdisk["Media"],
+                "_src": self.__class__.__name__,
             }
         return pds
 
     def _get_virtual_drives_map(self):
         pds = {}
-        res = omreport('storage vdisk controller={}'.format(
-            self.controller_index
-        ))
+        res = omreport("storage vdisk controller={}".format(self.controller_index))
         for vdisk in [d for d in list(res.values())[0]]:
-            vdisk_id = vdisk['ID']
-            device = vdisk['Device Name']
+            vdisk_id = vdisk["ID"]
+            device = vdisk["Device Name"]
             mount_points = get_mount_points()
-            mp = mount_points.get(device, 'n/a')
-            size = re.sub('B .*$', 'B', vdisk['Size'])
+            mp = mount_points.get(device, "n/a")
+            size = re.sub("B .*$", "B", vdisk["Size"])
             vd = {
-                'vd_array': vdisk_id,
-                'vd_size': size,
-                'vd_consistency': vdisk['State'],
-                'vd_raid_type': vdisk['Layout'],
-                'vd_device': vdisk['Device Name'],
-                'mount_point': ', '.join(sorted(mp)),
+                "vd_array": vdisk_id,
+                "vd_size": size,
+                "vd_consistency": vdisk["State"],
+                "vd_raid_type": vdisk["Layout"],
+                "vd_device": vdisk["Device Name"],
+                "mount_point": ", ".join(sorted(mp)),
             }
             drives_res = omreport(
-                'storage pdisk controller={} vdisk={}'.format(
+                "storage pdisk controller={} vdisk={}".format(
                     self.controller_index, vdisk_id
-            ))
+                )
+            )
             for pdisk in [d for d in list(drives_res.values())[0]]:
-                pds[pdisk['ID']] = vd
+                pds[pdisk["ID"]] = vd
         return pds
 
     def get_physical_disks(self):
@@ -114,27 +105,23 @@ class OmreportController(RaidController):
         for pd_identifier, vd in vds.items():
             if pd_identifier not in pds:
                 logging.error(
-                    'Physical drive {} listed in virtual drive {} not '
-                    'found in drives list'.format(
-                      pd_identifier, vd['vd_array']
-                    )
+                    "Physical drive {} listed in virtual drive {} not "
+                    "found in drives list".format(pd_identifier, vd["vd_array"])
                 )
                 continue
-            pds[pd_identifier].setdefault('custom_fields', {}).update(vd)
-            pds[pd_identifier]['custom_fields']['pd_identifier'] = pd_identifier
+            pds[pd_identifier].setdefault("custom_fields", {}).update(vd)
+            pds[pd_identifier]["custom_fields"]["pd_identifier"] = pd_identifier
         return list(pds.values())
 
 
 class OmreportRaid(Raid):
     def __init__(self):
         self.controllers = []
-        res = omreport('storage controller')
+        res = omreport("storage controller")
 
-        for controller in res['Controller']:
-            ctrl_index = controller['ID']
-            self.controllers.append(
-                OmreportController(ctrl_index, controller)
-            )
+        for controller in res["Controller"]:
+            ctrl_index = controller["ID"]
+            self.controllers.append(OmreportController(ctrl_index, controller))
 
     def get_controllers(self):
         return self.controllers

netbox_agent/raid/storcli.py

@@ -1,11 +1,12 @@
-from netbox_agent.raid.base import Raid, RaidController
-from netbox_agent.misc import get_vendor, get_mount_points
-from netbox_agent.config import config
-import subprocess
-import logging
 import json
-import re
+import logging
 import os
+import re
+import subprocess
+
+from netbox_agent.config import config
+from netbox_agent.misc import get_mount_points, get_vendor
+from netbox_agent.raid.base import Raid, RaidController
 
 
 class StorcliControllerError(Exception):
@@ -16,29 +17,23 @@ def storecli(sub_command):
     command = ["storcli"]
     command.extend(sub_command.split())
     command.append("J")
-    p = subprocess.Popen(
-        command,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT
-    )
+    p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 
     stdout, stderr = p.communicate()
     if stderr:
-        mesg = "Failed to execute command '{}':\n{}".format(
-            " ".join(command), stdout
-        )
+        mesg = "Failed to execute command '{}':\n{}".format(" ".join(command), stdout)
         raise StorcliControllerError(mesg)
 
     stdout = stdout.decode("utf-8")
     data = json.loads(stdout)
 
-    controllers = dict([
-        (
-            c['Command Status']['Controller'],
-            c['Response Data']
-        ) for c in data['Controllers']
-        if c['Command Status']['Status'] == 'Success'
-    ])
+    controllers = dict(
+        [
+            (c["Command Status"]["Controller"], c["Response Data"])
+            for c in data["Controllers"]
+            if c["Command Status"]["Status"] == "Success"
+        ]
+    )
     if not controllers:
         logging.error(
             "Failed to execute command '{}'. "
@@ -54,23 +49,23 @@ class StorcliController(RaidController):
         self.controller_index = controller_index
 
     def get_product_name(self):
-        return self.data['Product Name']
+        return self.data["Product Name"]
 
     def get_manufacturer(self):
         return None
 
     def get_serial_number(self):
-        return self.data['Serial Number']
+        return self.data["Serial Number"]
 
     def get_firmware_version(self):
-        return self.data['FW Package Build']
+        return self.data["FW Package Build"]
 
     def _get_physical_disks(self):
         pds = {}
-        cmd = '/c{}/eall/sall show all'.format(self.controller_index)
+        cmd = "/c{}/eall/sall show all".format(self.controller_index)
         controllers = storecli(cmd)
         pd_info = controllers[self.controller_index]
-        pd_re = re.compile(r'^Drive (/c\d+/e\d+/s\d+)$')
+        pd_re = re.compile(r"^Drive (/c\d+/e\d+/s\d+)$")
 
         for section, attrs in pd_info.items():
             reg = pd_re.search(section)
@@ -78,28 +73,28 @@ class StorcliController(RaidController):
                 continue
             pd_name = reg.group(1)
             pd_attr = attrs[0]
-            pd_identifier = pd_attr['EID:Slt']
-            size = pd_attr.get('Size', '').strip()
-            media_type = pd_attr.get('Med', '').strip()
-            pd_details = pd_info['{} - Detailed Information'.format(section)]
-            pd_dev_attr = pd_details['{} Device attributes'.format(section)]
-            model = pd_dev_attr.get('Model Number', '').strip()
+            pd_identifier = pd_attr["EID:Slt"]
+            size = pd_attr.get("Size", "").strip()
+            media_type = pd_attr.get("Med", "").strip()
+            pd_details = pd_info["{} - Detailed Information".format(section)]
+            pd_dev_attr = pd_details["{} Device attributes".format(section)]
+            model = pd_dev_attr.get("Model Number", "").strip()
             pd = {
-                'Model': model,
-                'Vendor': get_vendor(model),
-                'SN': pd_dev_attr.get('SN', '').strip(),
-                'Size': size,
-                'Type': media_type,
-                '_src': self.__class__.__name__,
+                "Model": model,
+                "Vendor": get_vendor(model),
+                "SN": pd_dev_attr.get("SN", "").strip(),
+                "Size": size,
+                "Type": media_type,
+                "_src": self.__class__.__name__,
             }
             if config.process_virtual_drives:
-                pd.setdefault('custom_fields', {})['pd_identifier'] = pd_name
+                pd.setdefault("custom_fields", {})["pd_identifier"] = pd_name
             pds[pd_identifier] = pd
         return pds
 
     def _get_virtual_drives_map(self):
         vds = {}
-        cmd = '/c{}/vall show all'.format(self.controller_index)
+        cmd = "/c{}/vall show all".format(self.controller_index)
         controllers = storecli(cmd)
         vd_info = controllers[self.controller_index]
         mount_points = get_mount_points()
@@ -109,9 +104,9 @@ class StorcliController(RaidController):
                 continue
             volume = vd_identifier.split("/")[-1].lstrip("v")
             vd_attr = vd_attrs[0]
-            vd_pd_identifier = 'PDs for VD {}'.format(volume)
+            vd_pd_identifier = "PDs for VD {}".format(volume)
             vd_pds = vd_info[vd_pd_identifier]
-            vd_prop_identifier = 'VD{} Properties'.format(volume)
+            vd_prop_identifier = "VD{} Properties".format(volume)
             vd_properties = vd_info[vd_prop_identifier]
             for pd in vd_pds:
                 pd_identifier = pd["EID:Slt"]
@@ -125,7 +120,7 @@ class StorcliController(RaidController):
                     "vd_consistency": vd_attr["Consist"],
                     "vd_raid_type": vd_attr["TYPE"],
                     "vd_device": device,
-                    "mount_point": ", ".join(sorted(mp))
+                    "mount_point": ", ".join(sorted(mp)),
                 }
         return vds
 
@@ -139,9 +134,7 @@ class StorcliController(RaidController):
             if pd_identifier not in pds:
                 logging.error(
                     "Physical drive {} listed in virtual drive {} not "
-                    "found in drives list".format(
-                      pd_identifier, vd["vd_array"]
-                    )
+                    "found in drives list".format(pd_identifier, vd["vd_array"])
                 )
                 continue
             pds[pd_identifier].setdefault("custom_fields", {}).update(vd)
@@ -152,14 +145,9 @@ class StorcliController(RaidController):
 class StorcliRaid(Raid):
     def __init__(self):
         self.controllers = []
-        controllers =  storecli('/call show')
+        controllers = storecli("/call show")
         for controller_id, controller_data in controllers.items():
-            self.controllers.append(
-                StorcliController(
-                    controller_id,
-                    controller_data
-                )
-            )
+            self.controllers.append(StorcliController(controller_id, controller_data))
 
     def get_controllers(self):
         return self.controllers

netbox_agent/server.py

@@ -1,46 +1,57 @@
+import logging
+import socket
+import subprocess
+import sys
+from pprint import pprint
+
 import netbox_agent.dmidecode as dmidecode
 from netbox_agent.config import config
 from netbox_agent.config import netbox_instance as nb
 from netbox_agent.inventory import Inventory
 from netbox_agent.location import Datacenter, Rack, Tenant
-from netbox_agent.misc import create_netbox_tags, get_device_role, get_device_type, get_device_platform
+from netbox_agent.misc import (
+    create_netbox_tags,
+    get_device_platform,
+    get_device_role,
+    get_device_type,
+)
 from netbox_agent.network import ServerNetwork
 from netbox_agent.power import PowerSupply
-from pprint import pprint
-import subprocess
-import logging
-import socket
-import sys
 
 
-class ServerBase():
+class ServerBase:
     def __init__(self, dmi=None):
         if dmi:
             self.dmi = dmi
         else:
             self.dmi = dmidecode.parse()
 
-        self.baseboard = dmidecode.get_by_type(self.dmi, 'Baseboard')
-        self.bios = dmidecode.get_by_type(self.dmi, 'BIOS')
-        self.chassis = dmidecode.get_by_type(self.dmi, 'Chassis')
-        self.system = dmidecode.get_by_type(self.dmi, 'System')
+        self.baseboard = dmidecode.get_by_type(self.dmi, "Baseboard")
+        self.bios = dmidecode.get_by_type(self.dmi, "BIOS")
+        self.chassis = dmidecode.get_by_type(self.dmi, "Chassis")
+        self.system = dmidecode.get_by_type(self.dmi, "System")
         self.device_platform = get_device_platform(config.device.platform)
 
         self.network = None
 
-        self.tags = list(set([
-            x.strip() for x in config.device.tags.split(',') if x.strip()
-        ])) if config.device.tags else []
+        self.tags = (
+            list(set([x.strip() for x in config.device.tags.split(",") if x.strip()]))
+            if config.device.tags
+            else []
+        )
         self.nb_tags = list(create_netbox_tags(self.tags))
-        config_cf = set([
-            f.strip() for f in config.device.custom_fields.split(",")
-            if f.strip()
-        ])
+        config_cf = set(
+            [f.strip() for f in config.device.custom_fields.split(",") if f.strip()]
+        )
         self.custom_fields = {}
-        self.custom_fields.update(dict([
-            (k.strip(), v.strip()) for k, v in
-            [f.split("=", 1) for f in config_cf]
-        ]))
+        self.custom_fields.update(
+            dict(
+                [
+                    (k.strip(), v.strip())
+                    for k, v in [f.split("=", 1) for f in config_cf]
+                ]
+            )
+        )
 
     def get_tenant(self):
         tenant = Tenant()
@@ -50,9 +61,7 @@ class ServerBase():
         tenant = self.get_tenant()
         if tenant is None:
             return None
-        nb_tenant = nb.tenancy.tenants.get(
-            slug=self.get_tenant()
-        )
+        nb_tenant = nb.tenancy.tenants.get(slug=self.get_tenant())
         return nb_tenant
 
     def get_datacenter(self):
@@ -81,22 +90,22 @@ class ServerBase():
 
         update = False
         if dc and server.site and server.site.slug != nb_dc.slug:
-            logging.info('Datacenter location has changed from {} to {}, updating'.format(
+            logging.info(
+                "Datacenter location has changed from {} to {}, updating".format(
                     server.site.slug,
                     nb_dc.slug,
-            ))
+                )
+            )
             update = True
             server.site = nb_dc.id
 
-        if (
-            server.rack
-            and nb_rack
-            and server.rack.id != nb_rack.id
-        ):
-            logging.info('Rack location has changed from {} to {}, updating'.format(
+        if server.rack and nb_rack and server.rack.id != nb_rack.id:
+            logging.info(
+                "Rack location has changed from {} to {}, updating".format(
                     server.rack,
                     nb_rack,
-            ))
+                )
+            )
             update = True
             server.rack = nb_rack
             if nb_rack is None:
@@ -139,24 +148,24 @@ class ServerBase():
         """
         Return the Chassis Name from dmidecode info
         """
-        return self.system[0]['Product Name'].strip()
+        return self.system[0]["Product Name"].strip()
 
     def get_service_tag(self):
         """
         Return the Service Tag from dmidecode info
         """
-        return self.system[0]['Serial Number'].strip()
+        return self.system[0]["Serial Number"].strip()
 
     def get_expansion_service_tag(self):
         """
         Return the virtual Service Tag from dmidecode info host
         with 'expansion'
         """
-        return self.system[0]['Serial Number'].strip() + " expansion"
+        return self.system[0]["Serial Number"].strip() + " expansion"
 
     def get_hostname(self):
         if config.hostname_cmd is None:
-            return '{}'.format(socket.gethostname())
+            return "{}".format(socket.gethostname())
         return subprocess.getoutput(config.hostname_cmd)
 
     def is_blade(self):
@@ -193,8 +202,7 @@ class ServerBase():
         device_type = get_device_type(self.get_chassis())
         device_role = get_device_role(config.device.chassis_role)
         serial = self.get_chassis_service_tag()
-        logging.info('Creating chassis blade (serial: {serial})'.format(
-            serial=serial))
+        logging.info("Creating chassis blade (serial: {serial})".format(serial=serial))
         new_chassis = nb.dcim.devices.create(
             name=self.get_chassis_name(),
             device_type=device_type.id,
@@ -203,7 +211,7 @@ class ServerBase():
             site=datacenter.id if datacenter else None,
             tenant=tenant.id if tenant else None,
             rack=rack.id if rack else None,
-            tags=[{'name': x} for x in self.tags],
+            tags=[{"name": x} for x in self.tags],
             custom_fields=self.custom_fields,
         )
         return new_chassis
@@ -214,9 +222,10 @@ class ServerBase():
         serial = self.get_service_tag()
         hostname = self.get_hostname()
         logging.info(
-            'Creating blade (serial: {serial}) {hostname} on chassis {chassis_serial}'.format(
+            "Creating blade (serial: {serial}) {hostname} on chassis {chassis_serial}".format(
                 serial=serial, hostname=hostname, chassis_serial=chassis.serial
-            ))
+            )
+        )
         new_blade = nb.dcim.devices.create(
             name=hostname,
             serial=serial,
@@ -226,7 +235,7 @@ class ServerBase():
             site=datacenter.id if datacenter else None,
             tenant=tenant.id if tenant else None,
             rack=rack.id if rack else None,
-            tags=[{'name': x} for x in self.tags],
+            tags=[{"name": x} for x in self.tags],
             custom_fields=self.custom_fields,
         )
         return new_blade
@@ -237,9 +246,10 @@ class ServerBase():
         serial = self.get_expansion_service_tag()
         hostname = self.get_hostname() + " expansion"
         logging.info(
-            'Creating expansion (serial: {serial}) {hostname} on chassis {chassis_serial}'.format(
+            "Creating expansion (serial: {serial}) {hostname} on chassis {chassis_serial}".format(
                 serial=serial, hostname=hostname, chassis_serial=chassis.serial
-            ))
+            )
+        )
         new_blade = nb.dcim.devices.create(
             name=hostname,
             serial=serial,
@@ -249,7 +259,7 @@ class ServerBase():
             site=datacenter.id if datacenter else None,
             tenant=tenant.id if tenant else None,
             rack=rack.id if rack else None,
-            tags=[{'name': x} for x in self.tags],
+            tags=[{"name": x} for x in self.tags],
         )
         return new_blade
 
@@ -267,8 +277,11 @@ class ServerBase():
             raise Exception('Chassis "{}" doesn\'t exist'.format(self.get_chassis()))
         serial = self.get_service_tag()
         hostname = self.get_hostname()
-        logging.info('Creating server (serial: {serial}) {hostname}'.format(
-            serial=serial, hostname=hostname))
+        logging.info(
+            "Creating server (serial: {serial}) {hostname}".format(
+                serial=serial, hostname=hostname
+            )
+        )
         new_server = nb.dcim.devices.create(
             name=hostname,
             serial=serial,
@@ -278,7 +291,7 @@ class ServerBase():
             site=datacenter.id if datacenter else None,
             tenant=tenant.id if tenant else None,
             rack=rack.id if rack else None,
-            tags=[{'name': x} for x in self.tags],
+            tags=[{"name": x} for x in self.tags],
         )
         return new_server
 
@@ -290,14 +303,16 @@ class ServerBase():
 
     def _netbox_set_or_update_blade_slot(self, server, chassis, datacenter):
         # before everything check if right chassis
-        actual_device_bay = server.parent_device.device_bay \
-                if server.parent_device else None
-        actual_chassis = actual_device_bay.device \
-                if actual_device_bay else None
+        actual_device_bay = (
+            server.parent_device.device_bay if server.parent_device else None
+        )
+        actual_chassis = actual_device_bay.device if actual_device_bay else None
         slot = self.get_blade_slot()
-        if actual_chassis and \
-           actual_chassis.serial == chassis.serial and \
-           actual_device_bay.name == slot:
+        if (
+            actual_chassis
+            and actual_chassis.serial == chassis.serial
+            and actual_device_bay.name == slot
+        ):
             return
 
         real_device_bays = nb.dcim.device_bays.filter(
@@ -310,10 +325,11 @@ class ServerBase():
         )
         if real_device_bays:
             logging.info(
-                'Setting device ({serial}) new slot on {slot} '
-                '(Chassis {chassis_serial})..'.format(
+                "Setting device ({serial}) new slot on {slot} "
+                "(Chassis {chassis_serial})..".format(
                     serial=server.serial, slot=slot, chassis_serial=chassis.serial
-                ))
+                )
+            )
             # reset actual device bay if set
             if actual_device_bay:
                 # Forces the evaluation of the installed_device attribute to
@@ -327,18 +343,22 @@ class ServerBase():
             real_device_bay.installed_device = server
             real_device_bay.save()
         else:
-            logging.error('Could not find slot {slot} for chassis'.format(
-                slot=slot
-            ))
+            logging.error("Could not find slot {slot} for chassis".format(slot=slot))
 
-    def _netbox_set_or_update_blade_expansion_slot(self, expansion, chassis, datacenter):
+    def _netbox_set_or_update_blade_expansion_slot(
+        self, expansion, chassis, datacenter
+    ):
         # before everything check if right chassis
-        actual_device_bay = expansion.parent_device.device_bay if expansion.parent_device else None
+        actual_device_bay = (
+            expansion.parent_device.device_bay if expansion.parent_device else None
+        )
         actual_chassis = actual_device_bay.device if actual_device_bay else None
         slot = self.get_blade_expansion_slot()
-        if actual_chassis and \
-           actual_chassis.serial == chassis.serial and \
-           actual_device_bay.name == slot:
+        if (
+            actual_chassis
+            and actual_chassis.serial == chassis.serial
+            and actual_device_bay.name == slot
+        ):
             return
 
         real_device_bays = nb.dcim.device_bays.filter(
@@ -346,15 +366,16 @@ class ServerBase():
             name=slot,
         )
         if not real_device_bays:
-            logging.error('Could not find slot {slot} expansion for chassis'.format(
-                slot=slot
-            ))
+            logging.error(
+                "Could not find slot {slot} expansion for chassis".format(slot=slot)
+            )
             return
         logging.info(
-            'Setting device expansion ({serial}) new slot on {slot} '
-            '(Chassis {chassis_serial})..'.format(
+            "Setting device expansion ({serial}) new slot on {slot} "
+            "(Chassis {chassis_serial})..".format(
                 serial=expansion.serial, slot=slot, chassis_serial=chassis.serial
-            ))
+            )
+        )
         # reset actual device bay if set
         if actual_device_bay:
             # Forces the evaluation of the installed_device attribute to
@@ -388,9 +409,7 @@ class ServerBase():
             self._netbox_deduplicate_server()
 
         if self.is_blade():
-            chassis = nb.dcim.devices.get(
-                serial=self.get_chassis_service_tag()
-            )
+            chassis = nb.dcim.devices.get(serial=self.get_chassis_service_tag())
             # Chassis does not exist
             if not chassis:
                 chassis = self._netbox_create_chassis(datacenter, tenant, rack)
@@ -406,13 +425,14 @@ class ServerBase():
             if not server:
                 server = self._netbox_create_server(datacenter, tenant, rack)
 
-        logging.debug('Updating Server...')
+        logging.debug("Updating Server...")
         # check network cards
         if config.register or config.update_all or config.update_network:
             self.network = ServerNetwork(server=self)
             self.network.create_or_update_netbox_network_cards()
-        update_inventory = config.inventory and (config.register or
-                config.update_all or config.update_inventory)
+        update_inventory = config.inventory and (
+            config.register or config.update_all or config.update_inventory
+        )
         # update inventory if feature is enabled
         self.inventory = Inventory(server=self)
         if update_inventory:
@@ -425,12 +445,16 @@ class ServerBase():
 
         expansion = nb.dcim.devices.get(serial=self.get_expansion_service_tag())
         if self.own_expansion_slot() and config.expansion_as_device:
-            logging.debug('Update Server expansion...')
+            logging.debug("Update Server expansion...")
             if not expansion:
-                expansion = self._netbox_create_blade_expansion(chassis, datacenter, tenant, rack)
+                expansion = self._netbox_create_blade_expansion(
+                    chassis, datacenter, tenant, rack
+                )
 
             # set slot for blade expansion
-            self._netbox_set_or_update_blade_expansion_slot(expansion, chassis, datacenter)
+            self._netbox_set_or_update_blade_expansion_slot(
+                expansion, chassis, datacenter
+            )
             if update_inventory:
                 # Updates expansion inventory
                 inventory = Inventory(server=self, update_expansion=True)
@@ -474,7 +498,7 @@ class ServerBase():
 
         if expansion:
             update = 0
-            expansion_name = server.name + ' expansion'
+            expansion_name = server.name + " expansion"
             if expansion.name != expansion_name:
                 expansion.name = expansion_name
                 update += 1
@@ -482,22 +506,24 @@ class ServerBase():
                 update += 1
             if update:
                 expansion.save()
-        logging.debug('Finished updating Server!')
+        logging.debug("Finished updating Server!")
 
     def print_debug(self):
         self.network = ServerNetwork(server=self)
-        print('Datacenter:', self.get_datacenter())
-        print('Netbox Datacenter:', self.get_netbox_datacenter())
-        print('Rack:', self.get_rack())
-        print('Netbox Rack:', self.get_netbox_rack())
-        print('Is blade:', self.is_blade())
-        print('Got expansion:', self.own_expansion_slot())
-        print('Product Name:', self.get_product_name())
-        print('Platform:', self.device_platform)
-        print('Chassis:', self.get_chassis())
-        print('Chassis service tag:', self.get_chassis_service_tag())
-        print('Service tag:', self.get_service_tag())
-        print('NIC:',)
+        print("Datacenter:", self.get_datacenter())
+        print("Netbox Datacenter:", self.get_netbox_datacenter())
+        print("Rack:", self.get_rack())
+        print("Netbox Rack:", self.get_netbox_rack())
+        print("Is blade:", self.is_blade())
+        print("Got expansion:", self.own_expansion_slot())
+        print("Product Name:", self.get_product_name())
+        print("Platform:", self.device_platform)
+        print("Chassis:", self.get_chassis())
+        print("Chassis service tag:", self.get_chassis_service_tag())
+        print("Service tag:", self.get_service_tag())
+        print(
+            "NIC:",
+        )
         pprint(self.network.get_network_cards())
         pass
 

netbox_agent/vendors/dell.py

@@ -8,10 +8,10 @@ from netbox_agent.server import ServerBase
 class DellHost(ServerBase):
     def __init__(self, *args, **kwargs):
         super(DellHost, self).__init__(*args, **kwargs)
-        self.manufacturer = 'Dell'
+        self.manufacturer = "Dell"
 
     def is_blade(self):
-        return self.get_product_name().startswith('PowerEdge M')
+        return self.get_product_name().startswith("PowerEdge M")
 
     def get_blade_slot(self):
         """
@@ -20,48 +20,48 @@ class DellHost(ServerBase):
         `        Location In Chassis: Slot 03`
         """
         if self.is_blade():
-            return self.baseboard[0].get('Location In Chassis').strip()
+            return self.baseboard[0].get("Location In Chassis").strip()
         return None
 
     def get_chassis_name(self):
         if not self.is_blade():
             return None
-        return 'Chassis {}'.format(self.get_service_tag())
+        return "Chassis {}".format(self.get_service_tag())
 
     def get_chassis(self):
         if self.is_blade():
-            return self.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.chassis[0]['Serial Number'].strip()
+            return self.chassis[0]["Serial Number"].strip()
         return self.get_service_tag()
 
     def get_power_consumption(self):
-        '''
+        """
         Parse omreport output like this
 
         Amperage
         PS1 Current 1 : 1.8 A
         PS2 Current 2 : 1.4 A
-        '''
+        """
         value = []
 
-        if not is_tool('omreport'):
-            logging.error('omreport does not seem to be installed, please debug')
+        if not is_tool("omreport"):
+            logging.error("omreport does not seem to be installed, please debug")
             return value
 
-        data = subprocess.getoutput('omreport chassis pwrmonitoring')
+        data = subprocess.getoutput("omreport chassis pwrmonitoring")
         amperage = False
         for line in data.splitlines():
-            if line.startswith('Amperage'):
+            if line.startswith("Amperage"):
                 amperage = True
                 continue
 
             if amperage:
-                if line.startswith('PS'):
-                    amp_value = line.split(':')[1].split()[0]
+                if line.startswith("PS"):
+                    amp_value = line.split(":")[1].split()[0]
                     value.append(amp_value)
                 else:
                     break

netbox_agent/vendors/generic.py

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

netbox_agent/vendors/hp.py

@@ -1,6 +1,6 @@
 import netbox_agent.dmidecode as dmidecode
-from netbox_agent.server import ServerBase
 from netbox_agent.inventory import Inventory
+from netbox_agent.server import ServerBase
 
 
 class HPHost(ServerBase):
@@ -13,8 +13,9 @@ class HPHost(ServerBase):
 
     def is_blade(self):
         blade = self.product.startswith("ProLiant BL")
-        blade |= self.product.startswith("ProLiant m") and \
-            self.product.endswith("Server Cartridge")
+        blade |= self.product.startswith("ProLiant m") and self.product.endswith(
+            "Server Cartridge"
+        )
         return blade
 
     def _find_rack_locator(self):
@@ -36,7 +37,9 @@ class HPHost(ServerBase):
                 }
 
             # HP ProLiant m750, m710x, m510 Server Cartridge
-            if self.product.startswith("ProLiant m") and self.product.endswith("Server Cartridge"):
+            if self.product.startswith("ProLiant m") and self.product.endswith(
+                "Server Cartridge"
+            ):
                 locator = dmidecode.get_by_type(self.dmi, 2)
                 chassis = dmidecode.get_by_type(self.dmi, 3)
                 return {
@@ -72,10 +75,14 @@ class HPHost(ServerBase):
         """
         Expansion slot are always the compute bay number + 1
         """
-        if self.is_blade() and self.own_gpu_expansion_slot() or \
-                self.own_disk_expansion_slot() or True:
-            return 'Bay {}'.format(
-                str(int(self.hp_rack_locator['Server Bay'].strip()) + 1)
+        if (
+            self.is_blade()
+            and self.own_gpu_expansion_slot()
+            or self.own_disk_expansion_slot()
+            or True
+        ):
+            return "Bay {}".format(
+                str(int(self.hp_rack_locator["Server Bay"].strip()) + 1)
             )
         return None
 
@@ -102,7 +109,7 @@ class HPHost(ServerBase):
         Indicates if the device hosts a GPU expansion card based
         on the product name
         """
-        return self.get_product_name().endswith('Graphics Exp')
+        return self.get_product_name().endswith("Graphics Exp")
 
     def own_disk_expansion_slot(self):
         """

netbox_agent/vendors/qct.py

@@ -4,29 +4,29 @@ from netbox_agent.server import ServerBase
 class QCTHost(ServerBase):
     def __init__(self, *args, **kwargs):
         super(QCTHost, self).__init__(*args, **kwargs)
-        self.manufacturer = 'QCT'
+        self.manufacturer = "QCT"
 
     def is_blade(self):
-        return 'Location In Chassis' in self.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.baseboard[0].get('Location In Chassis').strip()
+            return "Slot {}".format(
+                self.baseboard[0].get("Location In Chassis").strip()
             )
         return None
 
     def get_chassis_name(self):
         if not self.is_blade():
             return None
-        return 'Chassis {}'.format(self.get_service_tag())
+        return "Chassis {}".format(self.get_service_tag())
 
     def get_chassis(self):
         if self.is_blade():
-            return self.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.chassis[0]['Serial Number'].strip()
+            return self.chassis[0]["Serial Number"].strip()
         return self.get_service_tag()

netbox_agent/vendors/supermicro.py

@@ -18,22 +18,22 @@ class SupermicroHost(ServerBase):
 
     def __init__(self, *args, **kwargs):
         super(SupermicroHost, self).__init__(*args, **kwargs)
-        self.manufacturer = 'Supermicro'
+        self.manufacturer = "Supermicro"
 
     def is_blade(self):
-        product_name = self.system[0]['Product Name'].strip()
+        product_name = self.system[0]["Product Name"].strip()
         # Blades
-        blade = product_name.startswith('SBI')
-        blade |= product_name.startswith('SBA')
+        blade = product_name.startswith("SBI")
+        blade |= product_name.startswith("SBA")
         # Twin
-        blade |= 'TR-' in product_name
+        blade |= "TR-" in product_name
         # TwinPro
-        blade |= 'TP-' in product_name
+        blade |= "TP-" in product_name
         # BigTwin
-        blade |= 'BT-' in product_name
+        blade |= "BT-" in product_name
         # Microcloud
-        blade |= product_name.startswith('SYS-5039')
-        blade |= product_name.startswith('SYS-5038')
+        blade |= product_name.startswith("SYS-5039")
+        blade |= product_name.startswith("SYS-5038")
         return blade
 
     def get_blade_slot(self):
@@ -47,28 +47,28 @@ class SupermicroHost(ServerBase):
 
     def get_service_tag(self):
         if self.is_blade():
-            return self.baseboard[0]['Serial Number'].strip()
-        return self.system[0]['Serial Number'].strip()
+            return self.baseboard[0]["Serial Number"].strip()
+        return self.system[0]["Serial Number"].strip()
 
     def get_product_name(self):
         if self.is_blade():
-            return self.baseboard[0]['Product Name'].strip()
-        return self.system[0]['Product Name'].strip()
+            return self.baseboard[0]["Product Name"].strip()
+        return self.system[0]["Product Name"].strip()
 
     def get_chassis(self):
         if self.is_blade():
-            return self.system[0]['Product Name'].strip()
+            return self.system[0]["Product Name"].strip()
         return self.get_product_name()
 
     def get_chassis_service_tag(self):
         if self.is_blade():
-            return self.system[0]['Serial Number'].strip()
+            return self.system[0]["Serial Number"].strip()
         return self.get_service_tag()
 
     def get_chassis_name(self):
         if not self.is_blade():
             return None
-        return 'Chassis {}'.format(self.get_chassis_service_tag())
+        return "Chassis {}".format(self.get_chassis_service_tag())
 
     def get_expansion_product(self):
         """

netbox_agent/virtualmachine.py

@@ -1,34 +1,32 @@
 import os
+from pprint import pprint
 
 import netbox_agent.dmidecode as dmidecode
 from netbox_agent.config import config
 from netbox_agent.config import netbox_instance as nb
 from netbox_agent.location import Tenant
 from netbox_agent.logging import logging  # NOQA
-from netbox_agent.misc import create_netbox_tags, get_hostname, get_device_platform
+from netbox_agent.misc import create_netbox_tags, get_device_platform, get_hostname
 from netbox_agent.network import VirtualNetwork
 
 
 def is_vm(dmi):
-    bios = dmidecode.get_by_type(dmi, 'BIOS')[0]
-    system = dmidecode.get_by_type(dmi, 'System')[0]
+    bios = dmidecode.get_by_type(dmi, "BIOS")[0]
+    system = dmidecode.get_by_type(dmi, "System")[0]
 
     return (
+        "Hyper-V" in bios["Version"]
+        or "Xen" in bios["Version"]
+        or "Google Compute Engine" in system["Product Name"]
+    ) or (
         (
-            'Hyper-V' in bios['Version'] or
-            'Xen' in bios['Version'] or
-            'Google Compute Engine' in system['Product Name']
-        ) or
-        (
-            (
-                'Amazon EC2' in system['Manufacturer'] and
-                not system['Product Name'].endswith('.metal')
-            ) or
-            'RHEV Hypervisor' in system['Product Name'] or
-            'QEMU' in system['Manufacturer'] or
-            'VirtualBox' in bios['Version'] or
-            'VMware' in system['Manufacturer']
+            "Amazon EC2" in system["Manufacturer"]
+            and not system["Product Name"].endswith(".metal")
         )
+        or "RHEV Hypervisor" in system["Product Name"]
+        or "QEMU" in system["Manufacturer"]
+        or "VirtualBox" in bios["Version"]
+        or "VMware" in system["Manufacturer"]
     )
 
 
@@ -41,13 +39,16 @@ class VirtualMachine(object):
         self.network = None
         self.device_platform = get_device_platform(config.device.platform)
 
-        self.tags = list(set(config.device.tags.split(','))) if config.device.tags else []
-        if self.tags and len(self.tags):
-            create_netbox_tags(self.tags)
+        self.tags = (
+            list(set(config.device.tags.split(","))) if config.device.tags else []
+        )
+        self.nb_tags = create_netbox_tags(self.tags)
 
     def get_memory(self):
-        mem_bytes = os.sysconf('SC_PAGE_SIZE') * os.sysconf('SC_PHYS_PAGES')  # e.g. 4015976448
-        mem_gib = mem_bytes / (1024.**2)  # e.g. 3.74
+        mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf(
+            "SC_PHYS_PAGES"
+        )  # e.g. 4015976448
+        mem_gib = mem_bytes / (1024.0**2)  # e.g. 3.74
         return int(mem_gib)
 
     def get_vcpus(self):
@@ -55,9 +56,7 @@ class VirtualMachine(object):
 
     def get_netbox_vm(self):
         hostname = get_hostname(config)
-        vm = nb.virtualization.virtual_machines.get(
-            name=hostname
-        )
+        vm = nb.virtualization.virtual_machines.get(name=hostname)
         return vm
 
     def get_netbox_cluster(self, name):
@@ -80,13 +79,11 @@ class VirtualMachine(object):
         tenant = self.get_tenant()
         if tenant is None:
             return None
-        nb_tenant = nb.tenancy.tenants.get(
-            slug=self.get_tenant()
-        )
+        nb_tenant = nb.tenancy.tenants.get(slug=self.get_tenant())
         return nb_tenant
 
     def netbox_create_or_update(self, config):
-        logging.debug('It\'s a virtual machine')
+        logging.debug("It's a virtual machine")
         created = False
         updated = 0
 
@@ -97,7 +94,7 @@ class VirtualMachine(object):
         memory = self.get_memory()
         tenant = self.get_netbox_tenant()
         if not vm:
-            logging.debug('Creating Virtual machine..')
+            logging.debug("Creating Virtual machine..")
             cluster = self.get_netbox_cluster(config.virtual.cluster_name)
 
             vm = nb.virtualization.virtual_machines.create(
@@ -107,7 +104,7 @@ class VirtualMachine(object):
                 vcpus=vcpus,
                 memory=memory,
                 tenant=tenant.id if tenant else None,
-                tags=self.tags,
+                tags=[{"name": x} for x in self.tags],
             )
             created = True
 
@@ -121,12 +118,34 @@ class VirtualMachine(object):
             if vm.memory != memory:
                 vm.memory = memory
                 updated += 1
-            if sorted(set(vm.tags)) != sorted(set(self.tags)):
-                vm.tags = self.tags
+
+            vm_tags = sorted(set([x.name for x in vm.tags]))
+            tags = sorted(set(self.tags))
+            if vm_tags != tags:
+                new_tags_ids = [x.id for x in self.nb_tags]
+                if not config.preserve_tags:
+                    vm.tags = new_tags_ids
+                else:
+                    vm_tags_ids = [x.id for x in vm.tags]
+                    vm.tags = sorted(set(new_tags_ids + vm_tags_ids))
                 updated += 1
+
             if vm.platform != self.device_platform:
                 vm.platform = self.device_platform
                 updated += 1
 
         if updated:
             vm.save()
+
+    def print_debug(self):
+        self.network = VirtualNetwork(server=self)
+        print("Cluster:", self.get_netbox_cluster(config.virtual.cluster_name))
+        print("Platform:", self.device_platform)
+        print("VM:", self.get_netbox_vm())
+        print("vCPU:", self.get_vcpus())
+        print("Memory:", f"{self.get_memory()} MB")
+        print(
+            "NIC:",
+        )
+        pprint(self.network.get_network_cards())
+        pass

nix/netifaces2.nix

@@ -0,0 +1,46 @@
+{
+  lib,
+  buildPythonPackage,
+  fetchFromGitHub,
+  cargo,
+  rustPlatform,
+  rustc,
+  typing-extensions,
+}:
+
+buildPythonPackage rec {
+  pname = "netifaces-2";
+  version = "0.0.22";
+  pyproject = true;
+
+  src = fetchFromGitHub {
+    owner = "SamuelYvon";
+    repo = "netifaces-2";
+    rev = "V${version}";
+    hash = "sha256-XO3HWq8FOVzvpbK8mIBOup6hFMnhDpqOK/5bPziPZQ8=";
+  };
+
+  cargoDeps = rustPlatform.fetchCargoTarball {
+    inherit src;
+    name = "${pname}-${version}";
+    hash = "sha256-uoUa6DSBuIV3RrE7svT1TVLxPHdx8BFu/C6mbpRmor0=";
+  };
+
+  build-system = [
+    cargo
+    rustPlatform.cargoSetupHook
+    rustPlatform.maturinBuildHook
+    rustc
+  ];
+
+  dependencies = [ typing-extensions ];
+
+  pythonImportsCheck = [ "netifaces" ];
+
+  meta = {
+    description = "Netifaces reborn";
+    homepage = "https://github.com/SamuelYvon/netifaces-2.git";
+    license = lib.licenses.mit;
+    maintainers = with lib.maintainers; [ ];
+  };
+}

npins/default.nix

@@ -0,0 +1,80 @@
+# Generated by npins. Do not modify; will be overwritten regularly
+let
+  data = builtins.fromJSON (builtins.readFile ./sources.json);
+  version = data.version;
+
+  mkSource =
+    spec:
+    assert spec ? type;
+    let
+      path =
+        if spec.type == "Git" then
+          mkGitSource spec
+        else if spec.type == "GitRelease" then
+          mkGitSource spec
+        else if spec.type == "PyPi" then
+          mkPyPiSource spec
+        else if spec.type == "Channel" then
+          mkChannelSource spec
+        else
+          builtins.throw "Unknown source type ${spec.type}";
+    in
+    spec // { outPath = path; };
+
+  mkGitSource =
+    {
+      repository,
+      revision,
+      url ? null,
+      hash,
+      branch ? null,
+      ...
+    }:
+    assert repository ? type;
+    # At the moment, either it is a plain git repository (which has an url), or it is a GitHub/GitLab repository
+    # In the latter case, there we will always be an url to the tarball
+    if url != null then
+      (builtins.fetchTarball {
+        inherit url;
+        sha256 = hash; # FIXME: check nix version & use SRI hashes
+      })
+    else
+      assert repository.type == "Git";
+      let
+        urlToName =
+          url: rev:
+          let
+            matched = builtins.match "^.*/([^/]*)(\\.git)?$" repository.url;
+
+            short = builtins.substring 0 7 rev;
+
+            appendShort = if (builtins.match "[a-f0-9]*" rev) != null then "-${short}" else "";
+          in
+          "${if matched == null then "source" else builtins.head matched}${appendShort}";
+        name = urlToName repository.url revision;
+      in
+      builtins.fetchGit {
+        url = repository.url;
+        rev = revision;
+        inherit name;
+        # hash = hash;
+      };
+
+  mkPyPiSource =
+    { url, hash, ... }:
+    builtins.fetchurl {
+      inherit url;
+      sha256 = hash;
+    };
+
+  mkChannelSource =
+    { url, hash, ... }:
+    builtins.fetchTarball {
+      inherit url;
+      sha256 = hash;
+    };
+in
+if version == 3 then
+  builtins.mapAttrs (_: mkSource) data.pins
+else
+  throw "Unsupported format version ${toString version} in sources.json. Try running `npins upgrade`"

npins/sources.json

@@ -0,0 +1,23 @@
+{
+  "pins": {
+    "git-hooks": {
+      "type": "Git",
+      "repository": {
+        "type": "GitHub",
+        "owner": "cachix",
+        "repo": "git-hooks.nix"
+      },
+      "branch": "master",
+      "revision": "3c3e88f0f544d6bb54329832616af7eb971b6be6",
+      "url": "https://github.com/cachix/git-hooks.nix/archive/3c3e88f0f544d6bb54329832616af7eb971b6be6.tar.gz",
+      "hash": "04pwjz423iq2nkazkys905gvsm5j39722ngavrnx42b8msr5k555"
+    },
+    "nixpkgs": {
+      "type": "Channel",
+      "name": "nixpkgs-unstable",
+      "url": "https://releases.nixos.org/nixpkgs/nixpkgs-24.11pre694416.ccc0c2126893/nixexprs.tar.xz",
+      "hash": "0cn1z4wzps8nfqxzr6l5mbn81adcqy2cy2ic70z13fhzicmxfsbx"
+    }
+  },
+  "version": 3
+}

pyproject.toml

@@ -0,0 +1,2 @@
+[tool.isort]
+profile = "black"

requirements.txt

@@ -1,6 +1,6 @@
 pynetbox==7.3.4
 netaddr==1.3.0
-netifaces==0.11.0
+netifaces2==0.0.22
 pyyaml==6.0.1
 jsonargparse==4.32.0
 python-slugify==8.0.4

setup.py

@@ -1,42 +1,38 @@
-from setuptools import find_packages, setup
 import os
 
+from setuptools import find_packages, setup
+
+
 def get_requirements():
-    reqs_path = os.path.join(
-        os.path.dirname(__file__),
-        'requirements.txt'
-    )
-    with open(reqs_path, 'r') as f:
-        reqs = [
-            r.strip() for r in f
-            if r.strip()
-        ]
+    reqs_path = os.path.join(os.path.dirname(__file__), "requirements.txt")
+    with open(reqs_path, "r") as f:
+        reqs = [r.strip() for r in f if r.strip()]
     return reqs
 
 
 setup(
-    name='netbox_agent',
-    version='0.7.1',
-    description='NetBox agent for server',
-    long_description=open('README.md', encoding="utf-8").read(),
-    long_description_content_type='text/markdown',
-    url='https://github.com/solvik/netbox_agent',
-    author='Solvik Blum',
-    author_email='solvik@solvik.fr',
-    license='Apache2',
+    name="netbox_agent",
+    version="0.7.1",
+    description="NetBox agent for server",
+    long_description=open("README.md", encoding="utf-8").read(),
+    long_description_content_type="text/markdown",
+    url="https://github.com/solvik/netbox_agent",
+    author="Solvik Blum",
+    author_email="solvik@solvik.fr",
+    license="Apache2",
     include_package_data=True,
     packages=find_packages(exclude=["*.tests", "*.tests.*", "tests.*", "tests"]),
     use_scm_version=True,
     install_requires=get_requirements(),
     zip_safe=False,
-    keywords=['netbox'],
+    keywords=["netbox"],
     classifiers=[
-        'Intended Audience :: Developers',
-        'Development Status :: 5 - Production/Stable',
-        'Programming Language :: Python :: 3',
-        'Programming Language :: Python :: 3.6',
+        "Intended Audience :: Developers",
+        "Development Status :: 5 - Production/Stable",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.6",
     ],
     entry_points={
-        'console_scripts': ['netbox_agent=netbox_agent.cli:main'],
-    }
+        "console_scripts": ["netbox_agent=netbox_agent.cli:main"],
+    },
 )

shell.nix

@@ -0,0 +1 @@
+(import ./. { }).devShell

tests/conftest.py

@@ -14,14 +14,15 @@ def get_fixture_paths(path):
     return fixture_paths
 
 
-def parametrize_with_fixtures(path, base_path='tests/fixtures',
-                              argname='fixture', only_filenames=None):
+def parametrize_with_fixtures(
+    path, base_path="tests/fixtures", argname="fixture", only_filenames=None
+):
     path = os.path.join(base_path, path)
     fixture_paths = get_fixture_paths(path)
     argvalues = []
     for path in fixture_paths:
-        with open(path, 'r') as f:
-            content = ''.join(f.readlines())
+        with open(path, "r") as f:
+            content = "".join(f.readlines())
         filename = os.path.basename(path)
         if only_filenames and filename not in only_filenames:
             continue
@@ -30,4 +31,5 @@ def parametrize_with_fixtures(path, base_path='tests/fixtures',
 
     def _decorator(test_function):
         return pytest.mark.parametrize(argname, argvalues)(test_function)
+
     return _decorator

tests/network.py

@@ -3,18 +3,22 @@ from tests.conftest import parametrize_with_fixtures
 
 
 @parametrize_with_fixtures(
-    'lldp/', only_filenames=[
-        'dedibox1.txt',
-    ])
+    "lldp/",
+    only_filenames=[
+        "dedibox1.txt",
+    ],
+)
 def test_lldp_parse_with_port_desc(fixture):
     lldp = LLDP(fixture)
-    assert lldp.get_switch_port('enp1s0f0') == 'RJ-9'
+    assert lldp.get_switch_port("enp1s0f0") == "RJ-9"
 
 
 @parametrize_with_fixtures(
-    'lldp/', only_filenames=[
-        'qfx.txt',
-    ])
+    "lldp/",
+    only_filenames=[
+        "qfx.txt",
+    ],
+)
 def test_lldp_parse_without_ifname(fixture):
     lldp = LLDP(fixture)
-    assert lldp.get_switch_port('eth0') == 'xe-0/0/1'
+    assert lldp.get_switch_port("eth0") == "xe-0/0/1"

tests/server.py

@@ -6,7 +6,7 @@ from netbox_agent.vendors.supermicro import SupermicroHost
 from tests.conftest import parametrize_with_fixtures
 
 
-@parametrize_with_fixtures('dmidecode/')
+@parametrize_with_fixtures("dmidecode/")
 def test_init(fixture):
     dmi = parse(fixture)
     server = ServerBase(dmi)
@@ -14,96 +14,78 @@ def test_init(fixture):
 
 
 @parametrize_with_fixtures(
-    'dmidecode/', only_filenames=[
-        'HP_SL4540_Gen8',
-        'HP_BL460c_Gen9',
-        'HP_DL380p_Gen8',
-        'HP_SL4540_Gen8'
-        'HP_ProLiant_BL460c_Gen10_Graphics_Exp'
-    ])
+    "dmidecode/",
+    only_filenames=[
+        "HP_SL4540_Gen8",
+        "HP_BL460c_Gen9",
+        "HP_DL380p_Gen8",
+        "HP_SL4540_Gen8" "HP_ProLiant_BL460c_Gen10_Graphics_Exp",
+    ],
+)
 def test_hp_service_tag(fixture):
     dmi = parse(fixture)
     server = HPHost(dmi)
-    assert server.get_service_tag() == '4242'
+    assert server.get_service_tag() == "4242"
 
 
-@parametrize_with_fixtures(
-    'dmidecode/', only_filenames=[
-        'HP_ProLiant_m710x'
-    ])
+@parametrize_with_fixtures("dmidecode/", only_filenames=["HP_ProLiant_m710x"])
 def test_moonshot_blade(fixture):
     dmi = parse(fixture)
     server = HPHost(dmi)
-    assert server.get_service_tag() == 'CN66480BLA'
-    assert server.get_chassis_service_tag() == 'CZ3702MD5K'
+    assert server.get_service_tag() == "CN66480BLA"
+    assert server.get_chassis_service_tag() == "CZ3702MD5K"
     assert server.is_blade() is True
     assert server.own_expansion_slot() is False
 
 
-@parametrize_with_fixtures(
-    'dmidecode/', only_filenames=[
-        'SYS-5039MS-H12TRF-OS012.txt'
-    ])
+@parametrize_with_fixtures("dmidecode/", only_filenames=["SYS-5039MS-H12TRF-OS012.txt"])
 def test_supermicro_blade(fixture):
     dmi = parse(fixture)
     server = SupermicroHost(dmi)
-    assert server.get_service_tag() == 'E235735X6B01665'
-    assert server.get_chassis_service_tag() == 'C9390AF40A20098'
-    assert server.get_chassis() == 'SYS-5039MS-H12TRF-OS012'
+    assert server.get_service_tag() == "E235735X6B01665"
+    assert server.get_chassis_service_tag() == "C9390AF40A20098"
+    assert server.get_chassis() == "SYS-5039MS-H12TRF-OS012"
     assert server.is_blade() is True
 
 
-@parametrize_with_fixtures(
-    'dmidecode/', only_filenames=[
-        'SM_SYS-6018R'
-    ])
+@parametrize_with_fixtures("dmidecode/", only_filenames=["SM_SYS-6018R"])
 def test_supermicro_pizza(fixture):
     dmi = parse(fixture)
     server = SupermicroHost(dmi)
-    assert server.get_service_tag() == 'A177950X7709591'
-    assert server.get_chassis() == 'SYS-6018R-TDTPR'
+    assert server.get_service_tag() == "A177950X7709591"
+    assert server.get_chassis() == "SYS-6018R-TDTPR"
     assert server.is_blade() is False
 
 
-@parametrize_with_fixtures(
-    'dmidecode/', only_filenames=[
-        'QCT_X10E-9N'
-    ])
+@parametrize_with_fixtures("dmidecode/", only_filenames=["QCT_X10E-9N"])
 def test_qct_x10(fixture):
     dmi = parse(fixture)
     server = QCTHost(dmi)
-    assert server.get_service_tag() == 'QTFCQ57140285'
+    assert server.get_service_tag() == "QTFCQ57140285"
 
 
-@parametrize_with_fixtures(
-    'dmidecode/', only_filenames=[
-        'unknown.txt'
-    ])
+@parametrize_with_fixtures("dmidecode/", only_filenames=["unknown.txt"])
 def test_generic_host_service_tag(fixture):
     dmi = parse(fixture)
     server = ServerBase(dmi)
-    assert server.get_service_tag() == '42'
+    assert server.get_service_tag() == "42"
 
 
-@parametrize_with_fixtures(
-    'dmidecode/', only_filenames=[
-        'unknown.txt'
-    ])
+@parametrize_with_fixtures("dmidecode/", only_filenames=["unknown.txt"])
 def test_generic_host_product_name(fixture):
     dmi = parse(fixture)
     server = ServerBase(dmi)
-    assert server.get_product_name() == 'SR'
+    assert server.get_product_name() == "SR"
 
 
 @parametrize_with_fixtures(
-    'dmidecode/', only_filenames=[
-        'HP_ProLiant_BL460c_Gen10_Graphics_Exp'
-    ])
+    "dmidecode/", only_filenames=["HP_ProLiant_BL460c_Gen10_Graphics_Exp"]
+)
 def test_hp_blade_with_gpu_expansion(fixture):
     dmi = parse(fixture)
     server = HPHost(dmi)
-    assert server.get_service_tag() == '4242'
-    assert server.get_chassis_service_tag() == '4343'
+    assert server.get_service_tag() == "4242"
+    assert server.get_chassis_service_tag() == "4343"
     assert server.is_blade() is True
     assert server.own_expansion_slot() is True
-    assert server.get_expansion_service_tag() == '4242 expansion'
+    assert server.get_expansion_service_tag() == "4242 expansion"