review feedback

netbox/circuits/tests/test_models.py

@@ -367,6 +367,7 @@ class CircuitTerminationChangeLoggingTestCase(TestCase):
         ObjectChange.objects.all().delete()
 
         def _move():
+            termination.snapshot()
             termination.circuit = self.circuits[1]
             termination.save()
 
@@ -392,6 +393,7 @@ class CircuitTerminationChangeLoggingTestCase(TestCase):
         ObjectChange.objects.all().delete()
 
         def _flip():
+            termination.snapshot()
             termination.term_side = 'Z'
             termination.save()
 
@@ -438,6 +440,7 @@ class CircuitTerminationChangeLoggingTestCase(TestCase):
         ))
         ObjectChange.objects.all().delete()
 
+        termination.snapshot()
         self._tracked(termination.save)
 
         self.assertFalse(self._circuit_changes(self.circuits[0]).exists())
@@ -451,6 +454,7 @@ class CircuitTerminationChangeLoggingTestCase(TestCase):
         ObjectChange.objects.all().delete()
 
         def _move():
+            termination.snapshot()
             termination.circuit = self.circuits[1]
             termination.save(update_fields=('circuit',))
 

netbox/core/signals.py

@@ -4,10 +4,10 @@ from threading import local
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ObjectDoesNotExist, ValidationError
 from django.core.signals import request_finished
-from django.db import transaction
+from django.db import router, transaction
 from django.db.models import CASCADE, RESTRICT
 from django.db.models.fields.reverse_related import ManyToManyRel, ManyToOneRel
-from django.db.models.signals import m2m_changed, post_migrate, post_save, pre_delete
+from django.db.models.signals import m2m_changed, post_migrate, post_save, pre_delete, pre_save
 from django.dispatch import Signal, receiver
 from django.utils.translation import gettext_lazy as _
 from django.utils.translation import ngettext
@@ -90,6 +90,26 @@ def update_object_types(sender, **kwargs):
 _signals_received = local()
 
 
+@receiver(pre_save)
+def snapshot_unsnapshotted_object(sender, instance, raw=False, using=None, **kwargs):
+    """
+    Generate a pre-change snapshot from the database for an existing object being saved without one.
+    """
+    if raw or not isinstance(instance, ChangeLoggingMixin) or current_request.get() is None:
+        return
+    if instance._state.adding or instance.pk is None or hasattr(instance, '_prechange_snapshot'):
+        return
+
+    db = using or router.db_for_write(sender, instance=instance)
+    if (original := sender._base_manager.using(db).filter(pk=instance.pk).first()) is None:
+        return
+    logger.warning(f"No pre-change snapshot taken for {sender._meta.label} {instance.pk}; reading it from the database")
+    original.snapshot()
+    instance._prechange_snapshot = original._prechange_snapshot
+    # Discard after this save so that a later save of the instance takes a fresh snapshot
+    instance._prechange_snapshot_from_db = True
+
+
 @receiver((post_save, m2m_changed))
 def handle_changed_object(sender, instance, **kwargs):
     """
@@ -135,6 +155,8 @@ def handle_changed_object(sender, instance, **kwargs):
         OBJECT_DELETED: ObjectChangeActionChoices.ACTION_DELETE,
     }[event_type]
     objectchange = instance.to_objectchange(action)
+    if instance.__dict__.pop('_prechange_snapshot_from_db', False):
+        del instance._prechange_snapshot
     object_change_id = None
     # If this is a many-to-many field change, check for a previous ObjectChange instance recorded
     # for this object by this request and update it

netbox/core/tests/test_signals.py

@@ -78,6 +78,7 @@ class HandleChangedObjectSignalTestCase(TestCase):
         request = _build_request(self.user)
 
         with event_tracking(request):
+            site.snapshot()
             site.description = 'updated'
             site.save()
 
@@ -87,6 +88,41 @@ class HandleChangedObjectSignalTestCase(TestCase):
         ).order_by('-pk')
         self.assertEqual(ocs.first().action, ObjectChangeActionChoices.ACTION_UPDATE)
 
+    def test_update_without_snapshot_records_prechange_from_database(self):
+        """An update saved without a snapshot records its pre-change data from the database. Ref: #23617"""
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        request = _build_request(self.user)
+
+        with self.assertLogs('netbox.core.signals', level='WARNING'), event_tracking(request):
+            site.description = 'first'
+            site.save()
+            site.description = 'second'
+            site.save()
+
+        ocs = ObjectChange.objects.filter(
+            changed_object_type=ContentType.objects.get_for_model(Site),
+            changed_object_id=site.pk,
+        ).order_by('pk')
+        self.assertEqual([oc.prechange_data['description'] for oc in ocs], ['', 'first'])
+        self.assertEqual([oc.postchange_data['description'] for oc in ocs], ['first', 'second'])
+        self.assertFalse(hasattr(site, '_prechange_snapshot'))
+
+    def test_update_keeps_existing_snapshot(self):
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        request = _build_request(self.user)
+
+        with self.assertNoLogs('netbox.core.signals', level='WARNING'), event_tracking(request):
+            site.snapshot()
+            site.description = 'first'
+            site.save()
+
+        oc = ObjectChange.objects.get(
+            changed_object_type=ContentType.objects.get_for_model(Site),
+            changed_object_id=site.pk,
+        )
+        self.assertEqual(oc.prechange_data['description'], '')
+        self.assertEqual(oc.postchange_data['description'], 'first')
+
     def test_no_request_skips_objectchange(self):
         # Saving outside a request context (no event_tracking) should not record any
         # ObjectChange entries.

