Address review feedback

docs/configuration/index.md

@@ -21,7 +21,6 @@ Some configuration parameters are primarily controlled via NetBox's admin interf
 * [`BANNER_BOTTOM`](./miscellaneous.md#banner_bottom)
 * [`BANNER_LOGIN`](./miscellaneous.md#banner_login)
 * [`BANNER_TOP`](./miscellaneous.md#banner_top)
-* [`CHANGELOG_RETAIN_CREATE_LAST_UPDATE`](./miscellaneous.md#changelog_retain_create_last_update)
 * [`CHANGELOG_RETENTION`](./miscellaneous.md#changelog_retention)
 * [`CUSTOM_VALIDATORS`](./data-validation.md#custom_validators)
 * [`DEFAULT_USER_PREFERENCES`](./default-values.md#default_user_preferences)

docs/configuration/miscellaneous.md

@@ -73,27 +73,6 @@ This data enables the project maintainers to estimate how many NetBox deployment
 
 ---
 
-## CHANGELOG_RETAIN_CREATE_LAST_UPDATE
-
-!!! tip "Dynamic Configuration Parameter"
-
-Default: `True`
-
-When pruning expired changelog entries (per `CHANGELOG_RETENTION`), retain each non-deleted object's original `create`
-change record and its most recent `update` change record. If an object has a `delete` change record, its changelog
-entries are pruned normally according to `CHANGELOG_RETENTION`.
-
-!!! note
-    For objects without a `delete` change record, the original `create` record and most recent `update` record are
-    exempt from pruning. All other changelog records (including intermediate `update` records and all `delete` records)
-    remain subject to pruning per `CHANGELOG_RETENTION`.
-
-!!! warning
-    This setting is enabled by default. Upgrading deployments that rely on complete pruning of expired changelog entries
-    should explicitly set `CHANGELOG_RETAIN_CREATE_LAST_UPDATE = False` to preserve the previous behavior.
-
----
-
 ## CHANGELOG_RETENTION
 
 !!! tip "Dynamic Configuration Parameter"

netbox/core/forms/model_forms.py

@@ -165,10 +165,9 @@ class ConfigRevisionForm(forms.ModelForm, metaclass=ConfigFormMetaclass):
         FieldSet('PAGINATE_COUNT', 'MAX_PAGE_SIZE', name=_('Pagination')),
         FieldSet('CUSTOM_VALIDATORS', 'PROTECTION_RULES', name=_('Validation')),
         FieldSet('DEFAULT_USER_PREFERENCES', name=_('User Preferences')),
-        FieldSet('CHANGELOG_RETENTION', 'CHANGELOG_RETAIN_CREATE_LAST_UPDATE', name=_('Change Log')),
         FieldSet(
-            'MAINTENANCE_MODE', 'COPILOT_ENABLED', 'GRAPHQL_ENABLED', 'JOB_RETENTION', 'MAPS_URL',
-            name=_('Miscellaneous'),
+            'MAINTENANCE_MODE', 'COPILOT_ENABLED', 'GRAPHQL_ENABLED', 'CHANGELOG_RETENTION', 'JOB_RETENTION',
+            'MAPS_URL', name=_('Miscellaneous'),
         ),
         FieldSet('comment', name=_('Config Revision'))
     )

netbox/core/jobs.py

@@ -5,7 +5,6 @@ from importlib import import_module
 import requests
 from django.conf import settings
 from django.core.cache import cache
-from django.db.models import Exists, OuterRef, Subquery
 from django.utils import timezone
 from packaging import version
 
@@ -15,7 +14,7 @@ from netbox.jobs import JobRunner, system_job
 from netbox.search.backends import search_backend
 from utilities.proxy import resolve_proxies
 
-from .choices import DataSourceStatusChoices, JobIntervalChoices, ObjectChangeActionChoices
+from .choices import DataSourceStatusChoices, JobIntervalChoices
 from .models import DataSource
 
 
@@ -127,51 +126,19 @@ class SystemHousekeepingJob(JobRunner):
         """
         Delete any ObjectChange records older than the configured changelog retention time (if any).
         """
-        self.logger.info('Pruning old changelog entries...')
+        self.logger.info("Pruning old changelog entries...")
         config = Config()
         if not config.CHANGELOG_RETENTION:
-            self.logger.info('No retention period specified; skipping.')
+            self.logger.info("No retention period specified; skipping.")
             return
 
         cutoff = timezone.now() - timedelta(days=config.CHANGELOG_RETENTION)
-        self.logger.debug(f'Changelog retention period: {config.CHANGELOG_RETENTION} days ({cutoff:%Y-%m-%d %H:%M:%S})')
+        self.logger.debug(
+            f"Changelog retention period: {config.CHANGELOG_RETENTION} days ({cutoff:%Y-%m-%d %H:%M:%S})"
+        )
 
-        expired_qs = ObjectChange.objects.filter(time__lt=cutoff)
-
-        # When enabled, retain each object's original create record and most recent update record while pruning expired
-        # changelog entries. This applies only to objects without a delete record.
-        if config.CHANGELOG_RETAIN_CREATE_LAST_UPDATE:
-            self.logger.debug('Retaining changelog create records and last update records (excluding deleted objects)')
-
-            deleted_exists = ObjectChange.objects.filter(
-                action=ObjectChangeActionChoices.ACTION_DELETE,
-                changed_object_type_id=OuterRef('changed_object_type_id'),
-                changed_object_id=OuterRef('changed_object_id'),
-            )
-
-            # Keep create records only where no delete exists for that object
-            create_pks_to_keep = (
-                ObjectChange.objects.filter(action=ObjectChangeActionChoices.ACTION_CREATE)
-                .annotate(has_delete=Exists(deleted_exists))
-                .filter(has_delete=False)
-                .values('pk')
-            )
-
-            # Keep the most recent update per object only where no delete exists for the object
-            latest_update_pks_to_keep = (
-                ObjectChange.objects.filter(action=ObjectChangeActionChoices.ACTION_UPDATE)
-                .annotate(has_delete=Exists(deleted_exists))
-                .filter(has_delete=False)
-                .order_by('changed_object_type_id', 'changed_object_id', '-time', '-pk')
-                .distinct('changed_object_type_id', 'changed_object_id')
-                .values('pk')
-            )
-
-            expired_qs = expired_qs.exclude(pk__in=Subquery(create_pks_to_keep))
-            expired_qs = expired_qs.exclude(pk__in=Subquery(latest_update_pks_to_keep))
-
-        count = expired_qs.delete()[0]
-        self.logger.info(f'Deleted {count} expired changelog records')
+        count = ObjectChange.objects.filter(time__lt=cutoff).delete()[0]
+        self.logger.info(f"Deleted {count} expired changelog records")
 
     def delete_expired_jobs(self):
         """

netbox/core/tests/test_changelog.py

@@ -1,16 +1,9 @@
-import logging
-import uuid
-from datetime import timedelta
-from unittest.mock import patch
-
 from django.contrib.contenttypes.models import ContentType
-from django.test import TestCase, override_settings
+from django.test import override_settings
 from django.urls import reverse
-from django.utils import timezone
 from rest_framework import status
 
 from core.choices import ObjectChangeActionChoices
-from core.jobs import SystemHousekeepingJob
 from core.models import ObjectChange, ObjectType
 from dcim.choices import InterfaceTypeChoices, ModuleStatusChoices, SiteStatusChoices
 from dcim.models import (
@@ -701,99 +694,3 @@ class ChangeLogAPITest(APITestCase):
         self.assertEqual(changes[3].changed_object_type, ContentType.objects.get_for_model(Module))
         self.assertEqual(changes[3].changed_object_id, module.pk)
         self.assertEqual(changes[3].action, ObjectChangeActionChoices.ACTION_DELETE)
-
-
-class ChangelogPruneRetentionTest(TestCase):
-    """Test suite for Changelog pruning retention settings."""
-
-    @staticmethod
-    def _make_oc(*, ct, obj_id, action, ts):
-        oc = ObjectChange.objects.create(
-            changed_object_type=ct,
-            changed_object_id=obj_id,
-            action=action,
-            user_name='test',
-            request_id=uuid.uuid4(),
-            object_repr=f'Object {obj_id}',
-        )
-        ObjectChange.objects.filter(pk=oc.pk).update(time=ts)
-        return oc.pk
-
-    @staticmethod
-    def _run_prune(*, retention_days, retain_create_last_update):
-        job = SystemHousekeepingJob.__new__(SystemHousekeepingJob)
-        job.logger = logging.getLogger('netbox.tests.changelog_prune')
-
-        with patch('core.jobs.Config') as MockConfig:
-            cfg = MockConfig.return_value
-            cfg.CHANGELOG_RETENTION = retention_days
-            cfg.CHANGELOG_RETAIN_CREATE_LAST_UPDATE = retain_create_last_update
-            job.prune_changelog()
-
-    def test_prune_retain_create_last_update_excludes_deleted_objects(self):
-        ct = ContentType.objects.get_for_model(Site)
-
-        retention_days = 90
-        now = timezone.now()
-        cutoff = now - timedelta(days=retention_days)
-
-        expired_old = cutoff - timedelta(days=10)
-        expired_newer = cutoff - timedelta(days=1)
-        not_expired = cutoff + timedelta(days=1)
-
-        # A) Not deleted: should keep CREATE + latest UPDATE, prune intermediate UPDATEs
-        a_create = self._make_oc(ct=ct, obj_id=1, action=ObjectChangeActionChoices.ACTION_CREATE, ts=expired_old)
-        a_update1 = self._make_oc(ct=ct, obj_id=1, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired_old)
-        a_update2 = self._make_oc(ct=ct, obj_id=1, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired_newer)
-
-        # B) Deleted (all expired): should keep NOTHING
-        b_create = self._make_oc(ct=ct, obj_id=2, action=ObjectChangeActionChoices.ACTION_CREATE, ts=expired_old)
-        b_update = self._make_oc(ct=ct, obj_id=2, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired_newer)
-        b_delete = self._make_oc(ct=ct, obj_id=2, action=ObjectChangeActionChoices.ACTION_DELETE, ts=expired_newer)
-
-        # C) Deleted but delete is not expired: create/update expired should be pruned; delete remains
-        c_create = self._make_oc(ct=ct, obj_id=3, action=ObjectChangeActionChoices.ACTION_CREATE, ts=expired_old)
-        c_update = self._make_oc(ct=ct, obj_id=3, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired_newer)
-        c_delete = self._make_oc(ct=ct, obj_id=3, action=ObjectChangeActionChoices.ACTION_DELETE, ts=not_expired)
-
-        self._run_prune(retention_days=retention_days, retain_create_last_update=True)
-
-        remaining = set(ObjectChange.objects.values_list('pk', flat=True))
-
-        # A) Not deleted -> create + latest update remain
-        self.assertIn(a_create, remaining)
-        self.assertIn(a_update2, remaining)
-        self.assertNotIn(a_update1, remaining)
-
-        # B) Deleted (all expired) -> nothing remains
-        self.assertNotIn(b_create, remaining)
-        self.assertNotIn(b_update, remaining)
-        self.assertNotIn(b_delete, remaining)
-
-        # C) Deleted, delete not expired -> delete remains, but create/update are pruned
-        self.assertNotIn(c_create, remaining)
-        self.assertNotIn(c_update, remaining)
-        self.assertIn(c_delete, remaining)
-
-    def test_prune_disabled_deletes_all_expired(self):
-        ct = ContentType.objects.get_for_model(Site)
-
-        retention_days = 90
-        now = timezone.now()
-        cutoff = now - timedelta(days=retention_days)
-        expired = cutoff - timedelta(days=1)
-        not_expired = cutoff + timedelta(days=1)
-
-        # expired create/update should be deleted when feature disabled
-        x_create = self._make_oc(ct=ct, obj_id=10, action=ObjectChangeActionChoices.ACTION_CREATE, ts=expired)
-        x_update = self._make_oc(ct=ct, obj_id=10, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired)
-
-        # non-expired delete should remain regardless
-        y_delete = self._make_oc(ct=ct, obj_id=11, action=ObjectChangeActionChoices.ACTION_DELETE, ts=not_expired)
-
-        self._run_prune(retention_days=retention_days, retain_create_last_update=False)
-
-        remaining = set(ObjectChange.objects.values_list('pk', flat=True))
-        self.assertNotIn(x_create, remaining)
-        self.assertNotIn(x_update, remaining)
-        self.assertIn(y_delete, remaining)

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,144 @@ 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 {tokens} placeholder(s) in a module bay at depth {level}."
+                            ).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 replicate_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,189 @@ 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())
+
 
 class ConsolePortTest(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTestCase):
     model = ConsolePort

netbox/extras/management/commands/housekeeping.py

@@ -6,11 +6,9 @@ import requests
 from django.conf import settings
 from django.core.cache import cache
 from django.core.management.base import BaseCommand
-from django.db.models import Exists, OuterRef, Subquery
 from django.utils import timezone
 from packaging import version
 
-from core.choices import ObjectChangeActionChoices
 from core.models import Job, ObjectChange
 from netbox.config import Config
 from utilities.proxy import resolve_proxies
@@ -49,63 +47,29 @@ class Command(BaseCommand):
 
         # Delete expired ObjectChanges
         if options['verbosity']:
-            self.stdout.write('[*] Checking for expired changelog records')
+            self.stdout.write("[*] Checking for expired changelog records")
         if config.CHANGELOG_RETENTION:
             cutoff = timezone.now() - timedelta(days=config.CHANGELOG_RETENTION)
             if options['verbosity'] >= 2:
-                self.stdout.write(f'\tRetention period: {config.CHANGELOG_RETENTION} days')
-                self.stdout.write(f'\tCut-off time: {cutoff}')
-
-            expired_qs = ObjectChange.objects.filter(time__lt=cutoff)
-
-            # When enabled, retain each object's original create and most recent update record while pruning expired
-            # changelog entries. This applies only to objects without a delete record.
-            if config.CHANGELOG_RETAIN_CREATE_LAST_UPDATE:
-                if options['verbosity'] >= 2:
-                    self.stdout.write('\tRetaining create & last update records for non-deleted objects')
-
-                deleted_exists = ObjectChange.objects.filter(
-                    action=ObjectChangeActionChoices.ACTION_DELETE,
-                    changed_object_type_id=OuterRef('changed_object_type_id'),
-                    changed_object_id=OuterRef('changed_object_id'),
-                )
-
-                # Keep create records only where no delete exists for that object
-                create_pks_to_keep = (
-                    ObjectChange.objects.filter(action=ObjectChangeActionChoices.ACTION_CREATE)
-                    .annotate(has_delete=Exists(deleted_exists))
-                    .filter(has_delete=False)
-                    .values('pk')
-                )
-
-                # Keep the most recent update per object only where no delete exists for the object
-                latest_update_pks_to_keep = (
-                    ObjectChange.objects.filter(action=ObjectChangeActionChoices.ACTION_UPDATE)
-                    .annotate(has_delete=Exists(deleted_exists))
-                    .filter(has_delete=False)
-                    .order_by('changed_object_type_id', 'changed_object_id', '-time', '-pk')
-                    .distinct('changed_object_type_id', 'changed_object_id')
-                    .values('pk')
-                )
-
-                expired_qs = expired_qs.exclude(pk__in=Subquery(create_pks_to_keep))
-                expired_qs = expired_qs.exclude(pk__in=Subquery(latest_update_pks_to_keep))
-
-            expired_records = expired_qs.count()
+                self.stdout.write(f"\tRetention period: {config.CHANGELOG_RETENTION} days")
+                self.stdout.write(f"\tCut-off time: {cutoff}")
+            expired_records = ObjectChange.objects.filter(time__lt=cutoff).count()
             if expired_records:
                 if options['verbosity']:
                     self.stdout.write(
-                        f'\tDeleting {expired_records} expired records... ', self.style.WARNING, ending=''
+                        f"\tDeleting {expired_records} expired records... ",
+                        self.style.WARNING,
+                        ending=""
                     )
                     self.stdout.flush()
-                expired_qs.delete()
+                ObjectChange.objects.filter(time__lt=cutoff).delete()
                 if options['verbosity']:
-                    self.stdout.write('Done.', self.style.SUCCESS)
+                    self.stdout.write("Done.", self.style.SUCCESS)
             elif options['verbosity']:
-                self.stdout.write('\tNo expired records found.', self.style.SUCCESS)
+                self.stdout.write("\tNo expired records found.", self.style.SUCCESS)
         elif options['verbosity']:
             self.stdout.write(
-                f'\tSkipping: No retention period specified (CHANGELOG_RETENTION = {config.CHANGELOG_RETENTION})'
+                f"\tSkipping: No retention period specified (CHANGELOG_RETENTION = {config.CHANGELOG_RETENTION})"
             )
 
         # Delete expired Jobs

netbox/netbox/config/__init__.py

@@ -10,7 +10,6 @@ from .parameters import PARAMS
 
 __all__ = (
     'PARAMS',
-    'Config',
     'ConfigItem',
     'clear_config',
     'get_config',

netbox/netbox/config/parameters.py

@@ -175,25 +175,6 @@ PARAMS = (
         field=forms.JSONField
     ),
 
-    # Change log
-    ConfigParam(
-        name='CHANGELOG_RETENTION',
-        label=_('Changelog retention'),
-        default=90,
-        description=_("Days to retain changelog history (set to zero for unlimited)"),
-        field=forms.IntegerField,
-    ),
-    ConfigParam(
-        name='CHANGELOG_RETAIN_CREATE_LAST_UPDATE',
-        label=_('Retain create & last update changelog records'),
-        default=True,
-        description=_(
-            "Retain each object's create record and most recent update record when pruning expired changelog entries "
-            "(excluding objects with a delete record)."
-        ),
-        field=forms.BooleanField,
-    ),
-
     # Miscellaneous
     ConfigParam(
         name='MAINTENANCE_MODE',
@@ -218,6 +199,13 @@ PARAMS = (
         description=_("Enable the GraphQL API"),
         field=forms.BooleanField
     ),
+    ConfigParam(
+        name='CHANGELOG_RETENTION',
+        label=_('Changelog retention'),
+        default=90,
+        description=_("Days to retain changelog history (set to zero for unlimited)"),
+        field=forms.IntegerField
+    ),
     ConfigParam(
         name='JOB_RETENTION',
         label=_('Job result retention'),

netbox/templates/core/inc/config_data.html

@@ -122,19 +122,6 @@
     {% endif %}
   </tr>
 
-  {# Changelog #}
-  <tr>
-    <td colspan="2" class="bg-secondary-subtle fs-5 fw-bold border-0 py-1">{% trans "Change log" %}</td>
-  </tr>
-  <tr>
-    <th scope="row" class="ps-3">{% trans "Changelog retention" %}</th>
-    <td>{{ config.CHANGELOG_RETENTION }}</td>
-  </tr>
-  <tr>
-    <th scope="row" class="ps-3">{% trans "Changelog retain create & last update records" %}</th>
-    <td>{% checkmark config.CHANGELOG_RETAIN_CREATE_LAST_UPDATE %}</td>
-  </tr>
-
   {# Miscellaneous #}
   <tr>
     <td colspan="2" class="bg-secondary-subtle fs-5 fw-bold border-0 py-1">{% trans "Miscellaneous" %}</td>
@@ -150,6 +137,10 @@
     <th scope="row" class="ps-3">{% trans "GraphQL enabled" %}</th>
     <td>{% checkmark config.GRAPHQL_ENABLED %}</td>
   </tr>
+  <tr>
+    <th scope="row" class="ps-3">{% trans "Changelog retention" %}</th>
+    <td>{{ config.CHANGELOG_RETENTION }}</td>
+  </tr>
   <tr>
     <th scope="row" class="ps-3">{% trans "Job retention" %}</th>
     <td>{{ config.JOB_RETENTION }}</td>

netbox/templates/core/objectchange_list.html

@@ -6,6 +6,6 @@
 {% block content %}
   {{ block.super }}
   <div class="text-muted px-3">
-    {% trans "Change log retention" %}: {% if config.CHANGELOG_RETENTION %}{{ config.CHANGELOG_RETENTION }} {% trans "days" %}{% if config.CHANGELOG_RETAIN_CREATE_LAST_UPDATE %} ({% trans "retaining create & last update records for non-deleted objects" %}){% endif %}{% else %}{% trans "Indefinite" %}{% endif %}
+    {% trans "Change log retention" %}: {% if config.CHANGELOG_RETENTION %}{{ config.CHANGELOG_RETENTION }} {% trans "days" %}{% else %}{% trans "Indefinite" %}{% endif %}
   </div>
 {% endblock content %}

netbox/templates/extras/object_changelog.html

@@ -12,7 +12,7 @@
         </div>
       </div>
       <div class="text-muted">
-        {% trans "Change log retention" %}: {% if config.CHANGELOG_RETENTION %}{{ config.CHANGELOG_RETENTION }} {% trans "days" %}{% if config.CHANGELOG_RETAIN_CREATE_LAST_UPDATE %} ({% trans "retaining create & last update records for non-deleted objects" %}){% endif %}{% else %}{% trans "Indefinite" %}{% endif %}
+        {% trans "Change log retention" %}: {% if config.CHANGELOG_RETENTION %}{{ config.CHANGELOG_RETENTION }} {% trans "days" %}{% else %}{% trans "Indefinite" %}{% endif %}
       </div>
     </div>
   </div>