Add single network view and IP Address type support

development/nautobot_config.py

@@ -230,6 +230,7 @@
         "infoblox_username": os.getenv("NAUTOBOT_SSOT_INFOBLOX_USERNAME"),
         "infoblox_verify_ssl": is_truthy(os.getenv("NAUTOBOT_SSOT_INFOBLOX_VERIFY_SSL", True)),
         "infoblox_wapi_version": os.getenv("NAUTOBOT_SSOT_INFOBLOX_WAPI_VERSION", "v2.12"),
+        "infoblox_network_view": os.getenv("NAUTOBOT_SSOT_INFOBLOX_NETWORK_VIEW", ""),
         "ipfabric_api_token": os.getenv("NAUTOBOT_SSOT_IPFABRIC_API_TOKEN"),
         "ipfabric_host": os.getenv("NAUTOBOT_SSOT_IPFABRIC_HOST"),
         "ipfabric_ssl_verify": is_truthy(os.getenv("NAUTOBOT_SSOT_IPFABRIC_SSL_VERIFY", "False")),

docs/admin/integrations/infoblox_setup.md

@@ -29,6 +29,7 @@ Integration behavior can be controlled with the following settings:
 | infoblox_import_objects_vlan_views         | False   | Import VLAN views from Infoblox to Nautobot.                             |
 | infoblox_import_objects_vlans              | False   | Import VLANs from Infoblox to Nautobot.                                  |
 | infoblox_import_subnets                    | N/A     | List of Subnets in CIDR string notation to filter import to.             |
+| infoblox_network_view                      | N/A     | Only load IPAddresses from a specific Infoblox Network View.             |
 
 Below is an example snippet from `nautobot_config.py` that demonstrates how to enable and configure Infoblox integration:
 

nautobot_ssot/integrations/infoblox/constant.py

@@ -12,6 +12,7 @@ def _read_plugin_config():
         "NAUTOBOT_INFOBLOX_PASSWORD": config["infoblox_password"],
         "NAUTOBOT_INFOBLOX_VERIFY_SSL": config["infoblox_verify_ssl"],
         "NAUTOBOT_INFOBLOX_WAPI_VERSION": config["infoblox_wapi_version"],
+        "NAUTOBOT_INFOBLOX_NETWORK_VIEW": config["infoblox_network_view"],
         "enable_sync_to_infoblox": config["infoblox_enable_sync_to_infoblox"],
         "enable_rfc1918_network_containers": config["infoblox_enable_rfc1918_network_containers"],
         "default_status": config["infoblox_default_status"],

nautobot_ssot/integrations/infoblox/diffsync/adapters/infoblox.py

@@ -3,6 +3,7 @@
 
 from diffsync import DiffSync
 from diffsync.enum import DiffSyncFlags
+from diffsync.exceptions import ObjectAlreadyExists
 from nautobot.extras.plugins.exceptions import PluginImproperlyConfigured
 from nautobot_ssot.integrations.infoblox.constant import PLUGIN_CFG
 from nautobot_ssot.integrations.infoblox.utils.client import get_default_ext_attrs, get_dns_name
@@ -82,7 +83,14 @@ def load_prefixes(self):
                 ext_attrs={**default_ext_attrs, **pf_ext_attrs},
                 vlans=build_vlan_map(vlans=_pf["vlans"]) if _pf.get("vlans") else {},
             )
-            self.add(new_pf)
+            try:
+                self.add(new_pf)
+            except ObjectAlreadyExists:
+                self.job.logger.warning(
+                    f"Duplicate prefix found: {new_pf}. Duplicate prefixes are not supported, "
+                    "and only the first occurrence will be included in the sync. To load data "
+                    "from a single Network View, use the 'infoblox_network_view' setting."
+                )
 
     def load_ipaddresses(self):
         """Load InfobloxIPAddress DiffSync model."""
@@ -100,6 +108,7 @@ def load_ipaddresses(self):
                 prefix_length=prefix_length,
                 dns_name=dns_name,
                 status=self.conn.get_ipaddr_status(_ip),
