automatically add the device content type to the site type

changes/329.added

@@ -0,0 +1 @@
+Updated Sync Devices from Network job to automatically update location type content types if required

nautobot_device_onboarding/jobs.py

@@ -60,7 +60,7 @@
 from nautobot_device_onboarding.nornir_plays.inventory_creator import _set_inventory
 from nautobot_device_onboarding.nornir_plays.logger import NornirLogger
 from nautobot_device_onboarding.nornir_plays.processor import TroubleshootingProcessor
-from nautobot_device_onboarding.utils.helper import onboarding_task_fqdn_to_ip
+from nautobot_device_onboarding.utils.helper import add_content_type, onboarding_task_fqdn_to_ip
 
 InventoryPluginRegister.register("empty-inventory", EmptyInventory)
 
@@ -371,6 +371,8 @@ def _process_csv_data(self, csv_file):
             self.logger.error("The CSV file contains no data!")
             return None
 
+        location_type_ids = set()
+
         self.logger.info("Processing CSV data...")
         processing_failed = False
         processed_csv_data = {}
@@ -386,6 +388,14 @@ def _process_csv_data(self, csv_file):
                 else:
                     query = query = f"location_name: {row.get('location_name')}"
                     location = Location.objects.get(name=row["location_name"].strip(), parent=None)
+
+                # Check and add content type if needed (only once per location type)
+                location_type = location.location_type
+                if location_type.id not in location_type_ids:
+                    if not location_type.content_types.filter(app_label="dcim", model="device").exists():
+                        add_content_type(self, model_to_add=Device, target_object=location_type)
+                    location_type_ids.add(location_type.id)
+
                 query = f"device_role: {row.get('device_role_name')}"
                 device_role = Role.objects.get(
                     name=row["device_role_name"].strip(),
@@ -552,6 +562,11 @@ def run(  # pylint: disable=too-many-positional-arguments
                     # TODO: We're only raising an exception if a csv file is not provided. Is that correct?
                     raise ValueError("Platform.network_driver missing")
 
+            if location:
+                location_type = location.location_type
+                if not location_type.content_types.filter(app_label="dcim", model="device").exists():
+                    add_content_type(self, model_to_add=Device, target_object=location_type)
+
             for ip_address in ip_addresses.replace(" ", "").split(","):
                 resolved = self._validate_ip_address(ip_address)
                 self.ip_address_inventory[resolved] = {"original_ip_address": ip_address, **default_values}

nautobot_device_onboarding/tests/test_jobs.py

@@ -179,6 +179,68 @@ def test_csv_process_pass_connectivity_test_flag(self, mock_sync_devices_command
         self.assertEqual(job.ip_address_inventory["172.23.0.8"]["platform"], None)
         self.assertEqual(log_level, 10)
 
+    def test_add_content_type_during_csv_sync(self):
+        """Test successful addition of content type to location type during CSV sync."""
+        # Create a location type without Device content type
+        location_type_without_device = self.testing_objects["location_2"].location_type
+        location_type_without_device.content_types.clear()
+        location_type_without_device.validated_save()
+
+        self.assertFalse(location_type_without_device.content_types.filter(app_label="dcim", model="device").exists())
+
+        # Run CSV processing which should add the content type
+        onboarding_job = jobs.SSOTSyncDevices()
+        with open("nautobot_device_onboarding/tests/fixtures/onboarding_csv_fixture.csv", "rb") as csv_file:
+            onboarding_job._process_csv_data(csv_file=csv_file)  # pylint: disable=protected-access
+
+        # Verify content type was added
+        location_type_without_device.refresh_from_db()
+        self.assertTrue(location_type_without_device.content_types.filter(app_label="dcim", model="device").exists())
+
+    @patch("nautobot_device_onboarding.diffsync.adapters.sync_devices_adapters.sync_devices_command_getter")
+    def test_add_content_type_during_manual_sync(self, device_data):
+        """Test that content type is added when running manual sync with location."""
+        device_data.return_value = sync_devices_fixture.sync_devices_mock_data_valid
+
+        # Create a location type without Device content type
+        location_type_without_device = self.testing_objects["location_2"].location_type
+        location_type_without_device.content_types.clear()
+        location_type_without_device.validated_save()
+
+        self.assertFalse(location_type_without_device.content_types.filter(app_label="dcim", model="device").exists())
+
+        job_form_inputs = {
+            "debug": True,
+            "connectivity_test": False,
+            "dryrun": False,
+            "csv_file": None,
+            "location": self.testing_objects["location_2"].pk,
+            "namespace": self.testing_objects["namespace"].pk,
+            "ip_addresses": "10.1.1.10,10.1.1.11",
+            "port": 22,
+            "timeout": 30,
+            "set_mgmt_only": True,
+            "update_devices_without_primary_ip": True,
+            "device_role": self.testing_objects["device_role"].pk,
+            "device_status": self.testing_objects["status"].pk,
+            "interface_status": self.testing_objects["status"].pk,
+            "ip_address_status": self.testing_objects["status"].pk,
+            "secrets_group": self.testing_objects["secrets_group"].pk,
+            "platform": None,
+            "memory_profiling": False,
+        }
+        job_result = create_job_result_and_run_job(
+            module="nautobot_device_onboarding.jobs", name="SSOTSyncDevices", **job_form_inputs
+        )
+        self.assertEqual(
+            job_result.status,
+            JobResultStatusChoices.STATUS_SUCCESS,
+            (job_result.traceback, list(job_result.job_log_entries.values_list("message", flat=True))),
+        )
+        # Verify content type was added
+        location_type_without_device.refresh_from_db()
+        self.assertTrue(location_type_without_device.content_types.filter(app_label="dcim", model="device").exists())
+
 
 class SSOTSyncNetworkDataTestCase(TransactionTestCase):
     """Test SSOTSyncNetworkData class."""

nautobot_device_onboarding/utils/helper.py

@@ -5,6 +5,7 @@
 import socket
 
 import netaddr
+from django.contrib.contenttypes.models import ContentType
 from nautobot.dcim.filters import DeviceFilterSet
 from nautobot.dcim.models import Device
 from netaddr.core import AddrFormatError
@@ -119,3 +120,28 @@
         return False
     except FileNotFoundError:
         return False
+
+
+def add_content_type(job, model_to_add, target_object):
+    """Add a content type to the valid content types of a target object.
+
+    Args:
+        job: The job object used for logging.
+        model_to_add: The model class to get the content type for.
+        target_object: The object to which the content type will be added.
+
+    Raises:
+        OnboardException: If adding the content type fails.
+    """
+    try:
+        job.logger.info(
+            "Adding %s content type to valid content types for location type %s",
+            model_to_add.__name__,
+            target_object,
+        )
+        content_type = ContentType.objects.get_for_model(model_to_add)
+        target_object.content_types.add(content_type)
+    except Exception as e:
+        err_msg = f"Failed to add {model_to_add.__name__} to valid content types for {target_object}: {e}"
+        job.logger.error(err_msg)
+        raise OnboardException("fail-general - " + err_msg) from e