netbox/dcim/models/cables.py

@@ -404,6 +404,9 @@ class Cable(PrimaryModel):
             self.update_terminations()
 
         super().save(*args, force_update=True, **save_kwargs)
+        # Discard the pre-termination snapshot so that a later save of this instance takes a fresh one
+        if _created:
+            self.__dict__.pop('_prechange_snapshot', None)
 
         try:
             trace_paths.send(Cable, instance=self, created=_created, status_written=status_written)

netbox/dcim/tests/test_models.py

@@ -3457,6 +3457,20 @@ class CableDisconnectChangeLoggingTestCase(TestCase):
         self.assertEqual(len(change.postchange_data['a_terminations']), 1)
         self.assertEqual(len(change.postchange_data['b_terminations']), 1)
 
+    @tag('regression')  # Ref: #23617
+    def test_resaving_new_cable_records_current_prechange(self):
+        # The creation snapshot must not be reused, so the re-save falls back to the database
+        with self.assertLogs('netbox.core.signals', level='WARNING'), event_tracking(self._request()):
+            cable = Cable(a_terminations=[self.interface1], b_terminations=[self.interface2])
+            cable.save()
+            cable.label = 'x'
+            cable.save()
+
+        change = self._updates(cable).last()
+        self.assertEqual(len(change.prechange_data['a_terminations']), 1)
+        self.assertEqual(len(change.prechange_data['b_terminations']), 1)
+        self.assertEqual(change.postchange_data['label'], 'x')
+
     @tag('regression')  # Ref: netbox-branching#631
     def test_cable_deletion_records_disconnect(self):
         cable = self._connect(self.interface1, self.interface2)

netbox/extras/tests/test_event_rules.py

@@ -1181,6 +1181,7 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         with event_tracking(request):
             site = Site(name='Site 1', slug='site-1')
             site.save()
+            site.snapshot()
             site.description = 'foo'
             site.save()
         self.assertEqual(self.queue.count, 1, msg="Duplicate jobs found in queue")
@@ -1191,8 +1192,10 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         # Test multiple updates
         site = Site.objects.create(name='Site 2', slug='site-2')
         with event_tracking(request):
+            site.snapshot()
             site.description = 'foo'
             site.save()
+            site.snapshot()
             site.description = 'bar'
             site.save()
         self.assertEqual(self.queue.count, 1, msg="Duplicate jobs found in queue")
@@ -1203,6 +1206,7 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         # Test update & delete
         site = Site.objects.create(name='Site 3', slug='site-3')
         with event_tracking(request):
+            site.snapshot()
             site.description = 'foo'
             site.save()
             site.delete()

netbox/extras/tests/test_signals.py

@@ -286,8 +286,10 @@ class NotifyObjectChangedSignalTestCase(TestCase):
         # Trigger two updates within a single request; only one Notification should remain.
         request = _build_request(user=self.user)
         with event_tracking(request):
+            site.snapshot()
             site.description = 'first'
             site.save()
+            site.snapshot()
             site.description = 'second'
             site.save()
 

netbox/netbox/models/features.py

@@ -702,8 +702,8 @@ class SyncedDataMixin(models.Model):
 
         :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'):
+        # ManagedFile is not change-logged and so has no snapshot(); keep any snapshot already taken
+        if save and hasattr(self, 'snapshot') and not hasattr(self, '_prechange_snapshot'):
             self.snapshot()
         self.sync_data()
         self.data_synced = timezone.now()