Adding mac initializer. Fixing cables, clusters, prefixes and virtualizzation_interfaces

Author: LuPo

src/netbox_initializers/initializers/__init__.py

@@ -22,6 +22,7 @@
 from .interfaces import InterfaceInitializer
 from .ip_addresses import IPAddressInitializer
 from .locations import LocationInitializer
+from .macs import MACAddressInitializer
 from .manufacturers import ManufacturerInitializer
 from .object_permissions import ObjectPermissionInitializer
 from .platforms import PlatformInitializer

src/netbox_initializers/initializers/base.py

@@ -2,6 +2,7 @@
 from typing import Tuple
 
 from core.models import ObjectType
+from dcim.models import MACAddress
 from django.core.exceptions import ObjectDoesNotExist
 from extras.models import CustomField, Tag
 from ruamel.yaml import YAML
@@ -90,6 +91,22 @@ def set_tags(self, entity, tags):
         if save:
             entity.save()
 
+    def set_mac_addresses(self, entity, mac_addresses):
+        if not mac_addresses:
+            return
+
+        if not hasattr(entity, "mac_addresses"):
+            raise Exception(f"⚠️ MAC Address cannot be applied to {entity}'s model")
+
+        save = False
+        for mac_address in MACAddress.objects.filter(mac_address__in=mac_addresses):
+
+            entity.mac_addresses.add(mac_address)
+            save = True
+
+        if save:
+            entity.save()
+
     def split_params(self, params: dict, unique_params: list = None) -> Tuple[dict, dict]:
         """Split params dict into dict with matching params and a dict with default values"""
 
@@ -140,6 +157,7 @@ class InitializationError(Exception):
     "prefix_vlan_roles",
     "vlan_groups",
     "vlans",
+    "macs",
     "devices",
     "interfaces",
     "route_targets",

src/netbox_initializers/initializers/cables.py

@@ -1,6 +1,6 @@
 from typing import Tuple
-
-from circuits.models import Circuit, CircuitTermination, ProviderNetwork
+from circuits.constants import CIRCUIT_TERMINATION_TERMINATION_TYPES
+from circuits.models import Circuit, CircuitTermination
 from dcim.models import (
     Cable,
     CableTermination,
@@ -13,12 +13,12 @@
     PowerPanel,
     PowerPort,
     RearPort,
-    Site,
 )
 from django.contrib.contenttypes.models import ContentType
 from django.db.models import Q
 
 from netbox_initializers.initializers.base import BaseInitializer, register_initializer
+from netbox_initializers.initializers.utils import get_scope_details
 
 CONSOLE_PORT_TERMINATION = ContentType.objects.get_for_model(ConsolePort)
 CONSOLE_SERVER_PORT_TERMINATION = ContentType.objects.get_for_model(ConsoleServerPort)
@@ -55,16 +55,11 @@ def get_termination_object(params: dict, side: str):
         circuit = Circuit.objects.get(cid=circuit_params.pop("cid"))
         term_side = circuit_params.pop("term_side").upper()
 
-        site_name = circuit_params.pop("site", None)
-        provider_network = circuit_params.pop("provider_network", None)
-
-        if site_name:
-            circuit_params["site"] = Site.objects.get(name=site_name)
-        elif provider_network:
-            circuit_params["provider_network"] = ProviderNetwork.objects.get(name=provider_network)
+        if scope := circuit_params.pop("scope", None):
+            circuit_params["termination_type"], circuit_params["termination_id"] = get_scope_details(scope, CIRCUIT_TERMINATION_TERMINATION_TYPES)
         else:
             raise ValueError(
-                f"⚠️ Missing one of required parameters: 'site' or 'provider_network' "
+                f"⚠️ Missing required parameter: 'scope'"
                 f"for side {term_side} of circuit {circuit}"
             )
 

src/netbox_initializers/initializers/clusters.py

@@ -7,7 +7,7 @@
 MATCH_PARAMS = ["name", "type"]
 REQUIRED_ASSOCS = {"type": (ClusterType, "name")}
 OPTIONAL_ASSOCS = {
-    "site": (Site, "name"),
+    "scope": (Site, "name"),
     "group": (ClusterGroup, "name"),
     "tenant": (Tenant, "name"),
 }

src/netbox_initializers/initializers/macs.py

@@ -0,0 +1,24 @@
+from dcim.models import MACAddress
+
+from netbox_initializers.initializers.base import BaseInitializer, register_initializer
+
+
+class MACAddressInitializer(BaseInitializer):
+    data_file_name = "macs.yml"
+
+    def load_data(self):
+        macs = self.load_yaml()
+        if macs is None:
+            return
+
+        for mac in macs:
+            tags = mac.pop("tags", None)
+            macaddress, created = MACAddress.objects.get_or_create(**mac)
+
+            if created:
+                print("🗺️ Created MAC Address", macaddress.mac_address)
+
+            self.set_tags(macaddress, tags)
+
+
+register_initializer("macs", MACAddressInitializer)

