Support Platform Mapping in Settings

.bandit.yml

@@ -0,0 +1,2 @@
+---
+skips: []

README.md

@@ -53,6 +53,12 @@ The plugin behavior can be controlled with the following list of settings
 - `default_device_role_color` string (default FF0000), color assigned to the device role if it needs to be created.
 - `default_management_interface` string (default "PLACEHOLDER"), name of the management interface that will be created, if one can't be identified on the device.
 - `default_management_prefix_length` integer ( default 0), length of the prefix that will be used for the management IP address, if the IP can't be found.
+- `platform_map` (dictionary), mapping of an **auto-detected** Netmiko platform to the **NetBox slug** name of your Platform. The dictionary should be in the format:
+    ```python
+    {
+      <Netmiko Platform>: <NetBox Slug> 
+    }
+    ```
 
 ## Usage
 

development/base_configuration.py

@@ -155,6 +155,7 @@
 # Enable installed plugins. Add the name of each plugin to the list.
 PLUGINS = ["netbox_onboarding"]
 
+PLUGINS_CONFIG = {"netbox_onboarding": {}}
 # Plugins configuration settings. These settings are used by various plugins that the user may have installed.
 # Each key in the dictionary is the name of an installed plugin and its value is a dictionary of settings.
 # PLUGINS_CONFIG = {}

development/docker-compose.yml

@@ -19,7 +19,7 @@ services:
     volumes:
       - ./base_configuration.py:/opt/netbox/netbox/netbox/base_configuration.py
       - ./netbox_${NETBOX_VER}/configuration.py:/opt/netbox/netbox/netbox/configuration.py
-      - ../netbox_onboarding:/source/netbox_onboarding
+      - ../:/source
     tty: true
   worker:
     build:

netbox_onboarding/__init__.py

@@ -39,6 +39,7 @@ class OnboardingConfig(PluginConfig):
         "default_management_prefix_length": 0,
         "default_device_status": "active",
         "create_management_interface_if_missing": True,
+        "platform_map": {},
     }
     caching_config = {}
 

netbox_onboarding/onboard.py

@@ -125,7 +125,25 @@ def check_reachability(self):
             raise OnboardException(reason="fail-connect", message=f"ERROR device unreachable: {ip_addr}:{port}")
 
     @staticmethod
-    def guess_netmiko_device_type(**kwargs):
+    def check_netmiko_conversion(guessed_device_type, platform_map=None):
+        """Method to convert Netmiko device type into the mapped type if defined in the settings file.
+
+        Args:
+            guessed_device_type (string): Netmiko device type guessed platform
+            test_platform_map (dict): Platform Map for use in testing
+
+        Returns:
+            string: Platform name
+        """
+        # If this is defined, process the mapping
+        if platform_map:
+            # Attempt to get a mapped slug. If there is no slug, return the guessed_device_type as the slug
+            return platform_map.get(guessed_device_type, guessed_device_type)
+
+        # There is no mapping configured, return what was brought in
+        return guessed_device_type
+
+    def guess_netmiko_device_type(self, **kwargs):
         """Guess the device type of host, based on Netmiko."""
         guessed_device_type = None
 
@@ -157,7 +175,8 @@ def guess_netmiko_device_type(**kwargs):
 
         logging.info("INFO device type is %s", guessed_device_type)
 
-        return guessed_device_type
+        # Get the platform map from the PLUGIN SETTINGS, Return the result of doing a check_netmiko_conversion
+        return self.check_netmiko_conversion(guessed_device_type, platform_map=PLUGIN_SETTINGS.get("platform_map", {}))
 
     def get_platform_slug(self):
         """Get platform slug in netmiko format (ie cisco_ios, cisco_xr etc)."""
@@ -173,17 +192,30 @@ def get_platform_slug(self):
         return platform_slug
 
     @staticmethod
-    def get_platform_object_from_netbox(platform_slug):
+    def get_platform_object_from_netbox(
+        platform_slug, create_platform_if_missing=PLUGIN_SETTINGS["create_platform_if_missing"]
+    ):
         """Get platform object from NetBox filtered by platform_slug.
 
+        Args:
+            platform_slug (string): slug of a platform object present in NetBox, object will be created if not present
+            and create_platform_if_missing is enabled
+
+        Return:
+            dcim.models.Platform object
+
+        Raises:
+            OnboardException
+
         Lookup is performed based on the object's slug field (not the name field)
         """
         try:
+            # Get the platform from the NetBox DB
             platform = Platform.objects.get(slug=platform_slug)
             logging.info("PLATFORM: found in NetBox %s", platform_slug)
         except Platform.DoesNotExist:
 
-            if not PLUGIN_SETTINGS["create_platform_if_missing"]:
+            if not create_platform_if_missing:
                 raise OnboardException(
                     reason="fail-general", message=f"ERROR platform not found in NetBox: {platform_slug}"
                 )
@@ -199,6 +231,12 @@ def get_platform_object_from_netbox(platform_slug):
             )
             platform.save()
 
+        else:
+            if not platform.napalm_driver:
+                raise OnboardException(
+                    reason="fail-general", message=f"ERROR platform is missing the NAPALM Driver: {platform_slug}",
+                )
+
         return platform
 
     def check_ip(self):

netbox_onboarding/tests/test_onboard.py → netbox_onboarding/tests/test_netbox_keeper.py

@@ -32,6 +32,7 @@ def setUp(self):
 
         self.manufacturer1 = Manufacturer.objects.create(name="Juniper", slug="juniper")
         self.platform1 = Platform.objects.create(name="JunOS", slug="junos")
