Fixes #23617: Snapshot objects before saving updates made on their behalf

netbox/dcim/models/cables.py

@@ -395,6 +395,8 @@ def save(self, *args, force_insert=False, force_update=False, using=None, update
             super().save(*args, force_insert=True, **save_kwargs)
             # Update the private PK used in __str__()
             self._pk = self.pk
+            # Record the pre-termination state for the update made by the save below
+            self.snapshot()
 
         if profile_written and self._orig_profile != self.profile:
             self.update_terminations(force=True)

netbox/dcim/models/devices.py

@@ -1175,6 +1175,7 @@ def save(self, *args, **kwargs):
         # Update Site and Rack assignment for any child Devices
         devices = Device.objects.filter(parent_bay__device=self)
         for device in devices:
+            device.snapshot()
             device.site = self.site
             device.rack = self.rack
             device.location = self.location

netbox/dcim/signals.py

@@ -498,5 +498,6 @@ def update_mac_address_interface(instance, created, raw, **kwargs):
     assign the MACAddress to the interface.
     """
     if created and not raw and instance.primary_mac_address:
+        instance.primary_mac_address.snapshot()
         instance.primary_mac_address.assigned_object = instance
         instance.primary_mac_address.save()

netbox/dcim/tests/test_models.py

@@ -54,6 +54,26 @@ def setUpTestData(cls):
         cls.interface.primary_mac_address = cls.mac_a
         cls.interface.save()
 
+    @tag('regression')  # Ref: #23617
+    def test_primary_mac_assignment_on_interface_creation_records_prechange(self):
+        mac = MACAddress.objects.create(mac_address='1234567890cd')
+        request = RequestFactory().get('/')
+        request.id = uuid.uuid4()
+        request.user = User.objects.create_user(username='testuser')
+
+        with event_tracking(request):
+            interface = Interface.objects.create(
+                device=self.interface.device, name='Interface 2', type=InterfaceTypeChoices.TYPE_1GE_FIXED,
+                primary_mac_address=mac
+            )
+
+        change = ObjectChange.objects.get(
+            changed_object_type=ObjectType.objects.get_for_model(MACAddress),
+            changed_object_id=mac.pk,
+        )
+        self.assertIsNone(change.prechange_data['assigned_object_id'])
+        self.assertEqual(change.postchange_data['assigned_object_id'], interface.pk)
+
     @tag('regression')
     def test_clean_will_not_allow_removal_of_assigned_object_if_primary(self):
         self.mac_a.assigned_object = None
@@ -1325,6 +1345,29 @@ def test_cannot_install_device_in_disabled_bay(self):
         self.assertIn('installed_device', cm.exception.message_dict)
         self.assertIn('disabled device bay', str(cm.exception.message_dict['installed_device']))
 
+    @tag('regression')  # Ref: #23617
+    def test_moving_parent_device_records_child_prechange(self):
+        DeviceBay.objects.create(device=self.parent_device, name='Bay 1', installed_device=self.child_device)
+        old_site = self.parent_device.site
+        new_site = Site.objects.create(name='Test Site 2', slug='test-site-2')
+        request = RequestFactory().get('/')
+        request.id = uuid.uuid4()
+        request.user = User.objects.create_user(username='testuser')
+
+        with event_tracking(request):
+            parent = Device.objects.get(pk=self.parent_device.pk)
+            parent.snapshot()
+            parent.site = new_site
+            parent.save()
+
+        change = ObjectChange.objects.get(
+            changed_object_type=ObjectType.objects.get_for_model(Device),
+            changed_object_id=self.child_device.pk,
+            action=ObjectChangeActionChoices.ACTION_UPDATE,
+        )
+        self.assertEqual(change.prechange_data['site'], old_site.pk)
+        self.assertEqual(change.postchange_data['site'], new_site.pk)
+
     def test_can_disable_bay_with_existing_device(self):
         """
         Test that disabling a bay that already has a device installed does NOT raise an error
@@ -3403,6 +3446,17 @@ def _updates(self, obj):
             action=ObjectChangeActionChoices.ACTION_UPDATE,
         ).order_by('time')
 
+    @tag('regression')  # Ref: #23617
+    def test_cable_creation_records_termination_prechange(self):
+        cable = self._connect(self.interface1, self.interface2)
+
+        # The update recording the terminations follows the creation of the Cable itself
+        change = self._updates(cable).get()
+        self.assertEqual(change.prechange_data['a_terminations'], [])
+        self.assertEqual(change.prechange_data['b_terminations'], [])
+        self.assertEqual(len(change.postchange_data['a_terminations']), 1)
+        self.assertEqual(len(change.postchange_data['b_terminations']), 1)
+
     @tag('regression')  # Ref: netbox-branching#631
     def test_cable_deletion_records_disconnect(self):
         cable = self._connect(self.interface1, self.interface2)
@@ -3480,6 +3534,7 @@ def test_termination_removal_records_disconnect(self):
         request = self._request()
         with event_tracking(request):
             cable = Cable.objects.get(pk=cable_pk)
+            cable.snapshot()
             cable.b_terminations = [self.interface3]
             cable.save()
 

netbox/netbox/models/features.py

