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

Fixes #23072: Rebuild cable paths when applying or changing a cable profile (#23091) 

Set `_terminations_modified` flag when recreating terminations to ensure
paths are rebuilt even when endpoints remain unchanged.
Reset `_orig_status`, `_orig_profile`, and `_terminations_modified`
after saving a cable to prevent repeated saves from recreating
terminations and paths.
Add comprehensive test coverage for profile changes, trunk regrouping,
and mid-span cables.
This commit is contained in:
Martin Hauser 2026-09-01 17:46:17 +02:00 committed by GitHub
parent a39d5626fe
commit 9aa0c5c605
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 361 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -23,6 +23,7 @@ from dcim.utils import decompile_path_node, object_to_path_node
from netbox.choices import ColorChoices
from netbox.models import ChangeLoggedModel, PrimaryModel
from utilities.conversion import to_meters
from utilities.data import normalize_update_fields
from utilities.exceptions import AbortRequest
from utilities.fields import ColorField, GenericArrayForeignKey
from utilities.querysets import RestrictedQuerySet
@ -321,6 +322,11 @@ class Cable(PrimaryModel):
def save(self, *args, force_insert=False, force_update=False, using=None, update_fields=None):
_created = self.pk is None
save_kwargs = {
'using': using,
'update_fields': update_fields,
}
update_fields = normalize_update_fields(save_kwargs)
# Store the given length (if any) in meters for use in database ordering
if self.length is not None and self.length_unit:
@ -332,24 +338,35 @@ class Cable(PrimaryModel):
if self.length is None:
self.length_unit = None
# A field counts as changed only when this save actually writes it
status_written = update_fields is None or 'status' in update_fields
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:
super().save(*args, force_insert=True, using=using, update_fields=update_fields)
super().save(*args, force_insert=True, **save_kwargs)
# Update the private PK used in __str__()
self._pk = self.pk
if self._orig_profile != self.profile:
if profile_written and self._orig_profile != self.profile:
self.update_terminations(force=True)
elif self._terminations_modified:
self.update_terminations()
super().save(*args, force_update=True, using=using, update_fields=update_fields)
super().save(*args, force_update=True, **save_kwargs)
try:
trace_paths.send(Cable, instance=self, created=_created)
except UnsupportedCablePath as e:
raise AbortRequest(e)
# Reset change tracking for the next save of this instance
if status_written:
self._orig_status = self.status
if profile_written:
self._orig_profile = self.profile
self._terminations_modified = False
def delete(self, *args, **kwargs):
# Track this Cable as being deleted so the post_delete signal handler
# for cascaded CableTerminations can skip redundant path retracing;
@ -477,6 +494,9 @@ class Cable(PrimaryModel):
if 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 or (termination.pk and termination not in self.a_terminations):

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

