From 478755a8e510dff2c2281c44c388c1ff1f6e5c7f Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Thu, 4 Sep 2025 10:58:41 -0400 Subject: [PATCH 1/8] Adapt testing utilities to disregard "update" changelog records immediately following object creation --- netbox/utilities/testing/api.py | 8 ++++---- netbox/utilities/testing/views.py | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/netbox/utilities/testing/api.py b/netbox/utilities/testing/api.py index 8df8f44389a..1fe8813679e 100644 --- a/netbox/utilities/testing/api.py +++ b/netbox/utilities/testing/api.py @@ -247,9 +247,9 @@ def test_create_object(self): if issubclass(self.model, ChangeLoggingMixin): objectchange = ObjectChange.objects.get( changed_object_type=ContentType.objects.get_for_model(instance), - changed_object_id=instance.pk + changed_object_id=instance.pk, + action=ObjectChangeActionChoices.ACTION_CREATE, ) - self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_CREATE) self.assertEqual(objectchange.message, data['changelog_message']) def test_bulk_create_objects(self): @@ -298,11 +298,11 @@ def test_bulk_create_objects(self): ] objectchanges = ObjectChange.objects.filter( changed_object_type=ContentType.objects.get_for_model(self.model), - changed_object_id__in=id_list + changed_object_id__in=id_list, + action=ObjectChangeActionChoices.ACTION_CREATE, ) self.assertEqual(len(objectchanges), len(self.create_data)) for oc in objectchanges: - self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE) self.assertEqual(oc.message, changelog_message) class UpdateObjectViewTestCase(APITestCase): diff --git a/netbox/utilities/testing/views.py b/netbox/utilities/testing/views.py index da8a870984f..99a6dd43aea 100644 --- a/netbox/utilities/testing/views.py +++ b/netbox/utilities/testing/views.py @@ -655,11 +655,11 @@ def test_bulk_import_objects_with_permission(self): self.assertIsNotNone(request_id, "Unable to determine request ID from response") objectchanges = ObjectChange.objects.filter( changed_object_type=ContentType.objects.get_for_model(self.model), - request_id=request_id + request_id=request_id, + action=ObjectChangeActionChoices.ACTION_CREATE, ) self.assertEqual(len(objectchanges), len(self.csv_data) - 1) for oc in objectchanges: - self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE) self.assertEqual(oc.message, data['changelog_message']) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) From 48378c502fa8990ed0a7f1486aa75f0d783992a0 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Thu, 4 Sep 2025 11:18:53 -0400 Subject: [PATCH 2/8] Closes #20241: Record A & B terminations on cable changelog record --- netbox/dcim/models/cables.py | 86 +++++++++++++++++++++++++----------- 1 file changed, 60 insertions(+), 26 deletions(-) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index 69e07ed9432..5adf300221e 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -18,6 +18,7 @@ from utilities.exceptions import AbortRequest from utilities.fields import ColorField, GenericArrayForeignKey from utilities.querysets import RestrictedQuerySet +from utilities.serialization import serialize_object from wireless.models import WirelessLink from .device_components import FrontPort, RearPort, PathEndpoint @@ -119,6 +120,9 @@ def __str__(self): pk = self.pk or self._pk return self.label or f'#{pk}' + def get_status_color(self): + return LinkStatusChoices.colors.get(self.status) + @property def a_terminations(self): if hasattr(self, '_a_terminations'): @@ -208,7 +212,7 @@ def clean(self): for termination in self.b_terminations: CableTermination(cable=self, cable_end='B', termination=termination).clean() - def save(self, *args, **kwargs): + def save(self, *args, force_insert=False, force_update=False, using=None, update_fields=None): _created = self.pk is None # Store the given length (if any) in meters for use in database ordering @@ -221,39 +225,69 @@ def save(self, *args, **kwargs): if self.length is None: self.length_unit = None - super().save(*args, **kwargs) + # 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) + # Update the private PK used in __str__() + self._pk = self.pk - # Update the private pk used in __str__ in case this is a new object (i.e. just got its pk) - self._pk = self.pk + if self._terminations_modified: + self.update_terminations() - # Retrieve existing A/B terminations for the Cable - a_terminations = {ct.termination: ct for ct in self.terminations.filter(cable_end='A')} - b_terminations = {ct.termination: ct for ct in self.terminations.filter(cable_end='B')} + super().save(*args, force_update=True, using=using, update_fields=update_fields) - # Delete stale CableTerminations - if self._terminations_modified: - for termination, ct in a_terminations.items(): - if termination.pk and termination not in self.a_terminations: - ct.delete() - for termination, ct in b_terminations.items(): - if termination.pk and termination not in self.b_terminations: - ct.delete() - - # Save new CableTerminations (if any) - if self._terminations_modified: - for termination in self.a_terminations: - if not termination.pk or termination not in a_terminations: - CableTermination(cable=self, cable_end='A', termination=termination).save() - for termination in self.b_terminations: - if not termination.pk or termination not in b_terminations: - CableTermination(cable=self, cable_end='B', termination=termination).save() try: trace_paths.send(Cable, instance=self, created=_created) except UnsupportedCablePath as e: raise AbortRequest(e) - def get_status_color(self): - return LinkStatusChoices.colors.get(self.status) + def serialize_object(self, exclude=None): + data = serialize_object(self, exclude=exclude or []) + + # Add A & B terminations to the serialized data + a_terminations, b_terminations = self.get_terminations() + data['a_terminations'] = sorted([ct.pk for ct in a_terminations.values()]) + data['b_terminations'] = sorted([ct.pk for ct in b_terminations.values()]) + + return data + + def get_terminations(self): + """ + Return two dictionaries mapping A & B side terminating objects to their corresponding CableTerminations + for this Cable. + """ + a_terminations = {} + b_terminations = {} + + for ct in CableTermination.objects.filter(cable=self).prefetch_related('termination'): + if ct.cable_end == CableEndChoices.SIDE_A: + a_terminations[ct.termination] = ct + else: + b_terminations[ct.termination] = ct + + return a_terminations, b_terminations + + def update_terminations(self): + """ + Create/delete CableTerminations for this Cable to reflect its current state. + """ + a_terminations, b_terminations = self.get_terminations() + + # Delete any stale CableTerminations + for termination, ct in a_terminations.items(): + if termination.pk and termination not in self.a_terminations: + ct.delete() + for termination, ct in b_terminations.items(): + if termination.pk and termination not in self.b_terminations: + ct.delete() + + # Save any new CableTerminations + for termination in self.a_terminations: + if not termination.pk or termination not in a_terminations: + CableTermination(cable=self, cable_end='A', termination=termination).save() + for termination in self.b_terminations: + if not termination.pk or termination not in b_terminations: + CableTermination(cable=self, cable_end='B', termination=termination).save() class CableTermination(ChangeLoggedModel): From f5e69e5457924fab8b1f81896ad668b5d15f41bb Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Fri, 5 Sep 2025 14:34:20 -0400 Subject: [PATCH 3/8] Add deserialize_object() to Cable --- netbox/dcim/models/cables.py | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index 5adf300221e..b529fe98768 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -18,7 +18,7 @@ from utilities.exceptions import AbortRequest from utilities.fields import ColorField, GenericArrayForeignKey from utilities.querysets import RestrictedQuerySet -from utilities.serialization import serialize_object +from utilities.serialization import deserialize_object, serialize_object from wireless.models import WirelessLink from .device_components import FrontPort, RearPort, PathEndpoint @@ -108,6 +108,8 @@ def __init__(self, *args, a_terminations=None, b_terminations=None, **kwargs): # Cache the original status so we can check later if it's been changed self._orig_status = self.__dict__.get('status') + self._a_terminations = [] + self._b_terminations = [] self._terminations_modified = False # Assign or retrieve A/B terminations @@ -251,6 +253,24 @@ def serialize_object(self, exclude=None): return data + @classmethod + def deserialize_object(cls, data, pk=None): + a_terminations = data.pop('a_terminations', []) + b_terminations = data.pop('b_terminations', []) + + instance = deserialize_object(cls, data, pk=pk) + + # Assign A & B termination objects to the Cable instance + queryset = CableTermination.objects.prefetch_related('termination') + instance.a_terminations = [ + ct.termination for ct in queryset.filter(pk__in=a_terminations) + ] + instance.b_terminations = [ + ct.termination for ct in queryset.filter(pk__in=b_terminations) + ] + + return instance + def get_terminations(self): """ Return two dictionaries mapping A & B side terminating objects to their corresponding CableTerminations From 38610f7aafe1de84f4b6222c98b456615ce934f9 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Mon, 8 Sep 2025 16:54:48 -0400 Subject: [PATCH 4/8] Fix tests --- netbox/dcim/models/cables.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index b529fe98768..a033198c183 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -108,8 +108,8 @@ def __init__(self, *args, a_terminations=None, b_terminations=None, **kwargs): # Cache the original status so we can check later if it's been changed self._orig_status = self.__dict__.get('status') - self._a_terminations = [] - self._b_terminations = [] + # self._a_terminations = [] + # self._b_terminations = [] self._terminations_modified = False # Assign or retrieve A/B terminations From 43492181d4261bc20f5113f0bc97c64e51024bad Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Tue, 9 Sep 2025 08:46:42 -0400 Subject: [PATCH 5/8] Consolidate termination assignment into _get_terminations() and _set_terminations() methods --- netbox/dcim/models/cables.py | 56 ++++++++++++++++++++---------------- 1 file changed, 31 insertions(+), 25 deletions(-) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index a033198c183..d6a40704a0c 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -108,8 +108,6 @@ def __init__(self, *args, a_terminations=None, b_terminations=None, **kwargs): # Cache the original status so we can check later if it's been changed self._orig_status = self.__dict__.get('status') - # self._a_terminations = [] - # self._b_terminations = [] self._terminations_modified = False # Assign or retrieve A/B terminations @@ -125,43 +123,51 @@ def __str__(self): def get_status_color(self): return LinkStatusChoices.colors.get(self.status) - @property - def a_terminations(self): - if hasattr(self, '_a_terminations'): - return self._a_terminations + def _get_terminations(self, side): + """ + Return the terminating objects for the given cable end (A or B). + """ + if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B): + raise ValueError("Unknown cable side: {side") + attr = f'_{side.lower()}_terminations' + if hasattr(self, attr): + return getattr(self, attr) if not self.pk: return [] - - # Query self.terminations.all() to leverage cached results return [ - ct.termination for ct in self.terminations.all() if ct.cable_end == CableEndChoices.SIDE_A + # Query self.terminations.all() to leverage cached results + ct.termination for ct in self.terminations.all() if ct.cable_end == side ] - @a_terminations.setter - def a_terminations(self, value): - if not self.pk or self.a_terminations != list(value): + def _set_terminations(self, side, value): + """ + Set the terminating objects for the given cable end (A or B). + """ + if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B): + raise ValueError("Unknown cable side: {side") + public_attr = f'{side.lower()}_terminations' + private_attr = f'_{public_attr}' + + if not self.pk or getattr(self, public_attr) != list(value): self._terminations_modified = True - self._a_terminations = value + setattr(self, private_attr, value) @property - def b_terminations(self): - if hasattr(self, '_b_terminations'): - return self._b_terminations + def a_terminations(self): + return self._get_terminations(CableEndChoices.SIDE_A) - if not self.pk: - return [] + @a_terminations.setter + def a_terminations(self, value): + self._set_terminations(CableEndChoices.SIDE_A, value) - # Query self.terminations.all() to leverage cached results - return [ - ct.termination for ct in self.terminations.all() if ct.cable_end == CableEndChoices.SIDE_B - ] + @property + def b_terminations(self): + return self._get_terminations(CableEndChoices.SIDE_B) @b_terminations.setter def b_terminations(self, value): - if not self.pk or self.b_terminations != list(value): - self._terminations_modified = True - self._b_terminations = value + self._set_terminations(CableEndChoices.SIDE_B, value) @property def color_name(self): From 0402dce4c4486ae72361ed8665e42c04cc2be98f Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Tue, 9 Sep 2025 09:11:00 -0400 Subject: [PATCH 6/8] Support setting cable terminations from serialized CableTermination IDs --- netbox/dcim/models/cables.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index d6a40704a0c..c4a79f10a56 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -146,12 +146,19 @@ def _set_terminations(self, side, value): """ if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B): raise ValueError("Unknown cable side: {side") - public_attr = f'{side.lower()}_terminations' - private_attr = f'_{public_attr}' + _attr = f'_{side.lower()}_terminations' - if not self.pk or getattr(self, public_attr) != list(value): + # If the provided value is a list of CableTermination IDs, resolve them + # to their corresponding termination objects. + if all(isinstance(item, int) for item in value): + value = [ + ct.termination for ct in CableTermination.objects.filter(pk__in=value).prefetch_related('termination') + ] + + if not self.pk or getattr(self, _attr, []) != list(value): self._terminations_modified = True - setattr(self, private_attr, value) + + setattr(self, _attr, value) @property def a_terminations(self): From 4c66e9a96c7a7029f401a7c03ca926893c81761d Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Tue, 9 Sep 2025 09:22:35 -0400 Subject: [PATCH 7/8] Rename getter & setter utilities for clarity --- netbox/dcim/models/cables.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index c4a79f10a56..bf9da75568c 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -123,7 +123,7 @@ def __str__(self): def get_status_color(self): return LinkStatusChoices.colors.get(self.status) - def _get_terminations(self, side): + def _get_x_terminations(self, side): """ Return the terminating objects for the given cable end (A or B). """ @@ -140,7 +140,7 @@ def _get_terminations(self, side): ct.termination for ct in self.terminations.all() if ct.cable_end == side ] - def _set_terminations(self, side, value): + def _set_x_terminations(self, side, value): """ Set the terminating objects for the given cable end (A or B). """ @@ -162,19 +162,19 @@ def _set_terminations(self, side, value): @property def a_terminations(self): - return self._get_terminations(CableEndChoices.SIDE_A) + return self._get_x_terminations(CableEndChoices.SIDE_A) @a_terminations.setter def a_terminations(self, value): - self._set_terminations(CableEndChoices.SIDE_A, value) + self._set_x_terminations(CableEndChoices.SIDE_A, value) @property def b_terminations(self): - return self._get_terminations(CableEndChoices.SIDE_B) + return self._get_x_terminations(CableEndChoices.SIDE_B) @b_terminations.setter def b_terminations(self, value): - self._set_terminations(CableEndChoices.SIDE_B, value) + self._set_x_terminations(CableEndChoices.SIDE_B, value) @property def color_name(self): From 50b836473b707f536676d037516e1647862a3339 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Tue, 9 Sep 2025 11:57:26 -0400 Subject: [PATCH 8/8] Fix ValueError message formatting --- netbox/dcim/models/cables.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index bf9da75568c..89c9a99b419 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -128,7 +128,7 @@ def _get_x_terminations(self, side): Return the terminating objects for the given cable end (A or B). """ if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B): - raise ValueError("Unknown cable side: {side") + raise ValueError(f"Unknown cable side: {side}") attr = f'_{side.lower()}_terminations' if hasattr(self, attr): @@ -145,7 +145,7 @@ def _set_x_terminations(self, side, value): Set the terminating objects for the given cable end (A or B). """ if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B): - raise ValueError("Unknown cable side: {side") + raise ValueError(f"Unknown cable side: {side}") _attr = f'_{side.lower()}_terminations' # If the provided value is a list of CableTermination IDs, resolve them