diff --git a/netbox/circuits/models/circuits.py b/netbox/circuits/models/circuits.py index b8d7dce2b..67d88ff48 100644 --- a/netbox/circuits/models/circuits.py +++ b/netbox/circuits/models/circuits.py @@ -400,67 +400,48 @@ class CircuitTermination( super().save(*args, **kwargs) - # Clear the old termination reference if circuit or term_side changed + # Clear the old termination reference if circuit or term_side changed. Written via + # snapshot() + save() rather than a queryset update(), which emits no post_save and so + # records nothing in the changelog (#23134). Matching on the pointer's current value + # skips the write unless it actually references this termination. if circuit_changed or term_side_changed: old_termination_name = f'termination_{self._orig_term_side.lower()}' - self._set_circuit_terminations( - self._orig_circuit_id, {old_termination_name: None}, only_if_references=self.pk - ) + circuit = Circuit.objects.filter( + pk=self._orig_circuit_id, **{old_termination_name: self.pk} + ).first() + if circuit is not None: + circuit.snapshot() + setattr(circuit, old_termination_name, None) + circuit.save(update_fields=[old_termination_name, 'last_updated']) # Update the cache if this is a new termination or circuit/term_side changed if is_new or circuit_changed or term_side_changed: # Update the new circuit's termination reference termination_name = f'termination_{self.term_side.lower()}' - self._set_circuit_terminations(self.circuit_id, {termination_name: self.pk}) + # Re-fetched rather than reusing self.circuit, whose pointers may predate a sibling write + circuit = Circuit.objects.get(pk=self.circuit_id) + circuit.snapshot() + setattr(circuit, termination_name, self) + circuit.save(update_fields=[termination_name, 'last_updated']) # Update cached values for subsequent saves self._orig_circuit_id = self.circuit_id self._orig_term_side = self.term_side def delete(self, *args, **kwargs): - # Clear the circuit's reference before the row goes away, so that the change is recorded - # and precedes the DELETE. on_delete=SET_NULL would clear it with an unlogged bulk update. - self._set_circuit_terminations( - self.circuit_id, {'termination_a': None, 'termination_z': None}, only_if_references=self.pk - ) + # on_delete=SET_NULL would clear the circuit's reference with an unlogged bulk update. + # Clearing it here instead also puts the record ahead of the DELETE. + termination_name = f'termination_{self.term_side.lower()}' + circuit = Circuit.objects.filter(pk=self.circuit_id, **{termination_name: self.pk}).first() + if circuit is not None: + circuit.snapshot() + setattr(circuit, termination_name, None) + circuit.save(update_fields=[termination_name, 'last_updated']) return super().delete(*args, **kwargs) delete.alters_data = True - @staticmethod - def _set_circuit_terminations(circuit_id, fields, only_if_references=None): - """ - Set or clear a Circuit's cached `termination_a`/`termination_z` fields, recording the - change. `fields` maps field name to CircuitTermination PK (or None). A queryset update() - emits no post_save, and so records nothing in the changelog (#23134). - - Args: - circuit_id: PK of the Circuit to update - fields: Mapping of field name to the value to assign - only_if_references: If set, restricts the write to fields which currently hold this PK - """ - # Re-fetched rather than reusing a cached circuit, whose pointers may predate a sibling write - circuit = Circuit.objects.filter(pk=circuit_id).first() - if circuit is None: - return - - updates = {} - for field_name, value in fields.items(): - current = getattr(circuit, f'{field_name}_id') - if current == value: - continue - if only_if_references is not None and current != only_if_references: - continue - updates[field_name] = value - if not updates: - return - - circuit.snapshot() - for field_name, value in updates.items(): - setattr(circuit, f'{field_name}_id', value) - circuit.save(update_fields=[*updates, 'last_updated']) - def cache_related_objects(self): self._provider_network = self._region = self._site_group = self._site = self._location = None if self.termination_type: diff --git a/netbox/circuits/tests/test_models.py b/netbox/circuits/tests/test_models.py index 0c3c27193..a33bd588f 100644 --- a/netbox/circuits/tests/test_models.py +++ b/netbox/circuits/tests/test_models.py @@ -549,10 +549,10 @@ class CircuitTerminationChangeLoggingTestCase(TestCase): self.assertIsNone(self.circuits[0].termination_a_id) self.assertFalse(self._circuit_changes(self.circuits[0]).exists()) - @tag('regression') # Ref: #23134 - def test_deletion_clears_the_pointer_which_references_it(self): - # An in-memory term_side which diverges from the persisted one must clear this - # termination's own pointer, and leave the one belonging to its sibling alone + def test_deletion_leaves_pointer_for_another_termination(self): + # An in-memory term_side which diverges from the persisted one must not clear a pointer + # belonging to a different termination. The termination's own pointer is then left to + # on_delete=SET_NULL, and goes unrecorded. termination_a = self._tracked(lambda: CircuitTermination.objects.create( circuit=self.circuits[0], term_side='A', termination=self.sites[0], )) @@ -560,21 +560,13 @@ class CircuitTerminationChangeLoggingTestCase(TestCase): circuit=self.circuits[0], term_side='Z', termination=self.sites[1], )) ObjectChange.objects.all().delete() - termination_z_pk = termination_z.pk termination_z.term_side = 'A' self._tracked(termination_z.delete) self.circuits[0].refresh_from_db() self.assertEqual(self.circuits[0].termination_a_id, termination_a.pk) - self.assertIsNone(self.circuits[0].termination_z_id) - - changes = self._circuit_changes(self.circuits[0]) - self.assertEqual(changes.count(), 1) - self.assertEqual(changes[0].prechange_data['termination_a'], termination_a.pk) - self.assertEqual(changes[0].postchange_data['termination_a'], termination_a.pk) - self.assertEqual(changes[0].prechange_data['termination_z'], termination_z_pk) - self.assertIsNone(changes[0].postchange_data['termination_z']) + self.assertFalse(self._circuit_changes(self.circuits[0]).exists()) def test_circuit_deletion_records_no_pointer_update(self): self._tracked(lambda: CircuitTermination.objects.create(