+                ip_addr_type=self.conn.get_ipaddr_type(_ip),
                 description=_ip["comment"],
                 ext_attrs={**default_ext_attrs, **ip_ext_attrs},
             )

nautobot_ssot/integrations/infoblox/diffsync/adapters/nautobot.py

@@ -175,6 +175,7 @@ def load_ipaddresses(self):
                 address=addr,
                 prefix=str(prefix),
                 status=ipaddr.status.name if ipaddr.status else None,
+                ip_addr_type=ipaddr.type,
                 prefix_length=prefix.prefix_length if prefix else ipaddr.prefix_length,
                 dns_name=ipaddr.dns_name,
                 description=ipaddr.description,

nautobot_ssot/integrations/infoblox/diffsync/models/base.py

@@ -53,13 +53,14 @@ class IPAddress(DiffSyncModel):
 
     _modelname = "ipaddress"
     _identifiers = ("address", "prefix", "prefix_length")
-    _attributes = ("description", "dns_name", "status", "ext_attrs")
+    _attributes = ("description", "dns_name", "status", "ip_addr_type", "ext_attrs")
 
     address: str
     dns_name: str
     prefix: str
     prefix_length: int
     status: Optional[str]
+    ip_addr_type: Optional[str]
     description: Optional[str]
     ext_attrs: Optional[dict]
     pk: Optional[uuid.UUID] = None

nautobot_ssot/integrations/infoblox/diffsync/models/nautobot.py

@@ -5,7 +5,7 @@
 from nautobot.extras.choices import CustomFieldTypeChoices
 from nautobot.extras.models import RelationshipAssociation as OrmRelationshipAssociation
 from nautobot.extras.models import CustomField as OrmCF
-from nautobot.ipam.choices import IPAddressRoleChoices
+from nautobot.ipam.choices import IPAddressRoleChoices, IPAddressTypeChoices
 from nautobot.ipam.models import IPAddress as OrmIPAddress
 from nautobot.ipam.models import Prefix as OrmPrefix
 from nautobot.ipam.models import VLAN as OrmVlan
@@ -15,15 +15,15 @@
 from nautobot_ssot.integrations.infoblox.utils.nautobot import get_prefix_vlans
 
 
-def process_ext_attrs(diffsync, obj: object, extattrs: dict):
+def process_ext_attrs(diffsync, obj: object, extattrs: dict):  # pylint: disable=too-many-branches
     """Process Extensibility Attributes into Custom Fields or link to found objects.
 
     Args:
         diffsync (object): DiffSync Job
         obj (object): The object that's being created or updated and needs processing.
         extattrs (dict): The Extensibility Attributes to be analyzed and applied to passed `prefix`.
     """
-    for attr, attr_value in extattrs.items():
+    for attr, attr_value in extattrs.items():  # pylint: disable=too-many-nested-blocks
         if attr_value:
             if attr.lower() in ["site", "facility", "location"]:
                 try:
@@ -32,13 +32,28 @@ def process_ext_attrs(diffsync, obj: object, extattrs: dict):
                     diffsync.job.logger.warning(
                         f"Unable to find Location {attr_value} for {obj} found in Extensibility Attributes '{attr}'. {err}"
                     )
-            if attr.lower() == "vrf":
-                try:
-                    obj.vrfs.add(diffsync.vrf_map[attr_value])
-                except KeyError as err:
+                except TypeError as err:
                     diffsync.job.logger.warning(
-                        f"Unable to find VRF {attr_value} for {obj} found in Extensibility Attributes '{attr}'. {err}"
+                        f"Cannot set location values {attr_value} for {obj}. Multiple locations are assigned "
+                        f"in Extensibility Attributes '{attr}', but multiple location assignments are not "
+                        f"supported by Nautobot. {err}"
                     )
