From 4ab58f2da977108de34be6abaa96ffc4ff241e18 Mon Sep 17 00:00:00 2001 From: Daniel Sheppard Date: Tue, 4 Mar 2025 12:57:27 -0600 Subject: [PATCH] Fixes: #15016 - Catch AssertionError from cable trace and throw ValidationError (#16384) --- netbox/dcim/exceptions.py | 2 ++ netbox/dcim/models/cables.py | 35 ++++++++++++++++++---------- netbox/dcim/tests/test_cablepaths.py | 5 ++-- netbox/wireless/signals.py | 7 +++++- 4 files changed, 34 insertions(+), 15 deletions(-) create mode 100644 netbox/dcim/exceptions.py diff --git a/netbox/dcim/exceptions.py b/netbox/dcim/exceptions.py new file mode 100644 index 000000000..e4be1b5f1 --- /dev/null +++ b/netbox/dcim/exceptions.py @@ -0,0 +1,2 @@ +class UnsupportedCablePath(Exception): + pass diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index 7117ea7e0..81a742fe6 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -15,6 +15,7 @@ from dcim.fields import PathField from dcim.utils import decompile_path_node, object_to_path_node from netbox.models import ChangeLoggedModel, PrimaryModel from utilities.conversion import to_meters +from utilities.exceptions import AbortRequest from utilities.fields import ColorField from utilities.querysets import RestrictedQuerySet from wireless.models import WirelessLink @@ -26,6 +27,7 @@ __all__ = ( 'CableTermination', ) +from ..exceptions import UnsupportedCablePath trace_paths = Signal() @@ -236,8 +238,10 @@ class Cable(PrimaryModel): 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() - - trace_paths.send(Cable, instance=self, created=_created) + 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) @@ -531,8 +535,8 @@ class CablePath(models.Model): return None # Ensure all originating terminations are attached to the same link - if len(terminations) > 1: - assert all(t.link == terminations[0].link for t in terminations[1:]) + if len(terminations) > 1 and not all(t.link == terminations[0].link for t in terminations[1:]): + raise UnsupportedCablePath(_("All originating terminations must be attached to the same link")) path = [] position_stack = [] @@ -543,12 +547,13 @@ class CablePath(models.Model): while terminations: # Terminations must all be of the same type - assert all(isinstance(t, type(terminations[0])) for t in terminations[1:]) + if not all(isinstance(t, type(terminations[0])) for t in terminations[1:]): + raise UnsupportedCablePath(_("All mid-span terminations must have the same termination type")) # All mid-span terminations must all be attached to the same device - if not isinstance(terminations[0], PathEndpoint): - assert all(isinstance(t, type(terminations[0])) for t in terminations[1:]) - assert all(t.parent_object == terminations[0].parent_object for t in terminations[1:]) + if (not isinstance(terminations[0], PathEndpoint) and not + all(t.parent_object == terminations[0].parent_object for t in terminations[1:])): + raise UnsupportedCablePath(_("All mid-span terminations must have the same parent object")) # Check for a split path (e.g. rear port fanning out to multiple front ports with # different cables attached) @@ -571,8 +576,10 @@ class CablePath(models.Model): return None # Otherwise, halt the trace if no link exists break - assert all(type(link) in (Cable, WirelessLink) for link in links) - assert all(isinstance(link, type(links[0])) for link in links) + if not all(type(link) in (Cable, WirelessLink) for link in links): + raise UnsupportedCablePath(_("All links must be cable or wireless")) + if not all(isinstance(link, type(links[0])) for link in links): + raise UnsupportedCablePath(_("All links must match first link type")) # Step 3: Record asymmetric paths as split not_connected_terminations = [termination.link for termination in terminations if termination.link is None] @@ -653,14 +660,18 @@ class CablePath(models.Model): positions = position_stack.pop() # Ensure we have a number of positions equal to the amount of remote terminations - assert len(remote_terminations) == len(positions) + if len(remote_terminations) != len(positions): + raise UnsupportedCablePath( + _("All positions counts within the path on opposite ends of links must match") + ) # Get our front ports q_filter = Q() for rt in remote_terminations: position = positions.pop() q_filter |= Q(rear_port_id=rt.pk, rear_port_position=position) - assert q_filter is not Q() + if q_filter is Q(): + raise UnsupportedCablePath(_("Remote termination position filter is missing")) front_ports = FrontPort.objects.filter(q_filter) # Obtain the individual front ports based on the termination and position elif position_stack: diff --git a/netbox/dcim/tests/test_cablepaths.py b/netbox/dcim/tests/test_cablepaths.py index 1acc9a8a1..399478e70 100644 --- a/netbox/dcim/tests/test_cablepaths.py +++ b/netbox/dcim/tests/test_cablepaths.py @@ -5,6 +5,7 @@ from dcim.choices import LinkStatusChoices from dcim.models import * from dcim.svg import CableTraceSVG from dcim.utils import object_to_path_node +from utilities.exceptions import AbortRequest class CablePathTestCase(TestCase): @@ -2470,7 +2471,7 @@ class CablePathTestCase(TestCase): b_terminations=[frontport1, frontport3], label='C1' ) - with self.assertRaises(AssertionError): + with self.assertRaises(AbortRequest): cable1.save() self.assertPathDoesNotExist( @@ -2489,7 +2490,7 @@ class CablePathTestCase(TestCase): label='C3' ) - with self.assertRaises(AssertionError): + with self.assertRaises(AbortRequest): cable3.save() self.assertPathDoesNotExist( diff --git a/netbox/wireless/signals.py b/netbox/wireless/signals.py index ff7b1229c..b1a2d2feb 100644 --- a/netbox/wireless/signals.py +++ b/netbox/wireless/signals.py @@ -3,8 +3,10 @@ import logging from django.db.models.signals import post_save, post_delete from django.dispatch import receiver +from dcim.exceptions import UnsupportedCablePath from dcim.models import CablePath, Interface from dcim.utils import create_cablepath +from utilities.exceptions import AbortRequest from .models import WirelessLink @@ -34,7 +36,10 @@ def update_connected_interfaces(instance, created, raw=False, **kwargs): # Create/update cable paths if created: for interface in (instance.interface_a, instance.interface_b): - create_cablepath([interface]) + try: + create_cablepath([interface]) + except UnsupportedCablePath as e: + raise AbortRequest(e) @receiver(post_delete, sender=WirelessLink)