src/netbox_initializers/initializers/prefixes.py

@@ -1,13 +1,14 @@
-from dcim.models import Site
 from ipam.models import VLAN, VRF, Prefix, Role
 from netaddr import IPNetwork
 from tenancy.models import Tenant, TenantGroup
+from dcim.constants import LOCATION_SCOPE_TYPES
+from django.contrib.contenttypes.models import ContentType
 
 from netbox_initializers.initializers.base import BaseInitializer, register_initializer
+from netbox_initializers.initializers.utils import get_scope_details
 
-MATCH_PARAMS = ["prefix", "site", "vrf", "vlan"]
+MATCH_PARAMS = ["prefix", "scope", "vrf", "vlan"]
 OPTIONAL_ASSOCS = {
-    "site": (Site, "name"),
     "tenant": (Tenant, "name"),
     "tenant_group": (TenantGroup, "name"),
     "vlan": (VLAN, "name"),
@@ -29,6 +30,9 @@ def load_data(self):
 
             params["prefix"] = IPNetwork(params["prefix"])
 
+            if scope := params.pop("scope"):
+                params["scope_type"], params["scope_id"] = get_scope_details(scope, LOCATION_SCOPE_TYPES)
+
             for assoc, details in OPTIONAL_ASSOCS.items():
                 if assoc in params:
                     model, field = details

src/netbox_initializers/initializers/utils.py

@@ -0,0 +1,13 @@
+from django.contrib.contenttypes.models import ContentType
+
+
+def get_scope_details(scope: dict, allowed_termination_types: list):
+    try:
+        scope_type = ContentType.objects.get(app_label__in=["dcim", "circuits"], model=scope["type"])
+        if scope["type"] not in allowed_termination_types:
+            raise ValueError(f"{scope['type']} scope type is not permitted on {scope_type.app_label}")
+    except ContentType.DoesNotExist:
+        raise ValueError(f"⚠️ Invalid scope type: {scope['type']}")
+
+    scope_id = scope_type.model_class().objects.get(name=scope["name"]).id
+    return scope_type, scope_id

src/netbox_initializers/initializers/virtualization_interfaces.py

@@ -1,9 +1,11 @@
+from dcim.models import MACAddress
 from virtualization.models import VirtualMachine, VMInterface
 
 from netbox_initializers.initializers.base import BaseInitializer, register_initializer
 
 MATCH_PARAMS = ["name", "virtual_machine"]
 REQUIRED_ASSOCS = {"virtual_machine": (VirtualMachine, "name")}
+OPTIONAL_ASSOCS = {"primary_mac_address": (MACAddress, "mac_address")}
 
 
 class VMInterfaceInitializer(BaseInitializer):
@@ -16,13 +18,21 @@ def load_data(self):
         for params in interfaces:
             custom_field_data = self.pop_custom_fields(params)
             tags = params.pop("tags", None)
+            mac_addresses = params.pop("mac_addresses", None)
 
             for assoc, details in REQUIRED_ASSOCS.items():
                 model, field = details
                 query = {field: params.pop(assoc)}
 
                 params[assoc] = model.objects.get(**query)
 
+            for assoc, details in OPTIONAL_ASSOCS.items():
+                if assoc in params:
+                    model, field = details
+                    query = {field: params.pop(assoc)}
+
+                    params[assoc] = model.objects.get(**query)
+
             matching_params, defaults = self.split_params(params, MATCH_PARAMS)
             interface, created = VMInterface.objects.get_or_create(
                 **matching_params, defaults=defaults
@@ -33,6 +43,7 @@ def load_data(self):
 
             self.set_custom_fields_values(interface, custom_field_data)
             self.set_tags(interface, tags)
+            self.set_mac_addresses(interface, mac_addresses)
 
 
 register_initializer("virtualization_interfaces", VMInterfaceInitializer)

src/netbox_initializers/initializers/yaml/cables.yml

@@ -16,7 +16,9 @@
 # # termination_x_circuit:
 # #   term_side -> termination side of a circuit. Must be A or B
 # #   cid -> circuit ID value
-# #   site OR provider_network -> name of Site or ProviderNetwork respectively. If both provided, Site takes precedence
+# #   scope:
+# #     type -> select one of the following: region, site, sitegroup, location
+# #     name -> name of the object in the respective scope type
 # # ```
 # #
 # # If a termination is a power feed then the required parameter is termination_x_feed.
@@ -51,7 +53,9 @@
 #   termination_b_circuit:
 #     term_side: A
 #     cid: Circuit_ID-1
-#     site: AMS 1
+#     scope: 
+#       type: site
+#       name: AMS 1
 #   type: cat6
 
 # - termination_a_name: psu0

src/netbox_initializers/initializers/yaml/clusters.yml

@@ -4,4 +4,4 @@
 #   tenant: tenant1
 # - name: cluster2
 #   type: Hyper-V
-#   site: SING 1
+#   scope: SING 1