feat: add more defaults support for network and device discovery

device-discovery/device_discovery/policy/models.py

@@ -30,14 +30,29 @@ class Napalm(BaseModel):
         default=None, description="Optional arguments"
     )
 
+class ObjectParameters(BaseModel):
+    """Model for object parameters."""
+
+    comments: str | None = Field(default=None, description="Comments, optional")
+    description: str | None = Field(default=None, description="Description, optional")
+    tags: list[str] | None = Field(default=None, description="Tags, optional")
+
+class Defaults(BaseModel):
+    """Model for default configuration."""
+
+    site: str | None = Field(default=None, description="Site name, optional")
+    role: str | None = Field(default=None, description="Device Role name, optional")
+    tags: list[str] | None = Field(default=None, description="Tags, optional")
+    device: ObjectParameters | None = Field(default=None, description="Device parameters, optional")
+    interface: ObjectParameters | None = Field(default=None, description="Interface parameters, optional")
+    ipaddress: ObjectParameters | None = Field(default=None, description="IP Address parameters, optional")
+    prefix: ObjectParameters | None = Field(default=None, description="Prefix parameters, optional")
 
 class Config(BaseModel):
     """Model for discovery configuration."""
 
     schedule: str | None = Field(default=None, description="cron interval, optional")
-    defaults: dict[str, str] | None = Field(
-        default=None, description="NetBox configuration"
-    )
+    defaults: Defaults | None = Field(default=None, description="Default configuration, optional")
 
     @field_validator("schedule")
     @classmethod

device-discovery/device_discovery/policy/runner.py

@@ -13,7 +13,7 @@
 
 from device_discovery.client import Client
 from device_discovery.discovery import discover_device_driver, supported_drivers
-from device_discovery.policy.models import Config, Napalm, Status
+from device_discovery.policy.models import Config, Defaults, Napalm, Status
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)
@@ -125,10 +125,10 @@ def run(self, id: str, scope: Napalm, config: Config):
             ) as device:
                 data = {
                     "driver": scope.driver,
-                    "site": config.defaults.get("site", None),
                     "device": device.get_facts(),
                     "interface": device.get_interfaces(),
                     "interface_ip": device.get_interfaces_ip(),
+                    "defaults": config.defaults,
                 }
                 Client().ingest(scope.hostname, data)
         except Exception as e:

device-discovery/device_discovery/translate.py

@@ -15,6 +15,8 @@
     Prefix,
 )
 
+from device_discovery.policy.models import Defaults
+
 
 def int32_overflows(number: int) -> bool:
     """
@@ -34,19 +36,29 @@ def int32_overflows(number: int) -> bool:
     return not (INT32_MIN <= number <= INT32_MAX)
 
 
-def translate_device(device_info: dict) -> Device:
+def translate_device(device_info: dict, defaults: Defaults) -> Device:
     """
     Translate device information from NAPALM format to Diode SDK Device entity.
 
     Args:
     ----
         device_info (dict): Dictionary containing device information.
+        defaults (Defaults): Default configuration.
 
     Returns:
     -------
         Device: Translated Device entity.
 
     """
+    tags = list(defaults.tags) if defaults.tags else []
+    description = None
+    comments = None
+
+    if defaults.device:
+        tags.extend(defaults.device.tags)
+        description = defaults.device.description
+        comments = defaults.device.comments
+
     device = Device(
         name=device_info.get("hostname"),
         device_type=DeviceType(
@@ -55,15 +67,19 @@ def translate_device(device_info: dict) -> Device:
         platform=Platform(
             name=device_info.get("driver"), manufacturer=device_info.get("vendor")
         ),
+        role=defaults.role,
         serial=device_info.get("serial_number"),
         status="active",
-        site=device_info.get("site"),
+        site=defaults.site,
+        tags=tags,
+        description=description,
+        comments=comments,
     )
     return device
 
 
 def translate_interface(
-    device: Device, if_name: str, interface_info: dict
+    device: Device, if_name: str, interface_info: dict, defaults: Defaults
 ) -> Interface:
     """
     Translate interface information from NAPALM format to Diode SDK Interface entity.
