Fail tests that record an object update without a prechange snapshot

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

@@ -1,4 +1,5 @@
 import logging
+import sys
 from threading import local
 
 from django.contrib.contenttypes.models import ContentType
@@ -89,6 +90,8 @@ def update_object_types(sender, **kwargs):
 # Used to track received signals per object
 _signals_received = local()
 
+RUNNING_TESTS = 'test' in sys.argv
+
 
 @receiver((post_save, m2m_changed))
 def handle_changed_object(sender, instance, **kwargs):
@@ -135,6 +138,10 @@ def handle_changed_object(sender, instance, **kwargs):
         OBJECT_DELETED: ObjectChangeActionChoices.ACTION_DELETE,
     }[event_type]
     objectchange = instance.to_objectchange(action)
+    # Under test, fail any update recorded without a prechange snapshot
+    if RUNNING_TESTS and event_type == OBJECT_UPDATED and not m2m_changed:
+        from utilities.testing.changelog import assert_prechange_snapshot
+        assert_prechange_snapshot(instance)
     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,14 @@ class HandleChangedObjectSignalTestCase(TestCase):
         ).order_by('-pk')
         self.assertEqual(ocs.first().action, ObjectChangeActionChoices.ACTION_UPDATE)
 
+    def test_update_without_snapshot_fails_under_test(self):
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        request = _build_request(self.user)
+
+        with event_tracking(request), self.assertRaisesMessage(AssertionError, 'without calling snapshot()'):
+            site.description = 'updated'
+            site.save()
+
     def test_no_request_skips_objectchange(self):
         # Saving outside a request context (no event_tracking) should not record any
         # ObjectChange entries.
@@ -600,6 +609,7 @@ class HandleChangedObjectDirectHandlerTestCase(SimpleTestCase):
         instance = SimpleNamespace(
             pk=123,
             _meta=SimpleNamespace(model_name='device'),
+            _prechange_snapshot={'name': 'Device 1'},
             refresh_from_db=MagicMock(),
         )
         instance.to_objectchange = MagicMock(return_value=objectchange)

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/utilities/testing/changelog.py

@@ -0,0 +1,33 @@
+import sys
+
+__all__ = (
+    'PRECHANGE_SNAPSHOT_EXEMPT',
+    'assert_prechange_snapshot',
+)
+
+# Call sites permitted to record an update without first calling snapshot(), as (module, function qualname)
+PRECHANGE_SNAPSHOT_EXEMPT = {
+    # Known issues; remove each entry once fixed
+    ('dcim.models.cables', 'Cable.save'),  # The second save() when creating a Cable records a spurious update
+    ('dcim.signals', 'update_mac_address_interface'),
+    ('netbox.models.features', 'SyncedDataMixin.sync'),
+    ('wireless.signals', 'update_connected_interfaces'),
+}
+
+
+def assert_prechange_snapshot(instance):
+    """
+    Raise AssertionError if an update to `instance` is being recorded without a prechange snapshot, unless a call
+    site in PRECHANGE_SNAPSHOT_EXEMPT is on the stack.
+    """
+    if hasattr(instance, '_prechange_snapshot'):
+        return
+    frame = sys._getframe(1)
+    while frame is not None:
+        if (frame.f_globals.get('__name__'), frame.f_code.co_qualname) in PRECHANGE_SNAPSHOT_EXEMPT:
+            return
+        frame = frame.f_back
+    raise AssertionError(
+        f"{instance._meta.label} {instance.pk} was updated without calling snapshot() first; its change record "
+        f"will have no prechange data."
+    )