rewrited to use with netbox 3.5.9 and higher

.gitignore

@@ -59,6 +59,10 @@ coverage.xml
 local_settings.py
 db.sqlite3
 
+#Configs 
+device.yaml
+filter.yaml
+
 # Flask stuff:
 instance/
 .webassets-cache

.travis.yml

@@ -10,6 +10,8 @@ matrix:
       env: TOXENV=py37
     - python: 3.8-dev
       env: TOXENV=py38
+    - python: 3.10-dev
+      env: TOXENV=py310
 
 sudo: required
 

netbox_netprod_importer/importer.py

@@ -10,6 +10,7 @@
 from netbox_netprod_importer.vendors import DeviceParsers, StubParser
 from netbox_netprod_importer.tools import is_macaddr
 
+
 logger = logging.getLogger("netbox_importer")
 
 
@@ -156,6 +157,7 @@ def get_interfaces(self):
                     "access": "Access",
                     "trunk": "Tagged",
                     "static access": "Access",
+                    "trunk all": "Tagged (All)",
                     None: None
                 }[self.specific_parser.get_interface_mode(ifname)]
             except NotImplementedError:
@@ -175,7 +177,8 @@ def get_interfaces(self):
                 "type": _type,
                 "mode": mode,
                 "untagged_vlan": None,
-                "tagged_vlans": [],
+                "tagged_vlans": [""],
+                "custom_fields": {"voice_vlan" : None,},
             }
 
             try:
@@ -189,27 +192,40 @@ def get_interfaces(self):
                 try:
                     interfaces[ifname]["untagged_vlan"] = \
                         self.specific_parser.get_interface_access_vlan(ifname)
+                    interfaces[ifname]["custom_fields"]["voice_vlan"] = \
+                        self.specific_parser.get_interface_voice_vlan(ifname)                    
                 except NotImplementedError:
                     interfaces[ifname]["untagged_vlan"] = None
-
-            try:
-                vlans = self.specific_parser.get_interface_vlans(ifname)
-            except NotImplementedError:
-                vlans = None
-
-            if vlans:
-                interfaces[ifname]["tagged_vlans"] = vlans
+                    interfaces[ifname]["custom_fields"]["voice_vlan"] = None
+            if mode == "Tagged":
+                try:
+                    if self.specific_parser.get_interface_status(ifname)=="down":
+                        vlans = self.specific_parser.get_interface_tagged_vlans(ifname)
+                    else:
+                        vlans = self.specific_parser.get_interface_vlans(ifname)
+                except NotImplementedError:
+                    vlans = None
+                if vlans:
+                    interfaces[ifname]["tagged_vlans"] = vlans
 
         for ifname, data in interfaces.items():
             if data["mode"] == "Tagged":
                 try:
-                    native = self.specific_parser.get_interface_native_vlan(ifname)
+                    native = self.specific_parser.get_interface_native_vlan(ifname)                
                 except NotImplementedError:
                     native = None
                 if native in data["tagged_vlans"]:
-                    interfaces[ifname]["untagged_vlan"] = native
-                    interfaces[ifname]["tagged_vlans"].pop(native)
-
+                    interfaces[ifname]["untagged_vlan"] = int(native)
+                    try:
+                        interfaces[ifname]["tagged_vlans"].pop(int(native))
+                    except IndexError:
+                        interfaces[ifname]["tagged_vlans"].pop(-1)
+            if data["mode"] == "Tagged (All)":
+                try:
+                    native = int(self.specific_parser.get_interface_native_vlan(ifname))
+                except NotImplementedError:
+                    native = None
+                interfaces[ifname]["untagged_vlan"] = native            
         for trunk in trunks:
             if trunk in interfaces:
                 interfaces[trunk]["mode"] = "Tagged"

netbox_netprod_importer/push.py

@@ -5,10 +5,11 @@
 import logging
 from requests.exceptions import HTTPError
 import threading
-
 import cachetools
 from netboxapi import NetboxMapper
 from tqdm import tqdm
+import json
+from rich import print as rprint
 
 from netbox_netprod_importer.vendors.cisco import CiscoParser
 from netbox_netprod_importer.vendors.juniper import JuniperParser
