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 10 commits into
base: develop
Choose a base branch
from
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
35 changes: 23 additions & 12 deletions netbox/dcim/models/cables.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
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
Expand All @@ -27,6 +28,7 @@
'CableTermination',
)

from ..exceptions import UnsupportedCablePath

trace_paths = Signal()

Expand Down Expand Up @@ -238,8 +240,10 @@ def save(self, *args, **kwargs):
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)
Expand Down Expand Up @@ -533,8 +537,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 be attached to the same link"))

path = []
position_stack = []
Expand All @@ -545,12 +549,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 +578,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 +658,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
6 changes: 4 additions & 2 deletions netbox/dcim/tests/test_cablepaths.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,11 @@

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
from utilities.exceptions import AbortRequest


class CablePathTestCase(TestCase):
Expand Down Expand Up @@ -2304,7 +2306,7 @@ def test_401_exclude_midspan_devices(self):
b_terminations=[frontport1, frontport3],
label='C1'
)
with self.assertRaises(AssertionError):
with self.assertRaises(AbortRequest):
cable1.save()

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

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

self.assertPathDoesNotExist(
Expand Down
Loading