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 2 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
28 changes: 19 additions & 9 deletions netbox/dcim/models/cables.py
Original file line number Diff line number Diff line change
Expand Up @@ -237,7 +237,10 @@ def save(self, *args, **kwargs):
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 ValidationError as e:
raise ValidationError(f'{e}')
DanSheps marked this conversation as resolved.
Show resolved Hide resolved

def get_status_color(self):
return LinkStatusChoices.colors.get(self.status)
Expand Down Expand Up @@ -532,7 +535,9 @@ def from_origin(cls, terminations):

# 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:])
assert all(t.link == terminations[0].link for t in terminations[1:]), (
_("All originating terminations must start must be attached to the same link")
)

path = []
position_stack = []
Expand All @@ -543,12 +548,15 @@ 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:])
assert all(isinstance(t, type(terminations[0])) for t in terminations[1:]), (
_("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:])
assert all(t.parent_object == terminations[0].parent_object for t in terminations[1:]), (
_("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 @@ -571,8 +579,8 @@ 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)
assert all(type(link) in (Cable, WirelessLink) for link in links), _("All links must be cable or wireless")
assert all(isinstance(link, type(links[0])) for link in links), _("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 @@ -649,14 +657,16 @@ 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)
assert len(remote_terminations) == len(positions), (
_("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()
assert q_filter is not Q(), _("Remote termination query filter is empty, please open a bug report")
DanSheps marked this conversation as resolved.
Show resolved Hide resolved
front_ports = FrontPort.objects.filter(q_filter)
# Obtain the individual front ports based on the termination and position
elif position_stack:
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 @@ -3,6 +3,7 @@
from copy import deepcopy

from django.contrib import messages
from django.core.exceptions import ValidationError
from django.db import router, transaction
from django.db.models import ProtectedError, RestrictedError
from django.db.models.deletion import Collector
Expand Down Expand Up @@ -307,6 +308,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 ValidationError as e:
logger.debug(e.message)
form.add_error(None, e.message)
clear_events.send(sender=self)

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

Expand Down
Loading