@@ -30,9 +31,7 @@ def __init__(self, netbox_api, *args, **kwargs):
         self.netbox_api = netbox_api
 
         self._mappers = {
-            "dcim_choices": NetboxMapper(
-                self.netbox_api, app_name="dcim", model="_choices"
-            ), "devices": NetboxMapper(
+            "devices": NetboxMapper(
                 self.netbox_api, app_name="dcim", model="devices"
             ), "interfaces": NetboxMapper(
                 self.netbox_api, app_name="dcim", model="interfaces"
@@ -44,25 +43,25 @@ def __init__(self, netbox_api, *args, **kwargs):
                 self.netbox_api, app_name="ipam", model="vlans"
             )
         }
-        self._choices_cache = {}
+        self._options_cache = {}
 
     @abstractmethod
     def push(self):
         pass
 
-    def search_value_in_choices(self, mapper_name, id, label):
-        if mapper_name not in self._choices_cache:
+    def search_value_in_options(self, mapper_name, id, label):
+        if mapper_name not in self._options_cache:
             try:
                 mapper = self._mappers[mapper_name]
-                self._choices_cache[mapper_name] = next(mapper.get())
+                options = mapper.options()
+                self._options_cache[mapper_name] = options["actions"]["POST"]
             except StopIteration:
                 pass
-
-        for choice in self._choices_cache[mapper_name][id]:
-            if choice["label"] == label:
+        for choice in self._options_cache[mapper_name][id]['choices']:
+            if choice["display_name"] == label:
                 return choice["value"]
 
-        raise KeyError("Label {} not in choices".format(label))
+        raise KeyError("Label {} not in options".format(label))
 
 
 class NetboxDevicePropsPusher(_NetboxPusher):
@@ -116,8 +115,8 @@ def _push_interfaces(self):
 
         for if_name, if_prop in interfaces_props.items():
             if_prop = if_prop.copy()
-            if_prop["type"] = self.search_value_in_choices(
-                "dcim_choices", "interface:type", if_prop["type"]
+            if_prop["type"] = self.search_value_in_options(
+                "interfaces", "type", if_prop["type"]
             )
             interface_query = self._mappers["interfaces"].get(
                 device_id=self._device, name=if_name
@@ -136,20 +135,27 @@ def _push_interfaces(self):
             if if_prop.get("lag"):
                 interfaces_lag[if_name] = if_prop.pop("lag")
 
-            if not self.overwrite and "mode" in if_prop:
-                if_prop.pop("mode")
+            if not self.overwrite and "mode" in if_prop["mode"]:
+                self._handle_interface_mode(interface, if_prop["mode"])
+                if_prop.pop("mode")                
             elif if_prop.get("mode"):
                 # cannot really guess (yet) the interface mode, so only set it
                 # if overwrite
                 self._handle_interface_mode(interface, if_prop["mode"])
                 if_prop.pop("mode")
 
-            if if_prop["untagged_vlan"]:
+            if if_prop["untagged_vlan"]:                
                 vlan_id = self._get_vlan_id(if_prop["untagged_vlan"])
                 if vlan_id != -1:
                     setattr(interface, "untagged_vlan", vlan_id)
 
             if_prop.pop("untagged_vlan")
+
+            if if_prop["custom_fields"]["voice_vlan"]:                
+                vlan_id = self._get_vlan_id(if_prop["custom_fields"]["voice_vlan"])
+                if vlan_id != -1:                    
+                    if_prop["custom_fields"]["voice_vlan"]=vlan_id                    
+
             if len(if_prop["tagged_vlans"]):
                 setattr(interface, "tagged_vlans", [])
                 for vlan in if_prop["tagged_vlans"]:
@@ -158,7 +164,7 @@ def _push_interfaces(self):
                         interface.tagged_vlans.append(vlan_id)
             if_prop.pop("tagged_vlans")
 
-            for k, v in if_prop.items():
+            for k, v in if_prop.items():               
                 setattr(interface, k, v)
 
             try:
@@ -204,8 +210,8 @@ def _get_vlan_id(self, vlan):
 
 
     def _handle_interface_mode(self, netbox_if, mode):
-        netbox_mode = self.search_value_in_choices(
-            "dcim_choices", "interface:mode",
+        netbox_mode = self.search_value_in_options(
+            "interfaces", "mode",
             mode
         )
 
@@ -217,7 +223,7 @@ def _attach_interface_to_ip_addresses(self, netbox_if, *ip_addresses):
         addresses = []
         for ip in ip_addresses:
             # check if ip attached isn't already correct
-            try:
+            try:                
                 ip_netbox_obj = next(mapper.get(q=ip, interface_id=netbox_if))
             except StopIteration:
                 try:
@@ -232,7 +238,7 @@ def _attach_interface_to_ip_addresses(self, netbox_if, *ip_addresses):
                 ip_netbox_obj.interface = netbox_if
                 try:
                     ip_netbox_obj.put()
-                except HTTPError as e:
+                except HTTPError as e:                    
                     raise IPPushingError(ip_netbox_obj.address, e)
 
             addresses.append(ip_netbox_obj)

netbox_netprod_importer/vendors/base.py

@@ -34,6 +34,18 @@ def get_interface_vlans(self, interface):
     def get_interface_native_vlan(self, interface):
         pass
 
+    @abstractmethod
+    def get_interface_voice_vlan(self, interface):
+        pass
+
+    @abstractmethod
+    def get_interface_tagged_vlans(self, interface):
+        pass
+
+    @abstractmethod
+    def get_interface_status(self, interface):
+        pass
+
     def get_all_derivatives_for_netif(self, interface):
         """
         Get all possible derivatives for an interface name

netbox_netprod_importer/vendors/cisco/ios.py

@@ -7,7 +7,6 @@
 from .constants import InterfacesRegex
 from .base import CiscoParser
 
-
 logger = logging.getLogger("netbox_importer")
 
 
@@ -127,6 +126,7 @@ def get_interface_access_vlan(self, interface):
             logger.debug("Switch %s, show interface switchport cmd error",
                          self.device.hostname)
         return None
+
 
     def get_interface_native_vlan(self, interface):
         from pynxos.errors import CLIError
@@ -137,42 +137,83 @@ def get_interface_native_vlan(self, interface):
             logger.debug("Switch %s, show interface switchport cmd error",
                          self.device.hostname)
         return None
-
+
+    def get_interface_voice_vlan(self, interface):
+        from pynxos.errors import CLIError
+        try:
+            return self._get_interfaces_mode()[
+                self.get_abrev_if(interface)].get("voice_vlan")
+        except (KeyError, CLIError):
+            logger.debug("Switch %s, show interface switchport cmd error",
+                         self.device.hostname)
+        return None
+
+    def get_interface_tagged_vlans(self, interface):
+        from pynxos.errors import CLIError
+        try:
+            return self._get_interfaces_mode()[
+                self.get_abrev_if(interface)].get("tagged_vlans")
+        except (KeyError, CLIError):
+            logger.debug("Switch %s, show interface switchport cmd error",
+                         self.device.hostname)
+        return None
+
+    def get_interface_status(self, interface):
+        from pynxos.errors import CLIError
+        try:
+            return self._get_interfaces_mode()[
+                self.get_abrev_if(interface)].get("oper_status")
+        except (KeyError, CLIError):
+            logger.debug("Switch %s, show interface switchport cmd error",
+                         self.device.hostname)
+        return None
+
     def _get_interfaces_mode(self):
         cmd = "show interface switchport"
 
         if not self.cache.get("mode"):
+
             mode_conf_dump = self.device.cli([cmd])[cmd]
-            mode_conf_lines = re.split(r"(^Name: \S+$)", mode_conf_dump, flags=re.M)
+            mode_conf_lines = re.split(r"(^Name: \S+$)", mode_conf_dump, flags=re.MULTILINE)
             mode_conf_lines.pop(0)
             if len(mode_conf_lines) % 2 != 0:
                 raise ValueError("Unexpected output data in '{}':\n\n{}".format(
                     cmd, mode_conf_lines
                 ))
-            mode_conf_iter = iter(mode_conf_lines)
+            mode_conf_iter = iter(mode_conf_lines)                    
             try:
                 new_mode = [line + next(mode_conf_iter, "") for line in mode_conf_iter]
             except TypeError:
                 raise ValueError()
-
+           
             interfaces_mode = {}
             for entry in new_mode:
                 grp = [
                     r"^Name:\s+(?P<interface>\S+)",
                     r"^Administrative Mode:\s+(?P<oper_mode>static access|trunk|access)",
+                    r"^Operational Mode:\s+(?P<oper_status>down)",
                     r"^Access Mode VLAN:\s+(?P<access_vlan>\d+)",
-                    r"^Trunking Native Mode VLAN:\s+(?P<native_valn>\d+)"
+                    r"^Trunking Native Mode VLAN:\s+(?P<native_vlan>\d+)",                    
+                    r"^Voice VLAN:\s+(?P<voice_vlan>\d+)",
+                    r"^Trunking VLANs Enabled:\s+(?P<tagged_vlans>ALL|.*(.*?),+.*)",
                 ]
                 inf_mode = {}
                 for g in grp:
                     find = re.search(g, entry, re.MULTILINE)
                     if find:
                         inf_mode[find.lastgroup] = find.group(find.lastgroup)
                 if inf_mode.get("interface"):
+                    if not inf_mode.get("voice_vlan"):
+                        inf_mode.update({'voice_vlan' : None})
+                    if inf_mode.get("oper_mode") == "trunk" and inf_mode.get("tagged_vlans") == "ALL":
+                        inf_mode.update({'oper_mode' : 'trunk all'})                                            
+                    if not inf_mode.get("tagged_vlans") == "ALL" and not inf_mode.get("tagged_vlans") == "None":
+                        tagged_vlans = inf_mode.get("tagged_vlans")                        
+                        inf_mode.update({'tagged_vlans' : tagged_vlans.split(",")})
+                    else:
+                        inf_mode.update({'tagged_vlans':''})
                     interfaces_mode[inf_mode["interface"]] = inf_mode
-
             self.cache["mode"] = interfaces_mode
-
         return self.cache["mode"]
 
     def get_vlans(self):

setup.py

@@ -25,7 +25,9 @@
         "Programming Language :: Python :: 3",
         "Programming Language :: Python :: 3.5",
         "Programming Language :: Python :: 3.6",
-        "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.7",        
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.12",
     ],
     description="Microservice import into netbox devices in production",
     install_requires=requirements,
@@ -40,8 +42,8 @@
     setup_requires=setup_requirements,
     test_suite="tests",
     tests_require=test_requirements,
-    url="https://gitlab.infra.online.net/network/netbox_netprod_importer",
-    version="0.3.3",
+    url="https://github.com/Malustta/netbox-netprod-importer",
+    version="0.3.4",
     zip_safe=False,
     entry_points={
         'console_scripts': [

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = py36, py37, py38
+envlist = py36, py37, py38, p310
 
 [testenv]
 deps =