Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixes: #15016 - Catch AssertionError from cable trace and throw ValidationError #16384

Open
wants to merge 5 commits into
base: develop
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions netbox/dcim/exceptions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
class UnsupportedCablePath(Exception):
pass
28 changes: 18 additions & 10 deletions netbox/dcim/models/cables.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
'CableTermination',
)

from ..exceptions import UnsupportedCablePath

trace_paths = Signal()

Expand Down Expand Up @@ -533,8 +534,8 @@ def from_origin(cls, terminations):
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 start must be attached to the same link"))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please review the message content, the sentence is incorrect in its current state


path = []
position_stack = []
Expand All @@ -545,12 +546,13 @@ def from_origin(cls, terminations):
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)
Expand All @@ -573,8 +575,10 @@ def from_origin(cls, terminations):
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]
Expand Down Expand Up @@ -651,14 +655,18 @@ def from_origin(cls, terminations):
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:
Expand Down
5 changes: 3 additions & 2 deletions netbox/dcim/tests/test_cablepaths.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from circuits.models import *
from dcim.choices import LinkStatusChoices
from dcim.exceptions import UnsupportedCablePath
from dcim.models import *
from dcim.svg import CableTraceSVG
from dcim.utils import object_to_path_node
Expand Down Expand Up @@ -2261,7 +2262,7 @@ def test_401_exclude_midspan_devices(self):
b_terminations=[frontport1, frontport3],
label='C1'
)
with self.assertRaises(AssertionError):
with self.assertRaises(UnsupportedCablePath):
cable1.save()

self.assertPathDoesNotExist(
Expand All @@ -2280,7 +2281,7 @@ def test_401_exclude_midspan_devices(self):
label='C3'
)

with self.assertRaises(AssertionError):
with self.assertRaises(UnsupportedCablePath):
cable3.save()

self.assertPathDoesNotExist(
Expand Down
7 changes: 7 additions & 0 deletions netbox/netbox/views/generic/object_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from django.utils.safestring import mark_safe
from django.utils.translation import gettext as _

from dcim.exceptions import UnsupportedCablePath
from core.signals import clear_events
from utilities.error_handlers import handle_protectederror
from utilities.exceptions import AbortRequest, PermissionsViolation
Expand Down Expand Up @@ -319,6 +320,12 @@ def post(self, request, *args, **kwargs):
form.add_error(None, e.message)
clear_events.send(sender=self)

# Catch any validation errors thrown in the model.save() or form.save() methods
except UnsupportedCablePath as e:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think this is the right place to handle the exception, for two reasons. First, we want to avoid introducing any model-specific logic within a generic view. Second, this addresses only the UI workflow; it doesn't account for an API-driven change.

A cleaner approach might be to catch UnsupportedCablePath raised by trace_paths.send() under Cable.save() and raise an AbortRequest exception in turn. We should be able to preserve the error message text in doing so.

logger.debug(e.message)
form.add_error(None, e.message)
clear_events.send(sender=self)

else:
logger.debug("Form validation failed")

Expand Down