Fixes: #20123 - Add replicate_components and adopt_components write_only fields to ModuleSerializer (#21600)

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(

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