@ -15,7 +15,7 @@ class CablePathTestCase(BaseCablePathTestCase):
Tests are numbered as follows:
1XX: Test direct connections using each profile
2XX: Topology tests replicated from the legacy test case and adapted to use profiles
3XX: Dynamic port mapping and termination changes
3XX: Dynamic port mapping, profile and termination changes
"""
def test_101_cable_profile_single_1c1p(self):
@ -2512,3 +2512,276 @@ class CablePathTestCase(BaseCablePathTestCase):
is_complete=True,
is_active=True
)
def test_307_change_cable_profile_rebuilds_paths(self):
"""
[IF1] --C1-- [IF2]
Applying a profile to an existing cable rebuilds its 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.assertPathExists(
(interfaces[0], cable1, interfaces[1]),
is_complete=True,
is_active=True
)
self.assertPathExists(
(interfaces[1], cable1, interfaces[0]),
is_complete=True,
is_active=True
)
# Reload so _terminations_modified starts False, as it does in a bulk edit
cable1 = Cable.objects.get(pk=cable1.pk)
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)
def test_308_change_cable_profile_regroups_trunk_paths(self):
"""
[IF1] --C1-- [IF3]
[IF2] [IF4]
Applying a trunk profile to an existing cable regroups its paths by connector, and
clearing it again collapses them.
"""
interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'),
Interface.objects.create(device=self.device, name='Interface 2'),
Interface.objects.create(device=self.device, name='Interface 3'),
Interface.objects.create(device=self.device, name='Interface 4'),
]
# Create cable 1 without a profile
cable1 = Cable(
a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[interfaces[2], interfaces[3]],
)
cable1.clean()
cable1.save()
# Without a profile both terminations on each end share a single path
self.assertPathExists(
((interfaces[0], interfaces[1]), cable1, (interfaces[2], interfaces[3])),
is_complete=True,
is_active=True
)
self.assertPathExists(
((interfaces[2], interfaces[3]), cable1, (interfaces[0], interfaces[1])),
is_complete=True,
is_active=True
)
self.assertEqual(CablePath.objects.count(), 2)
# Reload so _terminations_modified starts False, as it does in a bulk edit
cable1 = Cable.objects.get(pk=cable1.pk)
cable1.profile = CableProfileChoices.TRUNK_2C1P
cable1.full_clean()
cable1.save()
path1 = self.assertPathExists(
(interfaces[0], cable1, interfaces[2]),
is_complete=True,
is_active=True
)
path2 = self.assertPathExists(
(interfaces[1], cable1, interfaces[3]),
is_complete=True,
is_active=True
)
path3 = self.assertPathExists(
(interfaces[2], cable1, interfaces[0]),
is_complete=True,
is_active=True
)
path4 = self.assertPathExists(
(interfaces[3], cable1, interfaces[1]),
is_complete=True,
is_active=True
)
self.assertEqual(CablePath.objects.count(), 4)
for interface in interfaces:
interface.refresh_from_db()
self.assertPathIsSet(interfaces[0], path1)
self.assertPathIsSet(interfaces[1], path2)
self.assertPathIsSet(interfaces[2], path3)
self.assertPathIsSet(interfaces[3], path4)
self.assertEqual(interfaces[0].cable_connector, 1)
self.assertEqual(interfaces[1].cable_connector, 2)
self.assertEqual(interfaces[2].cable_connector, 1)
self.assertEqual(interfaces[3].cable_connector, 2)
for interface in interfaces:
self.assertEqual(interface.cable_positions, [1])
# Clearing the profile is a bulk-edit action in its own right
cable1 = Cable.objects.get(pk=cable1.pk)
cable1.profile = ''
cable1.full_clean()
cable1.save()
path5 = self.assertPathExists(
((interfaces[0], interfaces[1]), cable1, (interfaces[2], interfaces[3])),
is_complete=True,
is_active=True
)
path6 = self.assertPathExists(
((interfaces[2], interfaces[3]), cable1, (interfaces[0], interfaces[1])),
is_complete=True,
is_active=True
)
self.assertEqual(CablePath.objects.count(), 2)
for interface in interfaces:
interface.refresh_from_db()
self.assertIsNone(interface.cable_connector)
self.assertPathIsSet(interfaces[0], path5)
self.assertPathIsSet(interfaces[1], path5)
self.assertPathIsSet(interfaces[2], path6)
self.assertPathIsSet(interfaces[3], path6)
def test_309_change_midspan_cable_profile_rebuilds_paths(self):
"""
[IF1] --C1-- [FP1][RP1] --C3-- [RP2][FP2] --C2-- [IF2]
Applying a profile to a cable which terminates on pass-through ports rebuilds the
paths traversing it. The rear ports are not path origins, so a missing rebuild
truncates those paths rather than deleting them.
"""
interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'),
Interface.objects.create(device=self.device, name='Interface 2'),
]
rear_ports = [
RearPort.objects.create(device=self.device, name='Rear Port 1'),
RearPort.objects.create(device=self.device, name='Rear Port 2'),
]
front_ports = [
FrontPort.objects.create(device=self.device, name='Front Port 1'),
FrontPort.objects.create(device=self.device, name='Front Port 2'),
]
for front_port, rear_port in zip(front_ports, rear_ports):
PortMapping.objects.create(
device=self.device,
front_port=front_port,
front_port_position=1,
rear_port=rear_port,
rear_port_position=1
)
cable1 = Cable(a_terminations=[interfaces[0]], b_terminations=[front_ports[0]])
cable1.clean()
cable1.save()
cable2 = Cable(a_terminations=[front_ports[1]], b_terminations=[interfaces[1]])
cable2.clean()
cable2.save()
# Create the mid-span cable without a profile
cable3 = Cable(a_terminations=[rear_ports[0]], b_terminations=[rear_ports[1]])
cable3.clean()
cable3.save()
nodes_a_to_b = (
interfaces[0], cable1, front_ports[0], rear_ports[0], cable3, rear_ports[1], front_ports[1], cable2,
interfaces[1],
)
nodes_b_to_a = (
interfaces[1], cable2, front_ports[1], rear_ports[1], cable3, rear_ports[0], front_ports[0], cable1,
interfaces[0],
)
self.assertPathExists(nodes_a_to_b, is_complete=True, is_active=True)
self.assertPathExists(nodes_b_to_a, is_complete=True, is_active=True)
self.assertEqual(CablePath.objects.count(), 2)
# Reload so _terminations_modified starts False, as it does in a bulk edit
cable3 = Cable.objects.get(pk=cable3.pk)
cable3.profile = CableProfileChoices.SINGLE_1C1P
cable3.full_clean()
cable3.save()
path1 = self.assertPathExists(nodes_a_to_b, is_complete=True, is_active=True)
path2 = self.assertPathExists(nodes_b_to_a, is_complete=True, is_active=True)
self.assertEqual(CablePath.objects.count(), 2)
for interface in interfaces:
interface.refresh_from_db()
self.assertPathIsSet(interfaces[0], path1)
self.assertPathIsSet(interfaces[1], path2)
def test_310_repeat_save_does_not_recreate_paths(self):
"""
[IF1] --C1-- [IF2]
Saving an unchanged cable again leaves its terminations and paths untouched.
"""
interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'),
Interface.objects.create(device=self.device, name='Interface 2'),
]
cable1 = Cable(
a_terminations=[interfaces[0]],
b_terminations=[interfaces[1]],
)
cable1.clean()
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)
# Saving the same instance again must not duplicate its paths
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
)
# Reload so _terminations_modified starts False, as it does in a bulk edit
cable1 = Cable.objects.get(pk=cable1.pk)
cable1.profile = CableProfileChoices.SINGLE_1C1P
cable1.full_clean()
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)
# The profile change is applied once, so a second save must not recreate anything
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
)

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