+            if attr.lower() == "vrf":
+                if isinstance(attr_value, list):
+                    for vrf in attr_value:
+                        try:
+                            obj.vrfs.add(diffsync.vrf_map[vrf])
+                        except KeyError as err:
+                            diffsync.job.logger.warning(
+                                f"Unable to find VRF {vrf} for {obj} found in Extensibility Attributes '{attr}'. {err}"
+                            )
+                else:
+                    try:
+                        obj.vrfs.add(diffsync.vrf_map[attr_value])
+                    except KeyError as err:
+                        diffsync.job.logger.warning(
+                            f"Unable to find VRF {attr_value} for {obj} found in Extensibility Attributes '{attr}'. {err}"
+                        )
             if "role" in attr.lower():
                 if isinstance(obj, OrmIPAddress) and attr_value.lower() in IPAddressRoleChoices.as_dict():
                     obj.role = attr_value.lower()
@@ -49,14 +64,25 @@ def process_ext_attrs(diffsync, obj: object, extattrs: dict):
                         diffsync.job.logger.warning(
                             f"Unable to find Role {attr_value} for {obj} found in Extensibility Attributes '{attr}'. {err}"
                         )
-
+                    except TypeError as err:
+                        diffsync.job.logger.warning(
+                            f"Cannot set role values {attr_value} for {obj}. Multiple roles are assigned "
+                            f"in Extensibility Attributes '{attr}', but multiple role assignments are not "
+                            f"supported by Nautobot. {err}"
+                        )
             if attr.lower() in ["tenant", "dept", "department"]:
                 try:
                     obj.tenant_id = diffsync.tenant_map[attr_value]
                 except KeyError as err:
                     diffsync.job.logger.warning(
                         f"Unable to find Tenant {attr_value} for {obj} found in Extensibility Attributes '{attr}'. {err}"
                     )