@@ -73,18 +89,29 @@ def translate_interface(
         device (Device): The device to which the interface belongs.
         if_name (str): The name of the interface.
         interface_info (dict): Dictionary containing interface information.
+        defaults (Defaults): Default configuration.
 
     Returns:
     -------
         Interface: Translated Interface entity.
 
     """
+    tags = list(defaults.tags) if defaults.tags else []
+    description = None
+
+    if defaults.interface:
+        tags.extend(defaults.interface.tags)
+        description = defaults.interface.description
+
+    description = interface_info.get("description", description)
+
     interface = Interface(
         device=device,
         name=if_name,
         enabled=interface_info.get("is_enabled"),
         mac_address=interface_info.get("mac_address"),
-        description=interface_info.get("description"),
+        description=description,
+        tags=tags,
     )
 
     # Convert napalm interface speed from Mbps to Netbox Kbps
@@ -100,7 +127,7 @@ def translate_interface(
 
 
 def translate_interface_ips(
-    interface: Interface, interfaces_ip: dict
+    interface: Interface, interfaces_ip: dict, defaults: Defaults
 ) -> Iterable[Entity]:
     """
     Translate IP address and Prefixes information for an interface.
@@ -110,12 +137,32 @@ def translate_interface_ips(
         interface (Interface): The interface entity.
         if_name (str): The name of the interface.
         interfaces_ip (dict): Dictionary containing interface IP information.
+        defaults (Defaults): Default configuration.
 
     Returns:
     -------
         Iterable[Entity]: Iterable of translated IP address and Prefixes entities.
 
     """
+    tags = defaults.tags if defaults.tags else []
+    ip_tags = list(tags)
+    ip_comments = None
+    ip_description = None
+
+    prefix_tags = list(tags)
+    prefix_comments = None
+    prefix_description = None
+
+    if defaults.ipaddress:
+        ip_tags.extend(defaults.ipaddress.tags)
+        ip_comments = defaults.ipaddress.comments
+        ip_description = defaults.ipaddress.description
+
+    if defaults.prefix:
+        prefix_tags.extend(defaults.prefix.tags)
+        prefix_comments = defaults.prefix.comments
+        prefix_description = defaults.prefix.description
+
     ip_entities = []
 
     for if_ip_name, ip_info in interfaces_ip.items():
@@ -127,14 +174,22 @@ def translate_interface_ips(
                     ip_entities.append(
                         Entity(
                             prefix=Prefix(
-                                prefix=str(network), site=interface.device.site
+                                prefix=str(network),
+                                site=interface.device.site,
+                                tags=prefix_tags,
+                                comments=prefix_comments,
+                                description=prefix_description,
                             )
                         )
                     )
                     ip_entities.append(
                         Entity(
                             ip_address=IPAddress(
-                                address=ip_address, interface=interface
+                                address=ip_address,
+                                interface=interface,
+                                tags=ip_tags,
+                                comments=ip_comments,
+                                description=ip_description,
                             )
                         )
                     )
@@ -157,20 +212,25 @@ def translate_data(data: dict) -> Iterable[Entity]:
     """
     entities = []
 
+    defaults = data.get("defaults", Defaults())
+
     device_info = data.get("device", {})
     interfaces = data.get("interface", {})
     interfaces_ip = data.get("interface_ip", {})
     if device_info:
         device_info["driver"] = data.get("driver")
-        device_info["site"] = data.get("site")
-        device = translate_device(device_info)
+        device = translate_device(device_info, defaults)
         entities.append(Entity(device=device))
 
         interface_list = device_info.get("interface_list", [])
         for if_name, interface_info in interfaces.items():
             if if_name in interface_list:
-                interface = translate_interface(device, if_name, interface_info)
+                interface = translate_interface(
+                    device, if_name, interface_info, defaults
+                )
                 entities.append(Entity(interface=interface))
-                entities.extend(translate_interface_ips(interface, interfaces_ip))
+                entities.extend(
+                    translate_interface_ips(interface, interfaces_ip, defaults)
+                )
 
     return entities

device-discovery/tests/policy/test_runner.py

@@ -7,7 +7,7 @@
 import pytest
 from apscheduler.triggers.date import DateTrigger
 
-from device_discovery.policy.models import Config, Napalm, Status
+from device_discovery.policy.models import Config, Defaults, Napalm, Status
 from device_discovery.policy.runner import PolicyRunner
 
 
@@ -20,7 +20,7 @@ def policy_runner():
 @pytest.fixture
 def sample_config():
     """Fixture for a sample Config object."""
-    return Config(schedule="0 * * * *", defaults={"site": "New York"})
+    return Config(schedule="0 * * * *", defaults=Defaults(site="New York"))
 
 
 @pytest.fixture

device-discovery/tests/test_client.py

@@ -2,6 +2,7 @@
 # Copyright 2024 NetBox Labs Inc
 """NetBox Labs - Client Unit Tests."""
 
+from types import SimpleNamespace
 from unittest.mock import patch
 
 import pytest
@@ -35,7 +36,7 @@ def sample_data():
             "GigabitEthernet0/0": {"ipv4": {"192.0.2.1": {"prefix_length": 24}}}
         },
         "driver": "ios",
-        "site": "New York",
+        "defaults": SimpleNamespace(site="New York", role=None, tags = None, device = None),
     }