@@ -702,6 +702,9 @@ def sync(self, save=False):
 
         :param save: If true, save() will be called after data has been synchronized
         """
+        # ManagedFile is not change-logged and so has no snapshot()
+        if save and hasattr(self, 'snapshot'):
+            self.snapshot()
         self.sync_data()
         self.data_synced = timezone.now()
         if save:

netbox/netbox/tests/test_model_features.py

@@ -1,15 +1,17 @@
 import hashlib
+import uuid
 from unittest import skipIf
+from unittest.mock import patch
 
 from django.apps import apps
 from django.conf import settings
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ValidationError
-from django.test import TestCase
+from django.test import RequestFactory, TestCase
 from django.utils.timezone import now
 from taggit.models import Tag
 
-from core.models import AutoSyncRecord, DataFile, DataSource
+from core.models import AutoSyncRecord, DataFile, DataSource, ManagedFile, ObjectChange
 from dcim.models import Site
 from extras.models import (
     ConfigContext,
@@ -20,7 +22,9 @@
 )
 from ipam.models import Prefix
 from netbox.constants import CORE_APPS
+from netbox.context_managers import event_tracking
 from netbox.models.features import CloningMixin, get_model_features, has_feature, model_is_public
+from users.models import User
 
 
 class ModelFeaturesTestCase(TestCase):
@@ -149,6 +153,39 @@ def test_synceddatamixin_synchronizes_empty_object(self):
                 obj.refresh_from_db()
                 self.assertEqual(getattr(obj, field_name), {})
 
+    def test_synceddatamixin_sync_records_prechange(self):
+        """Saving a sync records the object's state before the synchronized data was applied. Ref: #23617"""
+        datafile = self._create_data_file('template.j2', b'original\n')
+        obj = ConfigTemplate(name='Test', data_file=datafile)
+        obj.clean()
+        obj.save()
+        DataFile.objects.filter(pk=datafile.pk).update(data=b'updated\n')
+        request = RequestFactory().get('/')
+        request.id = uuid.uuid4()
+        request.user = User.objects.create_user(username='testuser')
+
+        with event_tracking(request):
+            ConfigTemplate.objects.get(pk=obj.pk).sync(save=True)
+
+        change = ObjectChange.objects.get(
+            changed_object_type=ContentType.objects.get_for_model(ConfigTemplate),
+            changed_object_id=obj.pk,
+        )
+        self.assertEqual(change.prechange_data['template_code'], 'original\n')
+        self.assertEqual(change.postchange_data['template_code'], 'updated\n')
+
+    @patch.object(ManagedFile, 'sync_data')
+    def test_synceddatamixin_sync_without_change_logging(self, mock_sync_data):
+        """Saving a sync succeeds for models which are not change-logged. Ref: #23617"""
+        datafile = self._create_data_file('script.py', b'# script\n')
+        obj = ManagedFile.objects.create(file_root='scripts', file_path='script.py', data_file=datafile)
+
+        obj.sync(save=True)
+
+        mock_sync_data.assert_called_once()
+        obj.refresh_from_db()
+        self.assertIsNotNone(obj.data_synced)
+
     def test_synceddatamixin_accepts_plain_text_template(self):
         """Text which is not a YAML document is still valid template content."""
         datafile = self._create_data_file('comment-only.j2', b'{# nothing here #}\n')

netbox/wireless/signals.py

@@ -27,10 +27,12 @@ def update_connected_interfaces(instance, created, raw=False, **kwargs):
 
     if instance.interface_a.wireless_link != instance:
         logger.debug(f"Updating interface A for wireless link {instance}")
+        instance.interface_a.snapshot()
         instance.interface_a.wireless_link = instance
         instance.interface_a.save()
     if instance.interface_b.wireless_link != instance:
         logger.debug(f"Updating interface B for wireless link {instance}")
+        instance.interface_b.snapshot()
         instance.interface_b.wireless_link = instance
         instance.interface_b.save()
 

netbox/wireless/tests/test_models.py

@@ -66,6 +66,31 @@ def setUpTestData(cls):
         cls.device = create_test_device('Device 1')
         cls.user = User.objects.create_user(username='testuser')
 
+    # Regression test for #23617
+    def test_creating_wireless_link_records_interface_prechange(self):
+        interfaces = [
+            Interface.objects.create(device=self.device, name=name, type=InterfaceTypeChoices.TYPE_80211AC)
+            for name in ('radio0', 'radio1')
+        ]
+        request = RequestFactory().post('/')
+        request.id = uuid4()
+        request.user = self.user
+
+        with event_tracking(request):
+            wireless_link = WirelessLink.objects.create(
+                interface_a=interfaces[0],
+                interface_b=interfaces[1],
+                ssid='LINK1',
+            )
+
+        for interface in interfaces:
+            change = ObjectChange.objects.get(
+                changed_object_type=ContentType.objects.get_for_model(Interface),
+                changed_object_id=interface.pk,
+            )
+            self.assertIsNone(change.prechange_data['wireless_link'])
+            self.assertEqual(change.postchange_data['wireless_link'], wireless_link.pk)
+
     def test_resaving_wireless_link_does_not_log_connected_interfaces(self):
         interface_a = Interface.objects.create(
             device=self.device,