RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

simplify

This commit is contained in:
Arthur 2026-09-16 09:45:22 -07:00
parent 5fd7bbe0ae
commit 5cc33a60c9
2 changed files with 29 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
netbox/circuits/models/circuits.py
View File

@ -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:

18
netbox/circuits/tests/test_models.py
View File

@ -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(