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

netbox/circuits/tests/test_models.py

@@ -415,6 +415,7 @@ class CircuitTerminationChangeLoggingTestCase(TestCase):
         ObjectChange.objects.all().delete()
 
         def _move():
+            termination.snapshot()
             termination.circuit = self.circuits[1]
             termination.save()
 
@@ -440,6 +441,7 @@ class CircuitTerminationChangeLoggingTestCase(TestCase):
         ObjectChange.objects.all().delete()
 
         def _flip():
+            termination.snapshot()
             termination.term_side = 'Z'
             termination.save()
 
@@ -486,6 +488,7 @@ class CircuitTerminationChangeLoggingTestCase(TestCase):
         ))
         ObjectChange.objects.all().delete()
 
+        termination.snapshot()
         self._tracked(termination.save)
 
         self.assertFalse(self._circuit_changes(self.circuits[0]).exists())
@@ -499,6 +502,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

@@ -7,7 +7,7 @@ from django.core.signals import request_finished
 from django.db import 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,19 @@ 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
+
+    instance.snapshot_from_db(using=using)
+
+
 @receiver((post_save, m2m_changed))
 def handle_changed_object(sender, instance, **kwargs):
     """
@@ -163,6 +176,10 @@ def handle_changed_object(sender, instance, **kwargs):
     enqueue_event(queue, instance, request, event_type, object_change_id=object_change_id)
     events_queue.set(queue)
 
+    # Discard a snapshot read from the database only once the event has captured it
+    if instance.__dict__.pop('_prechange_snapshot_from_db', False):
+        instance.__dict__.pop('_prechange_snapshot', None)
+
     # Increment metric counters
     if event_type == OBJECT_CREATED:
         model_inserts.labels(instance._meta.model_name).inc()

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,55 @@ 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.models.features', 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_without_snapshot_enqueues_prechange_from_database(self):
+        """The event for an update saved without a snapshot carries its pre-change data. Ref: #23617"""
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        request = _build_request(self.user)
+
+        with self.assertLogs('netbox.models.features', level='WARNING'), event_tracking(request):
+            site.description = 'updated'
+            site.save()
+            snapshots = events_queue.get()[f'dcim.site:{site.pk}']['snapshots']
+
+        self.assertEqual(snapshots['prechange']['description'], '')
+        self.assertEqual(snapshots['postchange']['description'], 'updated')
+        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.models.features', 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

@@ -21,6 +21,7 @@ from dcim.exceptions import UnsupportedCablePath
 from dcim.fields import PathField
 from dcim.utils import decompile_path_node, object_to_path_node, rebuild_cable_paths
 from netbox.choices import ColorChoices
+from netbox.context import current_request
 from netbox.models import ChangeLoggedModel, PrimaryModel
 from utilities.conversion import to_meters
 from utilities.data import normalize_update_fields
@@ -353,6 +354,8 @@ class Cable(PrimaryModel):
 
     def save(self, *args, force_insert=False, force_update=False, using=None, update_fields=None):
         _created = self.pk is None
+        # Unlike _created, also true for a new Cable with a preset pk (e.g. one restored by deserialize_object())
+        adding = self._state.adding
         save_kwargs = {
             'using': using,
             'update_fields': update_fields,
@@ -391,17 +394,24 @@ class Cable(PrimaryModel):
         profile_written = update_fields is None or 'profile' in update_fields
 
         # If this is a new Cable, save it before attempting to create its CableTerminations
-        if self._state.adding:
+        if adding:
             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)
