From 719effb548ef7fdc7a21acf43c1b0b26950c30a5 Mon Sep 17 00:00:00 2001 From: bctiemann Date: Mon, 9 Mar 2026 14:11:40 -0400 Subject: [PATCH] Fixes: #20123 - Add replicate_components and adopt_components write_only fields to ModuleSerializer (#21600) --- netbox/dcim/api/serializers_/devices.py | 132 +++++++++++++- netbox/dcim/tests/test_api.py | 232 ++++++++++++++++++++++++ 2 files changed, 363 insertions(+), 1 deletion(-) diff --git a/netbox/dcim/api/serializers_/devices.py b/netbox/dcim/api/serializers_/devices.py index 7be8e42c9..4da97ee78 100644 --- a/netbox/dcim/api/serializers_/devices.py +++ b/netbox/dcim/api/serializers_/devices.py @@ -6,7 +6,7 @@ from drf_spectacular.utils import extend_schema_field from rest_framework import serializers from dcim.choices import * -from dcim.constants import MACADDRESS_ASSIGNMENT_MODELS +from dcim.constants import MACADDRESS_ASSIGNMENT_MODELS, MODULE_TOKEN from dcim.models import Device, DeviceBay, MACAddress, Module, VirtualDeviceContext from extras.api.serializers_.configtemplates import ConfigTemplateSerializer from ipam.api.serializers_.ip import IPAddressSerializer @@ -150,15 +150,145 @@ class ModuleSerializer(PrimaryModelSerializer): module_bay = NestedModuleBaySerializer() module_type = ModuleTypeSerializer(nested=True) status = ChoiceField(choices=ModuleStatusChoices, required=False) + replicate_components = serializers.BooleanField( + required=False, + default=True, + write_only=True, + label=_('Replicate components'), + help_text=_('Automatically populate components associated with this module type (default: true)') + ) + adopt_components = serializers.BooleanField( + required=False, + default=False, + write_only=True, + label=_('Adopt components'), + help_text=_('Adopt already existing components') + ) class Meta: model = Module fields = [ 'id', 'url', 'display_url', 'display', 'device', 'module_bay', 'module_type', 'status', 'serial', 'asset_tag', 'description', 'owner', 'comments', 'tags', 'custom_fields', 'created', 'last_updated', + 'replicate_components', 'adopt_components', ] brief_fields = ('id', 'url', 'display', 'device', 'module_bay', 'module_type', 'description') + def validate(self, data): + # When used as a nested serializer (e.g. as the `module` field on device component + # serializers), `data` is already a resolved Module instance — skip our custom logic. + if self.nested: + return super().validate(data) + + # Pop write-only transient fields before ValidatedModelSerializer tries to + # construct a Module instance for full_clean(); restore them afterwards. + replicate_components = data.pop('replicate_components', True) + adopt_components = data.pop('adopt_components', False) + data = super().validate(data) + + # For updates these fields are not meaningful; omit them from validated_data so that + # ModelSerializer.update() does not set unexpected attributes on the instance. + if self.instance: + return data + + # Always pass the flags to create() so it can set the correct private attributes. + data['replicate_components'] = replicate_components + data['adopt_components'] = adopt_components + + # Skip conflict checks when no component operations are requested. + if not replicate_components and not adopt_components: + return data + + device = data.get('device') + module_type = data.get('module_type') + module_bay = data.get('module_bay') + + # Required-field validation fires separately; skip here if any are missing. + if not all([device, module_type, module_bay]): + return data + + # Build module bay tree for MODULE_TOKEN placeholder resolution (outermost to innermost) + module_bays = [] + current_bay = module_bay + while current_bay: + module_bays.append(current_bay) + current_bay = current_bay.module.module_bay if current_bay.module else None + module_bays.reverse() + + for templates_attr, component_attr in [ + ('consoleporttemplates', 'consoleports'), + ('consoleserverporttemplates', 'consoleserverports'), + ('interfacetemplates', 'interfaces'), + ('powerporttemplates', 'powerports'), + ('poweroutlettemplates', 'poweroutlets'), + ('rearporttemplates', 'rearports'), + ('frontporttemplates', 'frontports'), + ]: + installed_components = { + component.name: component + for component in getattr(device, component_attr).all() + } + + for template in getattr(module_type, templates_attr).all(): + resolved_name = template.name + if MODULE_TOKEN in template.name: + if not module_bay.position: + raise serializers.ValidationError( + _("Cannot install module with placeholder values in a module bay with no position defined.") + ) + if template.name.count(MODULE_TOKEN) != len(module_bays): + raise serializers.ValidationError( + _( + "Cannot install module with placeholder values in a module bay tree {level} in tree " + "but {tokens} placeholders given." + ).format( + level=len(module_bays), tokens=template.name.count(MODULE_TOKEN) + ) + ) + for bay in module_bays: + resolved_name = resolved_name.replace(MODULE_TOKEN, bay.position, 1) + + existing_item = installed_components.get(resolved_name) + + if adopt_components and existing_item and existing_item.module: + raise serializers.ValidationError( + _("Cannot adopt {model} {name} as it already belongs to a module").format( + model=template.component_model.__name__, + name=resolved_name + ) + ) + + if not adopt_components and resolved_name in installed_components: + raise serializers.ValidationError( + _("A {model} named {name} already exists").format( + model=template.component_model.__name__, + name=resolved_name + ) + ) + + return data + + def create(self, validated_data): + replicate_components = validated_data.pop('replicate_components', True) + adopt_components = validated_data.pop('adopt_components', False) + + # Tags are handled after save; pop them here to pass to _save_tags() + tags = validated_data.pop('tags', None) + + # _adopt_components and _disable_replication must be set on the instance before + # save() is called, so we cannot delegate to super().create() here. + instance = self.Meta.model(**validated_data) + if adopt_components: + instance._adopt_components = True + if not replicate_components: + instance._disable_replication = True + instance.save() + + if tags is not None: + self._save_tags(instance, tags) + + return instance + class MACAddressSerializer(PrimaryModelSerializer): assigned_object_type = ContentTypeField( diff --git a/netbox/dcim/tests/test_api.py b/netbox/dcim/tests/test_api.py index 70c212849..f574507f1 100644 --- a/netbox/dcim/tests/test_api.py +++ b/netbox/dcim/tests/test_api.py @@ -1699,6 +1699,238 @@ class ModuleTest(APIViewTestCases.APIViewTestCase): }, ] + def test_replicate_components(self): + """ + Installing a module with replicate_components=True (the default) should create + components from the module type's templates on the parent device. + """ + self.add_permissions('dcim.add_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for Replication Test') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Replication Test Module Type') + InterfaceTemplate.objects.create(module_type=module_type, name='eth0', type='1000base-t') + module_bay = ModuleBay.objects.create(device=device, name='Replication Bay') + + url = reverse('dcim-api:module-list') + data = { + 'device': device.pk, + 'module_bay': module_bay.pk, + 'module_type': module_type.pk, + 'replicate_components': True, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_201_CREATED) + self.assertTrue(device.interfaces.filter(name='eth0').exists()) + + def test_no_replicate_components(self): + """ + Installing a module with replicate_components=False should NOT create components + from the module type's templates. + """ + self.add_permissions('dcim.add_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for No Replication Test') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='No Replication Test Module Type') + InterfaceTemplate.objects.create(module_type=module_type, name='eth0', type='1000base-t') + module_bay = ModuleBay.objects.create(device=device, name='No Replication Bay') + + url = reverse('dcim-api:module-list') + data = { + 'device': device.pk, + 'module_bay': module_bay.pk, + 'module_type': module_type.pk, + 'replicate_components': False, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_201_CREATED) + self.assertFalse(device.interfaces.filter(name='eth0').exists()) + + def test_adopt_components(self): + """ + Installing a module with adopt_components=True should assign existing unattached + device components to the new module. + """ + self.add_permissions('dcim.add_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for Adopt Test') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Adopt Test Module Type') + InterfaceTemplate.objects.create(module_type=module_type, name='eth0', type='1000base-t') + module_bay = ModuleBay.objects.create(device=device, name='Adopt Bay') + existing_iface = Interface.objects.create(device=device, name='eth0', type='1000base-t') + + url = reverse('dcim-api:module-list') + data = { + 'device': device.pk, + 'module_bay': module_bay.pk, + 'module_type': module_type.pk, + 'adopt_components': True, + 'replicate_components': False, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_201_CREATED) + existing_iface.refresh_from_db() + self.assertIsNotNone(existing_iface.module) + + def test_replicate_components_conflict(self): + """ + Installing a module with replicate_components=True when a component with the same name + already exists should return a validation error. + """ + self.add_permissions('dcim.add_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for Conflict Test') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Conflict Test Module Type') + InterfaceTemplate.objects.create(module_type=module_type, name='eth0', type='1000base-t') + module_bay = ModuleBay.objects.create(device=device, name='Conflict Bay') + Interface.objects.create(device=device, name='eth0', type='1000base-t') + + url = reverse('dcim-api:module-list') + data = { + 'device': device.pk, + 'module_bay': module_bay.pk, + 'module_type': module_type.pk, + 'replicate_components': True, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + + def test_adopt_components_already_owned(self): + """ + Installing a module with adopt_components=True when an existing component already + belongs to another module should return a validation error. + """ + self.add_permissions('dcim.add_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for Adopt Owned Test') + owner_module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Owner Module Type') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Adopt Owned Test Module Type') + InterfaceTemplate.objects.create(module_type=module_type, name='eth0', type='1000base-t') + owner_bay = ModuleBay.objects.create(device=device, name='Owner Bay') + target_bay = ModuleBay.objects.create(device=device, name='Adopt Owned Bay') + + # Install a module that owns the interface + owner_module = Module.objects.create(device=device, module_bay=owner_bay, module_type=owner_module_type) + Interface.objects.create(device=device, name='eth0', type='1000base-t', module=owner_module) + + url = reverse('dcim-api:module-list') + data = { + 'device': device.pk, + 'module_bay': target_bay.pk, + 'module_type': module_type.pk, + 'adopt_components': True, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + + def test_patch_ignores_replicate_and_adopt(self): + """ + PATCH requests that include replicate_components or adopt_components should not + trigger component replication or adoption (these fields are create-only). + """ + self.add_permissions('dcim.change_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for PATCH Test') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='PATCH Test Module Type') + InterfaceTemplate.objects.create(module_type=module_type, name='eth0', type='1000base-t') + module_bay = ModuleBay.objects.create(device=device, name='PATCH Bay') + # Create the module without replication so we can verify PATCH doesn't trigger it + module = Module(device=device, module_bay=module_bay, module_type=module_type) + module._disable_replication = True + module.save() + + url = reverse('dcim-api:module-detail', kwargs={'pk': module.pk}) + data = { + 'replicate_components': True, + 'adopt_components': True, + 'serial': 'PATCHED', + } + response = self.client.patch(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + self.assertEqual(response.data['serial'], 'PATCHED') + # No interfaces should have been created by the PATCH + self.assertFalse(device.interfaces.exists()) + + def test_adopt_and_replicate_components(self): + """ + Installing a module with both adopt_components=True and replicate_components=True + should adopt existing unowned components and create new components for templates + that have no matching existing component. + """ + self.add_permissions('dcim.add_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for Adopt+Replicate Test') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Adopt+Replicate Test Module Type') + InterfaceTemplate.objects.create(module_type=module_type, name='eth0', type='1000base-t') + InterfaceTemplate.objects.create(module_type=module_type, name='eth1', type='1000base-t') + module_bay = ModuleBay.objects.create(device=device, name='Adopt+Replicate Bay') + # eth0 already exists (unowned); eth1 does not + existing_iface = Interface.objects.create(device=device, name='eth0', type='1000base-t') + + url = reverse('dcim-api:module-list') + data = { + 'device': device.pk, + 'module_bay': module_bay.pk, + 'module_type': module_type.pk, + 'adopt_components': True, + 'replicate_components': True, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_201_CREATED) + # eth0 should have been adopted (now owned by the new module) + existing_iface.refresh_from_db() + self.assertIsNotNone(existing_iface.module) + # eth1 should have been created + self.assertTrue(device.interfaces.filter(name='eth1').exists()) + + def test_module_token_no_position(self): + """ + Installing a module whose type has a template with a MODULE_TOKEN placeholder into a + module bay with no position defined should return a validation error. + """ + self.add_permissions('dcim.add_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for Token No-Position Test') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Token No-Position Module Type') + # Template name contains the MODULE_TOKEN placeholder + InterfaceTemplate.objects.create( + module_type=module_type, name=f'{MODULE_TOKEN}-eth0', type='1000base-t' + ) + # Module bay has no position + module_bay = ModuleBay.objects.create(device=device, name='No-Position Bay') + + url = reverse('dcim-api:module-list') + data = { + 'device': device.pk, + 'module_bay': module_bay.pk, + 'module_type': module_type.pk, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + + def test_module_token_depth_mismatch(self): + """ + Installing a module whose template name has more MODULE_TOKEN placeholders than the + depth of the module bay tree should return a validation error. + """ + self.add_permissions('dcim.add_module') + manufacturer = Manufacturer.objects.get(name='Generic') + device = create_test_device('Device for Token Depth Mismatch Test') + module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Token Depth Mismatch Module Type') + # Template name has two placeholders but the bay is at depth 1 + InterfaceTemplate.objects.create( + module_type=module_type, name=f'{MODULE_TOKEN}-{MODULE_TOKEN}-eth0', type='1000base-t' + ) + module_bay = ModuleBay.objects.create(device=device, name='Depth 1 Bay', position='1') + + url = reverse('dcim-api:module-list') + data = { + 'device': device.pk, + 'module_bay': module_bay.pk, + 'module_type': module_type.pk, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + class ConsolePortTest(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTestCase): model = ConsolePort