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

Revert "Fixes #23097: Prevent duplicate Cable Paths when Cable Terminations …" (#23149) 

This reverts commit 1745a7d9aa.
This commit is contained in:
Martin Hauser 2026-09-08 17:25:35 +02:00 committed by GitHub
parent 6895fb76c0
commit 5685c5218e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 7 additions and 271 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
netbox/dcim/forms/connections.py
View File

@ -142,10 +142,8 @@ def get_cable_form(a_type, b_type):
def clean(self):
super().clean()
# The field discards submission order, so a saved cable's end is assigned only when its members changed
for field_name in ('a_terminations', 'b_terminations'):
value = self.cleaned_data.get(field_name, [])
if not self.instance.pk or set(value) != set(self.initial.get(field_name, [])):
setattr(self.instance, field_name, value)
# Set the A/B terminations on the Cable instance
self.instance.a_terminations = self.cleaned_data.get('a_terminations', [])
self.instance.b_terminations = self.cleaned_data.get('b_terminations', [])
return _CableForm

23
netbox/dcim/models/cables.py
View File

@ -229,16 +229,6 @@ class Cable(PrimaryModel):
ct.termination for ct in self.terminations.all() if ct.cable_end == side
]
def _cache_stored_terminations(self):
"""
Fill each cold termination cache from the CableTermination rows, in their stored order.
"""
a_terminations, b_terminations = self.get_terminations()
if not hasattr(self, '_a_terminations'):
self._a_terminations = list(a_terminations.keys())
if not hasattr(self, '_b_terminations'):
self._b_terminations = list(b_terminations.keys())
def _set_x_terminations(self, side, value):
"""
Set the terminating objects for the given cable end (A or B).
@ -254,11 +244,7 @@ class Cable(PrimaryModel):
ct.termination for ct in CableTermination.objects.filter(pk__in=value).prefetch_related('termination')
]
# Compare a saved cable against its stored rows, not against a possibly stale prefetch of self.terminations
if self.pk and not hasattr(self, _attr):
self._cache_stored_terminations()
if not self.pk or getattr(self, _attr) != list(value):
if not self.pk or getattr(self, _attr, []) != list(value):
self._terminations_modified = True
setattr(self, _attr, value)
@ -524,10 +510,6 @@ class Cable(PrimaryModel):
force_a = force or self._connectors_reassigned(a_terminations, self.a_terminations)
force_b = force or self._connectors_reassigned(b_terminations, self.b_terminations)
# Recreating either end's terminations invalidates its paths, even when the endpoints are unchanged
if force_a or force_b:
self._terminations_modified = True
# When force-recreating terminations (e.g. after a profile change), cache the termination objects
# from the database before deleting, so they are available for recreation. Without this, the
# a_terminations/b_terminations properties would query the DB after deletion and return empty lists.
@ -536,6 +518,9 @@ class Cable(PrimaryModel):
if force_b and not hasattr(self, '_b_terminations'):
self._b_terminations = list(b_terminations.keys())
# Recreating terminations invalidates existing paths, even when the endpoints are unchanged
self._terminations_modified = True
# Delete any stale CableTerminations
for termination, ct in a_terminations.items():
if force_a or (termination.pk and termination not in self.a_terminations):

46
netbox/dcim/tests/test_cablepaths.py
View File

@ -2892,52 +2892,6 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
interface3.refresh_from_db()
self.assertPathIsNotSet(interface3)
def test_304_resave_cable_with_unchanged_terminations(self):
"""
[IF1] --C1-- [IF2]
"""
interface1 = Interface.objects.create(device=self.device, name='Interface 1')
interface2 = Interface.objects.create(device=self.device, name='Interface 2')
cable1 = Cable(
a_terminations=[interface1],
b_terminations=[interface2]
)
cable1.save()
path_pks = set(CablePath.objects.values_list('pk', flat=True))
termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True))
self.assertEqual(len(path_pks), 2)
self.assertEqual(len(termination_pks), 2)
# Reassign the same terminations on a freshly loaded instance
cable1 = Cable.objects.get(pk=cable1.pk)
cable1.a_terminations = [interface1]
cable1.b_terminations = [interface2]
cable1.label = 'Renamed'
cable1.save()
self.assertEqual(set(CablePath.objects.values_list('pk', flat=True)), path_pks)
self.assertEqual(
set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
termination_pks
)
path1 = self.assertPathExists(
(interface1, cable1, interface2),
is_complete=True,
is_active=True
)
path2 = self.assertPathExists(
(interface2, cable1, interface1),
is_complete=True,
is_active=True
)
interface1.refresh_from_db()
interface2.refresh_from_db()
self.assertPathIsSet(interface1, path1)
self.assertPathIsSet(interface2, path2)
def test_401_exclude_midspan_devices(self):
"""
[IF1] --C1-- [FP1][Test Device][RP1] --C2-- [RP2][Test Device][FP2] --C3-- [IF2]

