Changes for multiple criteria searching

Author: carbonarok

README.md

@@ -80,6 +80,10 @@ The plugin behavior can be controlled with the following list of settings
       <Napalm Driver Name>: <Loadable Python Module>
     }
     ```
+- `object_match_strategy` (string), defines the method for searching models. There are
+currently two strategies, strict and loose. Strict has to be a direct match, normally 
+using a slug. Loose allows a range of search criteria to match a single object. If multiple
+objects are returned an error is raised. 
 
 ## Usage
 

netbox_onboarding/__init__.py

@@ -43,6 +43,7 @@ class OnboardingConfig(PluginConfig):
         "skip_manufacturer_on_update": False,
         "platform_map": {},
         "onboarding_extensions_map": {"ios": "netbox_onboarding.onboarding_extensions.ios",},
+        "object_match_strategy": "loose",
     }
     caching_config = {}
 

netbox_onboarding/netbox_keeper.py

@@ -30,6 +30,43 @@
 PLUGIN_SETTINGS = settings.PLUGINS_CONFIG["netbox_onboarding"]
 
 
+def object_match(obj, search_array):
+    """Used to search models for multiple criteria.
+
+    Inputs:
+        obj:            The model used for searching.
+        search_array:   Nested dictionaries used to search models. First criteria will be used
+                        for strict searching. Loose searching will loop through the search_array
+                        until it finds a match. Example below.
+                        [
+                            {"slug__iexact": 'switch1'},
+                            {"model__iexact": 'Cisco'}
+                        ]
+    """
+    try:
+        result = obj.objects.get(**search_array[0])
+        return result
+    except obj.DoesNotExist:
+        if PLUGIN_SETTINGS["object_match_strategy"] == "loose":
+            for search_array_element in search_array[1:]:
+                try:
+                    result = obj.objects.get(**search_array_element)
+                    return result
+                except obj.DoesNotExist:
+                    pass
+                except obj.MultipleObjectsReturned:
+                    raise OnboardException(
+                        reason="fail-general",
+                        message=f"ERROR multiple objects found in {str(obj)} searching on {str(search_array_element)})",
+                    )
+        raise
+    except obj.MultipleObjectsReturned:
+        raise OnboardException(
+            reason="fail-general",
+            message=f"ERROR multiple objects found in {str(obj)} searching on {str(search_array_element)})",
+        )
+
+
 class NetboxKeeper:
     """Used to manage the information relating to the network device within the NetBox server."""
 
@@ -147,7 +184,8 @@ def ensure_device_manufacturer(
         nb_manufacturer_slug = slugify(self.netdev_vendor)
 
         try:
-            self.nb_manufacturer = Manufacturer.objects.get(slug=nb_manufacturer_slug)
+            search_array = [{"slug__iexact": nb_manufacturer_slug}]
+            self.nb_manufacturer = object_match(Manufacturer, search_array)
         except Manufacturer.DoesNotExist:
             if create_manufacturer:
                 self.nb_manufacturer = Manufacturer.objects.create(name=self.netdev_vendor, slug=nb_manufacturer_slug)
@@ -201,7 +239,13 @@ def ensure_device_type(
         nb_device_type_slug = slugify(nb_device_type_text)
 
         try:
-            self.nb_device_type = DeviceType.objects.get(slug=nb_device_type_slug)
+            search_array = [
+                {"slug__iexact": nb_device_type_slug},
+                {"model__iexact": self.netdev_model},
+                {"part_number__iexact": self.netdev_model},
+            ]
+
+            self.nb_device_type = object_match(DeviceType, search_array)
 
             if self.nb_device_type.manufacturer.id != self.nb_manufacturer.id:
                 raise OnboardException(

netbox_onboarding/tests/test_netbox_keeper.py

@@ -31,8 +31,9 @@ def setUp(self):
         """Create a superuser and token for API calls."""
         self.site1 = Site.objects.create(name="USWEST", slug="uswest")
 
-    def test_ensure_device_manufacturer_missing(self):
+    def test_ensure_device_manufacturer_strict_missing(self):
         """Verify ensure_device_manufacturer function when Manufacturer object is not present."""
+        PLUGIN_SETTINGS["object_match_strategy"] = "strict"
         onboarding_kwargs = {
             "netdev_hostname": "device1",
             "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"],
@@ -52,8 +53,54 @@ def test_ensure_device_manufacturer_missing(self):
         self.assertIsInstance(nbk.nb_manufacturer, Manufacturer)
         self.assertEqual(nbk.nb_manufacturer.slug, slugify(onboarding_kwargs["netdev_vendor"]))
 
-    def test_ensure_device_type_missing(self):
+    def test_ensure_device_manufacturer_loose_missing(self):
+        """Verify ensure_device_manufacturer function when Manufacturer object is not present."""
+        PLUGIN_SETTINGS["object_match_strategy"] = "loose"
+        onboarding_kwargs = {
+            "netdev_hostname": "device1",
+            "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"],
+            "netdev_vendor": "Cisco",
+            "netdev_model": "CSR1000v",
+            "netdev_nb_site_slug": self.site1.slug,
+        }
+
+        nbk = NetboxKeeper(**onboarding_kwargs)
+
+        with self.assertRaises(OnboardException) as exc_info:
+            nbk.ensure_device_manufacturer(create_manufacturer=False)
+            self.assertEqual(exc_info.exception.message, "ERROR manufacturer not found: Cisco")
+            self.assertEqual(exc_info.exception.reason, "fail-config")
+
+        nbk.ensure_device_manufacturer(create_manufacturer=True)
+        self.assertIsInstance(nbk.nb_manufacturer, Manufacturer)
+        self.assertEqual(nbk.nb_manufacturer.slug, slugify(onboarding_kwargs["netdev_vendor"]))
+
+    def test_ensure_device_type_strict_missing(self):
+        """Verify ensure_device_type function when DeviceType object is not present."""
+        PLUGIN_SETTINGS["object_match_strategy"] = "strict"
+        onboarding_kwargs = {
+            "netdev_hostname": "device1",
+            "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"],
+            "netdev_vendor": "Cisco",
+            "netdev_model": "CSR1000v",
+            "netdev_nb_site_slug": self.site1.slug,
+        }
+
+        nbk = NetboxKeeper(**onboarding_kwargs)
+        nbk.nb_manufacturer = Manufacturer.objects.create(name="Cisco", slug="cisco")
+
+        with self.assertRaises(OnboardException) as exc_info:
+            nbk.ensure_device_type(create_device_type=False)
+            self.assertEqual(exc_info.exception.message, "ERROR device type not found: CSR1000v")
+            self.assertEqual(exc_info.exception.reason, "fail-config")
+
+        nbk.ensure_device_type(create_device_type=True)
+        self.assertIsInstance(nbk.nb_device_type, DeviceType)
+        self.assertEqual(nbk.nb_device_type.slug, slugify(onboarding_kwargs["netdev_model"]))
+
+    def test_ensure_device_type_loose_missing(self):
         """Verify ensure_device_type function when DeviceType object is not present."""
+        PLUGIN_SETTINGS["object_match_strategy"] = "loose"
         onboarding_kwargs = {
             "netdev_hostname": "device1",
             "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"],
@@ -74,8 +121,30 @@ def test_ensure_device_type_missing(self):
         self.assertIsInstance(nbk.nb_device_type, DeviceType)
         self.assertEqual(nbk.nb_device_type.slug, slugify(onboarding_kwargs["netdev_model"]))
 
-    def test_ensure_device_type_present(self):
+    def test_ensure_device_type_strict_present(self):
+        """Verify ensure_device_type function when DeviceType object is already present."""
+        PLUGIN_SETTINGS["object_match_strategy"] = "strict"
+        manufacturer = Manufacturer.objects.create(name="Juniper", slug="juniper")
+
+        device_type = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=manufacturer)
+
+        onboarding_kwargs = {
+            "netdev_hostname": "device2",
+            "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"],
+            "netdev_vendor": "Juniper",
+            "netdev_nb_device_type_slug": device_type.slug,
+            "netdev_nb_site_slug": self.site1.slug,
+        }
+
+        nbk = NetboxKeeper(**onboarding_kwargs)
+        nbk.nb_manufacturer = manufacturer
+
+        nbk.ensure_device_type(create_device_type=False)
+        self.assertEqual(nbk.nb_device_type, device_type)
+
+    def test_ensure_device_type_loose_present(self):
         """Verify ensure_device_type function when DeviceType object is already present."""
+        PLUGIN_SETTINGS["object_match_strategy"] = "loose"
         manufacturer = Manufacturer.objects.create(name="Juniper", slug="juniper")
 
         device_type = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=manufacturer)