diff --git a/changes/329.added b/changes/329.added new file mode 100644 index 00000000..fd7918d9 --- /dev/null +++ b/changes/329.added @@ -0,0 +1 @@ +Updated Sync Devices from Network job to automatically update location type content types if required \ No newline at end of file diff --git a/nautobot_device_onboarding/jobs.py b/nautobot_device_onboarding/jobs.py index 02a70955..daa4c9db 100755 --- a/nautobot_device_onboarding/jobs.py +++ b/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} diff --git a/nautobot_device_onboarding/tests/test_jobs.py b/nautobot_device_onboarding/tests/test_jobs.py index 8928e56f..e78b6ede 100644 --- a/nautobot_device_onboarding/tests/test_jobs.py +++ b/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.""" diff --git a/nautobot_device_onboarding/utils/helper.py b/nautobot_device_onboarding/utils/helper.py index 8ddbcd17..1aae0175 100644 --- a/nautobot_device_onboarding/utils/helper.py +++ b/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 @@ def check_for_required_file(directory, filename): 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