46
netbox/dcim/tests/test_cablepaths2.py
View File

@ -2785,49 +2785,3 @@ class CablePathTestCase(BaseCablePathTestCase):
set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
termination_pks
)
def test_311_change_cable_profile_after_reassigning_unchanged_terminations(self):
"""
[IF1] --C1-- [IF2]
Applying a profile after both termination caches have been populated must still rebuild the paths.
"""
interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'),
Interface.objects.create(device=self.device, name='Interface 2'),
]
# Create cable 1 without a profile
cable1 = Cable(
a_terminations=[interfaces[0]],
b_terminations=[interfaces[1]],
)
cable1.clean()
cable1.save()
self.assertEqual(CablePath.objects.count(), 2)
# Reload and populate both termination caches by reassigning their stored values
cable1 = Cable.objects.get(pk=cable1.pk)
cable1.a_terminations = [interfaces[0]]
cable1.b_terminations = [interfaces[1]]
self.assertFalse(cable1._terminations_modified)
cable1.profile = CableProfileChoices.SINGLE_1C1P
cable1.full_clean()
cable1.save()
path1 = self.assertPathExists(
(interfaces[0], cable1, interfaces[1]),
is_complete=True,
is_active=True
)
path2 = self.assertPathExists(
(interfaces[1], cable1, interfaces[0]),
is_complete=True,
is_active=True
)
self.assertEqual(CablePath.objects.count(), 2)
interfaces[0].refresh_from_db()
interfaces[1].refresh_from_db()
self.assertPathIsSet(interfaces[0], path1)
self.assertPathIsSet(interfaces[1], path2)

44
netbox/dcim/tests/test_models.py
View File