-        elif self._terminations_modified:
-            self.update_terminations()
+        force_terminations = profile_written and self._orig_profile != self.profile
+        if force_terminations or self._terminations_modified:
+            # update_terminations() writes before the pre_save fallback runs, so snapshot the stored Cable now
+            if current_request.get() is not None and not hasattr(self, '_prechange_snapshot'):
+                self.snapshot_from_db(using=using)
+            self.update_terminations(force=force_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 adding:
+            self.__dict__.pop('_prechange_snapshot', None)
 
         try:
             trace_paths.send(Cable, instance=self, created=_created, status_written=status_written)

netbox/dcim/models/devices.py

@@ -1175,6 +1175,7 @@ class Device(
         # 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

@@ -501,5 +501,10 @@ 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:
+        # Keep any snapshot the caller took before making unsaved changes to the MAC address
+        if not hasattr(instance.primary_mac_address, '_prechange_snapshot'):
+            instance.primary_mac_address.snapshot()
         instance.primary_mac_address.assigned_object = instance
         instance.primary_mac_address.save()
+        # Discard the snapshot so that a later save of the MAC address takes a fresh one
+        instance.primary_mac_address.__dict__.pop('_prechange_snapshot', None)

netbox/dcim/tests/test_models.py

@@ -56,6 +56,49 @@ class MACAddressTestCase(TestCase):
         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)
+        self.assertFalse(hasattr(mac, '_prechange_snapshot'))
+
+    @tag('regression')  # Ref: #23617
+    def test_primary_mac_assignment_keeps_existing_snapshot(self):
+        mac = MACAddress.objects.create(mac_address='1234567890ce')
+        request = RequestFactory().get('/')
+        request.id = uuid.uuid4()
+        request.user = User.objects.create_user(username='testuser')
+
+        with event_tracking(request):
+            mac.snapshot()
+            mac.description = 'updated'
+            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.assertEqual(change.prechange_data['description'], '')
+        self.assertEqual(change.postchange_data['description'], 'updated')
+
     @tag('regression')
     def test_clean_will_not_allow_removal_of_assigned_object_if_primary(self):
         self.mac_a.assigned_object = None
@@ -1327,6 +1370,29 @@ class DeviceBayTestCase(TestCase):
         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
@@ -3440,6 +3506,65 @@ class CableDisconnectChangeLoggingTestCase(TestCase):
             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: #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.models.features', 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: #23617
+    def test_resaving_new_cable_with_preset_pk_records_current_prechange(self):
+        # A new Cable with a preset pk (e.g. restored by deserialize_object()) must also discard its creation snapshot
+        with self.assertLogs('netbox.models.features', level='WARNING'), event_tracking(self._request()):
+            cable = Cable(pk=1000000, 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: #23617
+    def test_termination_update_without_snapshot_records_prechange(self):
+        cable = self._connect(self.interface1, self.interface2)
+        old_b_termination = CableTermination.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_B)
+
+        request = self._request()
+        with self.assertLogs('netbox.models.features', level='WARNING') as logs, event_tracking(request):
+            cable = Cable.objects.get(pk=cable.pk)
+            cable.label = 'x'
+            cable.b_terminations = [self.interface3]
+            cable.save()
+
+        change = self._updates(cable).get(request_id=request.id)
+        self.assertEqual(change.prechange_data['b_terminations'], [old_b_termination.pk])
+        self.assertEqual(change.prechange_data['label'], '')
+        self.assertNotEqual(change.postchange_data['b_terminations'], [old_b_termination.pk])
+        self.assertEqual(change.postchange_data['label'], 'x')
+        self.assertFalse(hasattr(cable, '_prechange_snapshot'))
+        self.assertEqual(len(logs.records), 1)
+
     @tag('regression')  # Ref: netbox-branching#631
     def test_cable_deletion_records_disconnect(self):
         cable = self._connect(self.interface1, self.interface2)
@@ -3517,6 +3642,7 @@ class CableDisconnectChangeLoggingTestCase(TestCase):
         request = self._request()
         with event_tracking(request):
             cable = Cable.objects.get(pk=cable_pk)
+            cable.snapshot()
             cable.b_terminations = [self.interface3]
             cable.save()
 

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

@@ -1,4 +1,5 @@
 import json
+import logging
 from collections import defaultdict
 from functools import cached_property
 
@@ -50,6 +51,8 @@ __all__ = (
     'register_models',
 )
 
+logger = logging.getLogger('netbox.models.features')
+
 
 #
 # Feature mixins
@@ -102,6 +105,21 @@ class ChangeLoggingMixin(DeleteMixin, models.Model):
         self._prechange_snapshot = self.serialize_object(exclude=exclude_fields)
     snapshot.alters_data = True
 
+    def snapshot_from_db(self, using=None):
+        """
+        Snapshot the stored copy of an object being saved without a snapshot. The snapshot is discarded once the
+        save has been logged.
+        """
+        model = self.__class__
+        db = using or router.db_for_write(model, instance=self)
+        if (original := model._base_manager.using(db).filter(pk=self.pk).first()) is None:
+            return
+        logger.warning(f"No pre-change snapshot taken for {model._meta.label} {self.pk}; reading it from the database")
+        original.snapshot()
+        self._prechange_snapshot = original._prechange_snapshot
+        self._prechange_snapshot_from_db = True
+    snapshot_from_db.alters_data = True
+
     def to_objectchange(self, action):
         """
         Return a new ObjectChange representing a change made to this object. This will typically be called automatically
@@ -702,10 +720,15 @@ 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(); 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()
         if save:
             self.save()
+            # Discard the snapshot so that a later save of this object takes a fresh one
+            self.__dict__.pop('_prechange_snapshot', None)
     sync.alters_data = True
 
     def sync_data(self):

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 extras.models import (
 )
 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,69 @@ class ModelFeaturesTestCase(TestCase):
                 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')
+
+        template = ConfigTemplate.objects.get(pk=obj.pk)
+        with event_tracking(request):
+            template.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')
+        self.assertFalse(hasattr(template, '_prechange_snapshot'))
+
+    def test_synceddatamixin_sync_keeps_existing_snapshot(self):
+        """Saving a sync records a snapshot taken by the caller, then discards it. 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')
+
+        template = ConfigTemplate.objects.get(pk=obj.pk)
+        with self.assertNoLogs('netbox.models.features', level='WARNING'), event_tracking(request):
+            template.snapshot()
+            template.description = 'updated'
+            # Without save, sync() leaves the caller's snapshot for its own save
+            template.sync()
+            self.assertTrue(hasattr(template, '_prechange_snapshot'))
+            template.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['description'], '')
+        self.assertEqual(change.postchange_data['description'], 'updated')
+        self.assertFalse(hasattr(template, '_prechange_snapshot'))
+
+    @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,12 +27,20 @@ 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}")
+        # Keep any snapshot the caller took before making unsaved changes to the interface
+        if not hasattr(instance.interface_a, '_prechange_snapshot'):
+            instance.interface_a.snapshot()
         instance.interface_a.wireless_link = instance
         instance.interface_a.save()