@ -2133,6 +2133,33 @@ class CableTestCase(TestCase):
with self.assertRaises(ValidationError):
cable.clean()
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.
"""
cable = Cable.objects.first()
interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
termination_pks = set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True))
cable.profile = CableProfileChoices.SINGLE_1C1P
cable.save(update_fields=['label'])
interface1.refresh_from_db()
# Requery rather than refresh, so the pending profile stays on the instance under test
self.assertEqual(Cable.objects.get(pk=cable.pk).profile, '')
self.assertIsNone(interface1.cable_connector)
self.assertEqual(
set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)),
termination_pks
)
# _orig_profile was not advanced, so the pending change still applies here
cable.save()
interface1.refresh_from_db()
self.assertEqual(Cable.objects.get(pk=cable.pk).profile, CableProfileChoices.SINGLE_1C1P)
self.assertEqual(interface1.cable_connector, 1)
def test_cable_profile_change_preserves_terminations(self):
"""
When a Cable's profile is changed via save() without explicitly setting terminations (as happens during

39
netbox/dcim/tests/test_signals.py
View File

@ -796,8 +796,7 @@ class CableSignalTestCase(TestCase):
cable.save()
self.assertTrue(all(cp.is_active for cp in CablePath.objects.all()))
# Reload the cable so _orig_status reflects the persisted value and
# _terminations_modified resets to False.
# Reload to exercise status tracking on a freshly loaded instance, as a request does
cable = Cable.objects.get(pk=cable.pk)
cable.status = LinkStatusChoices.STATUS_PLANNED
cable.save()
@ -821,6 +820,42 @@ class CableSignalTestCase(TestCase):
self.assertTrue(all(cp.is_active for cp in CablePath.objects.all()))
def test_toggling_cable_status_on_one_instance_reactivates_paths(self):
interface_a = Interface.objects.create(device=self.device, name='Interface A')
interface_b = Interface.objects.create(device=self.device, name='Interface B')
cable = Cable(a_terminations=[interface_a], b_terminations=[interface_b])
cable.save()
# Reuse the same instance for both changes, as a script would
cable = Cable.objects.get(pk=cable.pk)
cable.status = LinkStatusChoices.STATUS_PLANNED
cable.save()
self.assertFalse(any(cp.is_active for cp in CablePath.objects.all()))
cable.status = LinkStatusChoices.STATUS_CONNECTED
cable.save()
self.assertTrue(all(cp.is_active for cp in CablePath.objects.all()))
def test_partial_save_does_not_consume_an_unwritten_status_change(self):
interface_a = Interface.objects.create(device=self.device, name='Interface A')
interface_b = Interface.objects.create(device=self.device, name='Interface B')
cable = Cable(
a_terminations=[interface_a],
b_terminations=[interface_b],
status=LinkStatusChoices.STATUS_PLANNED,
)
cable.save()
self.assertFalse(any(cp.is_active for cp in CablePath.objects.all()))
# A save that excludes status must not advance the status snapshot
cable.status = LinkStatusChoices.STATUS_CONNECTED
cable.save(update_fields=['label'])
self.assertFalse(any(cp.is_active for cp in CablePath.objects.all()))
# _orig_status was not advanced, so the change must still be detected
cable.save()
self.assertTrue(all(cp.is_active for cp in CablePath.objects.all()))
def test_deleting_cable_retraces_paths(self):
interface_a = Interface.objects.create(device=self.device, name='Interface A')
interface_b = Interface.objects.create(device=self.device, name='Interface B')