@ -2413,50 +2413,6 @@ class CableTestCase(TestCase):
with self.assertRaises(ValidationError):
cable.clean()
def test_reassigning_unchanged_terminations_does_not_flag_a_change(self):
"""
Assigning the stored terminations to a freshly loaded cable must leave them unflagged.
"""
interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0')
# A cable loaded from the database has no cached terminations
cable = Cable.objects.first()
cable.a_terminations = [interface1]
cable.b_terminations = [interface2]
self.assertFalse(cable._terminations_modified)
def test_reassigning_different_terminations_flags_a_change(self):
"""
Assigning a different termination to a freshly loaded cable must flag the change.
"""
interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
interface3 = Interface.objects.get(device__name='TestDevice2', name='eth1')
cable = Cable.objects.first()
cable.a_terminations = [interface1]
cable.b_terminations = [interface3]
self.assertTrue(cable._terminations_modified)
def test_reassigning_stale_prefetched_terminations_flags_a_change(self):
"""
A stale prefetched relation must not hide a real termination change.
"""
cable = Cable.objects.prefetch_related('terminations__termination').first()
stale_termination = cable.b_terminations[0]
current_termination = Interface.objects.get(device__name='TestDevice2', name='eth1')
# Moving the B end through a second instance leaves the prefetch above stale
moved = Cable.objects.get(pk=cable.pk)
moved.b_terminations = [current_termination]
moved.save()
# The value matches the stale prefetch but not the stored row
cable.b_terminations = [stale_termination]
self.assertTrue(cable._terminations_modified)
def test_partial_save_does_not_apply_an_unwritten_profile(self):
"""
A save excluding profile must leave the terminations alone but keep the change pending.

111
netbox/dcim/tests/test_views.py
View File

@ -5271,117 +5271,6 @@ class CableTestCase(
[(1, interfaces[1]), (2, interfaces[0])]
)
@tag('regression') # Issue #23097
def test_edit_with_unchanged_terminations_preserves_paths(self):
"""Editing a cable without changing its terminations must leave its paths in place."""
# The form's termination fields are restricted by view permission
self.add_permissions('dcim.change_cable', 'dcim.view_interface')
interface_a = Interface.objects.get(
device__name='Device 1', device__site__name='Site 1', name='Interface 1'
)
cable = interface_a.cable
interface_b = cable.b_terminations[0]
path_pks = set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True))
self.assertEqual(len(path_pks), 2)
data = {
'a_terminations': [interface_a.pk],
'b_terminations': [interface_b.pk],
'type': CableTypeChoices.TYPE_CAT6,
'status': LinkStatusChoices.STATUS_CONNECTED,
'label': 'Renamed',
'color': 'c0c0c0',
}
request = {
'path': self._get_url('edit', cable),
'data': post_data(data),
}
self.assertHttpStatus(self.client.post(**request), 302)
cable.refresh_from_db()
self.assertEqual(cable.label, 'Renamed')
self.assertEqual(
set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)),
path_pks
)
@tag('regression') # Issue #23097
def test_edit_with_unchanged_terminations_preserves_connector_order(self):
"""A label-only edit must keep the connectors of an end whose stored order differs from the form's."""
# The form's termination fields are restricted by view permission
self.add_permissions('dcim.change_cable', 'dcim.view_interface')
interface_a = Interface.objects.get(device__name='Device 3', name='Interface 1')
interfaces = list(Interface.objects.filter(device__name='Device 4').order_by('name')[:2])
cable = Cable(
a_terminations=[interface_a],
b_terminations=[interfaces[1], interfaces[0]],
profile=CableProfileChoices.BREAKOUT_1C2P_2C1P,
)
cable.save()
def b_terminations():
return list(
CableTermination.objects.filter(cable=cable, cable_end=CableEndChoices.SIDE_B)
.values_list('pk', 'connector', 'termination_id')
)
terminations = b_terminations()
self.assertEqual([t[1:] for t in terminations], [(1, interfaces[1].pk), (2, interfaces[0].pk)])
path_pks = set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True))
data = {
'a_terminations': [interface_a.pk],
'b_terminations': [interfaces[0].pk, interfaces[1].pk],
'profile': CableProfileChoices.BREAKOUT_1C2P_2C1P,
'status': LinkStatusChoices.STATUS_CONNECTED,
'label': 'Renamed',
}
request = {
'path': self._get_url('edit', cable),
'data': post_data(data),
}
self.assertHttpStatus(self.client.post(**request), 302)
cable.refresh_from_db()
self.assertEqual(cable.label, 'Renamed')
self.assertEqual(b_terminations(), terminations)
self.assertEqual(
set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)),
path_pks
)
def test_edit_with_changed_terminations_rewires_the_end(self):
"""Replacing a termination through the edit form must still rewrite that end."""
# The form's termination fields are restricted by view permission
self.add_permissions('dcim.change_cable', 'dcim.view_interface')
interface_a = Interface.objects.get(
device__name='Device 1', device__site__name='Site 1', name='Interface 1'
)
cable = interface_a.cable
interface_b = cable.b_terminations[0]
new_interface_b = Interface.objects.get(device__name='Device 4', name='Interface 3')
data = {
'a_terminations': [interface_a.pk],
'b_terminations': [new_interface_b.pk],
'type': CableTypeChoices.TYPE_CAT6,
'status': LinkStatusChoices.STATUS_CONNECTED,
}
request = {
'path': self._get_url('edit', cable),
'data': post_data(data),
}
self.assertHttpStatus(self.client.post(**request), 302)
self.assertEqual(Cable.objects.get(pk=cable.pk).b_terminations, [new_interface_b])
interface_b.refresh_from_db()
self.assertIsNone(interface_b.cable)
new_interface_b.refresh_from_db()
self.assertEqual(new_interface_b.cable, cable)
#
# Connections