+        # Discard the snapshot so that a later save of the interface takes a fresh one
+        instance.interface_a.__dict__.pop('_prechange_snapshot', None)
     if instance.interface_b.wireless_link != instance:
         logger.debug(f"Updating interface B for wireless link {instance}")
+        if not hasattr(instance.interface_b, '_prechange_snapshot'):
+            instance.interface_b.snapshot()
         instance.interface_b.wireless_link = instance
         instance.interface_b.save()
+        instance.interface_b.__dict__.pop('_prechange_snapshot', None)
 
     # Create/update cable paths
     if created:

netbox/wireless/tests/test_models.py

@@ -66,6 +66,54 @@ class WirelessLinkTestCase(TestCase):
         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)
+            self.assertFalse(hasattr(interface, '_prechange_snapshot'))
+
+    # Regression test for #23617
+    def test_creating_wireless_link_keeps_existing_interface_snapshot(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):
+            interfaces[0].snapshot()
+            interfaces[0].description = 'updated'
+            WirelessLink.objects.create(interface_a=interfaces[0], interface_b=interfaces[1], ssid='LINK1')
+
+        change = ObjectChange.objects.get(
+            changed_object_type=ContentType.objects.get_for_model(Interface),
+            changed_object_id=interfaces[0].pk,
+        )
+        self.assertEqual(change.prechange_data['description'], '')
+        self.assertEqual(change.postchange_data['description'], 'updated')
+
     def test_resaving_wireless_link_does_not_log_connected_interfaces(self):
         interface_a = Interface.objects.create(
             device=self.device,