+                except TypeError as err:
+                    diffsync.job.logger.warning(
+                        f"Cannot set tenant values {attr_value} for {obj}. Multiple tenants are assigned "
+                        f"in Extensibility Attributes '{attr}', but multiple tenant assignments are not "
+                        f"supported by Nautobot. {err}"
+                    )
             _cf_dict = {
                 "key": slugify(attr).replace("-", "_"),
                 "type": CustomFieldTypeChoices.TYPE_TEXT,
@@ -175,10 +201,18 @@ def create(cls, diffsync, ids, attrs):
         except KeyError:
             status = diffsync.status_map[PLUGIN_CFG.get("default_status", "Active")]
         addr = f"{ids['address']}/{ids['prefix_length']}"
-        diffsync.job.logger.debug(f"Creating IP Address {addr}")
+        if attrs.get("ip_addr_type"):
+            if attrs["ip_addr_type"].lower() in IPAddressTypeChoices.as_dict():
+                ip_addr_type = attrs["ip_addr_type"].lower()
+            else:
+                diffsync.logger.warning(f"unable to determine IPAddress Type for {addr}, defaulting to 'Host'")
+                ip_addr_type = "host"
+        if diffsync.job.debug:
+            diffsync.job.logger.debug(f"Creating IP Address {addr}")
         _ip = OrmIPAddress(
             address=addr,
             status_id=status,
+            type=ip_addr_type,
             description=attrs.get("description", ""),
             dns_name=attrs.get("dns_name", ""),
             parent_id=diffsync.prefix_map[ids["prefix"]],
@@ -204,6 +238,11 @@ def update(self, attrs):
             except KeyError:
                 status = self.diffsync.status_map[PLUGIN_CFG.get("default_status", "Active")]
             _ipaddr.status_id = status
+        if attrs.get("ip_addr_type"):
+            if attrs["ip_addr_type"].lower() in IPAddressTypeChoices.as_dict():
+                _ipaddr.type = attrs["ip_addr_type"].lower()
+            else:
+                _ipaddr.type = "host"
         if attrs.get("description"):
             _ipaddr.description = attrs["description"]
         if attrs.get("dns_name"):

nautobot_ssot/integrations/infoblox/utils/client.py

@@ -791,7 +791,8 @@ def get_all_subnets(self, prefix: str = None):
             "_return_fields": "network,network_view,comment,extattrs,rir_organization,rir,vlans",
             "_max_results": 10000,
         }
-
+        if PLUGIN_CFG.get("NAUTOBOT_INFOBLOX_NETWORK_VIEW"):
+            params.update({"network_view": PLUGIN_CFG["NAUTOBOT_INFOBLOX_NETWORK_VIEW"]})
         if prefix:
             params.update({"network": prefix})
         try:
@@ -1172,10 +1173,19 @@ def create_vlan(self, vlan_id, vlan_name, vlan_view):
 
     @staticmethod
     def get_ipaddr_status(ip_record: dict) -> str:
-        """Determine the IPAddress status based upon types and usage keys."""
+        """Determine the IPAddress status based on the status key."""
+        if "USED" in ip_record["status"]:
+            return "Active"
+        return "Reserved"
+
+    @staticmethod
+    def get_ipaddr_type(ip_record: dict) -> str:
+        """Determine the IPAddress type based on the usage key."""
         if "DHCP" in ip_record["usage"]:
-            return "DHCP"
-        return "Active"
+            return "dhcp"
+        if "SLAAC" in ip_record["usage"]:
+            return "slaac"
+        return "host"
 
     def _find_resource(self, resource, **params):
         """Find the resource for given parameters.
@@ -1275,6 +1285,8 @@ def get_network_containers(self, prefix: str = ""):
             "_return_fields": "network,comment,network_view,extattrs,rir_organization,rir",
             "_max_results": 100000,
         }
+        if PLUGIN_CFG.get("NAUTOBOT_INFOBLOX_NETWORK_VIEW"):
+            params.update({"network_view": PLUGIN_CFG["NAUTOBOT_INFOBLOX_NETWORK_VIEW"]})
         if prefix:
             params.update({"network": prefix})
         response = self._request("GET", url_path, params=params)
@@ -1317,6 +1329,8 @@ def get_child_network_containers(self, prefix: str):
             "_return_fields": "network,comment,network_view,extattrs,rir_organization,rir",
             "_max_results": 100000,
         }
+        if PLUGIN_CFG.get("NAUTOBOT_INFOBLOX_NETWORK_VIEW"):
+            params.update({"network_view": PLUGIN_CFG["NAUTOBOT_INFOBLOX_NETWORK_VIEW"]})
         params.update({"network_container": prefix})
         response = self._request("GET", url_path, params=params)
         response = response.json()
@@ -1363,7 +1377,8 @@ def get_child_subnets_from_container(self, prefix: str):
             "_return_fields": "network,network_view,comment,extattrs,rir_organization,rir,vlans",
             "_max_results": 10000,
         }
-
+        if PLUGIN_CFG.get("NAUTOBOT_INFOBLOX_NETWORK_VIEW"):
+            params.update({"network_view": PLUGIN_CFG["NAUTOBOT_INFOBLOX_NETWORK_VIEW"]})
         params.update({"network_container": prefix})
 
         try:

nautobot_ssot/tests/test_contrib.py

@@ -390,7 +390,7 @@ class BaseModelCustomFieldTest(TestCase):
     def test_custom_field_set(self):
         """Test whether setting a custom field value works."""
         custom_field_name = "Is Global"
-        custom_field = CustomField.objects.create(key=custom_field_name, label=custom_field_name, type="boolean")
+        custom_field = CustomField.objects.create(key="is_global", label=custom_field_name, type="boolean")
         custom_field.content_types.set([ContentType.objects.get_for_model(Provider)])
 
         class ProviderModel(NautobotModel):
@@ -402,7 +402,7 @@ class ProviderModel(NautobotModel):
 
             name: str
 
-            is_global: Annotated[bool, CustomFieldAnnotation(name=custom_field_name)] = False
+            is_global: Annotated[bool, CustomFieldAnnotation(name="is_global")] = False
 
         provider_name = "Test Provider"
         provider = Provider.objects.create(name=provider_name)
@@ -413,7 +413,7 @@ class ProviderModel(NautobotModel):
 
         provider.refresh_from_db()
         self.assertEqual(
-            provider.cf[custom_field_name],
+            provider.cf["is_global"],
             updated_custom_field_value,
             "Setting a custom field through 'NautobotModel' does not work.",
         )