+        self.platform2 = Platform.objects.create(name="Cisco NX-OS", slug="cisco-nx-os")
         self.device_type1 = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=self.manufacturer1)
         self.device_role1 = DeviceRole.objects.create(name="Firewall", slug="firewall")
 
@@ -45,10 +46,12 @@ def setUp(self):
         self.onboarding_task4 = OnboardingTask.objects.create(
             ip_address="ntc123.local", site=self.site1, role=self.device_role1, platform=self.platform1
         )
-
         self.onboarding_task5 = OnboardingTask.objects.create(
             ip_address="bad.local", site=self.site1, role=self.device_role1, platform=self.platform1
         )
+        self.onboarding_task6 = OnboardingTask.objects.create(
+            ip_address="192.0.2.2", site=self.site1, role=self.device_role1, platform=self.platform2
+        )
         self.onboarding_task7 = OnboardingTask.objects.create(
             ip_address="192.0.2.1/32", site=self.site1, role=self.device_role1, platform=self.platform1
         )
@@ -229,3 +232,32 @@ def test_failed_check_ip(self, mock_get_hostbyname):
             ndk7.check_ip()
             self.assertEqual(exc_info.exception.reason, "fail-prefix")
             self.assertEqual(exc_info.exception.message, "ERROR appears a prefix was entered: 192.0.2.1/32")
+
+    def test_platform_map(self):
+        """Verify platform mapping of netmiko to slug functionality."""
+        # Create static mapping
+        platform_map = {"cisco_ios": "ios", "arista_eos": "eos", "cisco_nxos": "cisco-nxos"}
+
+        # Generate an instance of a Cisco IOS device with the mapping defined
+        self.ndk1 = NetdevKeeper(self.onboarding_task1)
+
+        #
+        # Test positive assertions
+        #
+
+        # Test Cisco_ios
+        self.assertEqual(self.ndk1.check_netmiko_conversion("cisco_ios", platform_map=platform_map), "ios")
+        # Test Arista EOS
+        self.assertEqual(self.ndk1.check_netmiko_conversion("arista_eos", platform_map=platform_map), "eos")
+        # Test cisco_nxos
+        self.assertEqual(self.ndk1.check_netmiko_conversion("cisco_nxos", platform_map=platform_map), "cisco-nxos")
+
+        #
+        # Test Negative assertion
+        #
+
+        # Test a non-converting item
+        self.assertEqual(
+            self.ndk1.check_netmiko_conversion("cisco-device-platform", platform_map=platform_map),
+            "cisco-device-platform",
+        )

netbox_onboarding/tests/test_netdev_keeper.py

@@ -0,0 +1,58 @@
+"""Unit tests for netbox_onboarding.onboard module and its classes.
+
+(c) 2020 Network To Code
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+  http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+from django.test import TestCase
+
+from dcim.models import Platform
+from netbox_onboarding.onboard import NetdevKeeper, OnboardException
+
+
+class NetdevKeeperTestCase(TestCase):
+    """Test the NetdevKeeper Class."""
+
+    def setUp(self):
+        """Create a superuser and token for API calls."""
+        self.platform1 = Platform.objects.create(name="JunOS", slug="junos", napalm_driver="junos")
+        self.platform2 = Platform.objects.create(name="Cisco NX-OS", slug="cisco-nx-os")
+
+    def test_get_platform_object_from_netbox(self):
+        """Test of platform object from netbox."""
+        # Test assigning platform
+        platform = NetdevKeeper.get_platform_object_from_netbox("junos", create_platform_if_missing=False)
+        self.assertIsInstance(platform, Platform)
+
+        # Test creation of missing platform object
+        platform = NetdevKeeper.get_platform_object_from_netbox("arista_eos", create_platform_if_missing=True)
+        self.assertIsInstance(platform, Platform)
+        self.assertEqual(platform.napalm_driver, "eos")
+
+        # Test failed unable to find the device and not part of the NETMIKO TO NAPALM keys
+        with self.assertRaises(OnboardException) as exc_info:
+            platform = NetdevKeeper.get_platform_object_from_netbox("notthere", create_platform_if_missing=True)
+            self.assertEqual(
+                exc_info.exception.message,
+                "ERROR platform not found in NetBox and it's eligible for auto-creation: notthere",
+            )
+            self.assertEqual(exc_info.exception.reason, "fail-general")
+
+        # Test searching for an object, does not exist, but create_platform is false
+        with self.assertRaises(OnboardException) as exc_info:
+            platform = NetdevKeeper.get_platform_object_from_netbox("cisco_ios", create_platform_if_missing=False)
+            self.assertEqual(exc_info.exception.message, "ERROR platform not found in NetBox: cisco_ios")
+            self.assertEqual(exc_info.exception.reason, "fail-general")
+
+        # Test NAPALM Driver not defined in NetBox
+        with self.assertRaises(OnboardException) as exc_info:
+            platform = NetdevKeeper.get_platform_object_from_netbox("cisco-nx-os", create_platform_if_missing=False)
+            self.assertEqual(exc_info.exception.message, "ERROR platform is missing the NAPALM Driver: cisco-nx-os")
+            self.assertEqual(exc_info.exception.reason, "fail-general")

tasks.py

@@ -283,7 +283,7 @@ def bandit(context, netbox_ver=NETBOX_VER, python_ver=PYTHON_VER):
     """
     docker = f"docker-compose -f {COMPOSE_FILE} -p {BUILD_NAME} run netbox"
     context.run(
-        f'{docker} sh -c "cd /source && bandit --recursive ./"',
+        f'{docker} sh -c "cd /source && bandit --configfile .bandit.yml --recursive ./"',
         env={"NETBOX_VER": netbox_ver, "PYTHON_VER": python_ver},
         pty=True,
     )