mirror/netbox-community-netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
Code Activity

Merge branch 'develop' into 4121-filter-lookup-expressions

Browse Source
This commit is contained in:
John Anderson
2020-02-26 12:05:27 -05:00
committed by GitHub
parent 3b4607d30d 5a61bbec26
commit c908f132ec
97 changed files with 1272 additions and 587 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
netbox/utilities/api.py
View File

@@ -6,6 +6,7 @@ from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import FieldError, MultipleObjectsReturned, ObjectDoesNotExist
from django.db.models import ManyToManyField, ProtectedError
from django.http import Http404
from django.urls import reverse
from rest_framework.exceptions import APIException
from rest_framework.permissions import BasePermission
from rest_framework.relations import PrimaryKeyRelatedField, RelatedField
@@ -41,6 +42,14 @@ def get_serializer_for_model(model, prefix=''):
)
def is_api_request(request):
"""
Return True of the request is being made via the REST API.
"""
api_path = reverse('api-root')
return request.path_info.startswith(api_path)
#
# Authentication
#

7
netbox/utilities/fields.py
View File

@@ -56,8 +56,11 @@ class NaturalOrderingField(models.CharField):
"""
Generate a naturalized value from the target field
"""
value = getattr(model_instance, self.target_field)
return self.naturalize_function(value, max_length=self.max_length)
original_value = getattr(model_instance, self.target_field)
naturalized_value = self.naturalize_function(original_value, max_length=self.max_length)
setattr(model_instance, self.attname, naturalized_value)
return naturalized_value
def deconstruct(self):
kwargs = super().deconstruct()[3] # Pass kwargs from CharField

26
netbox/utilities/forms.py
View File

@@ -2,8 +2,9 @@ import csv
import json
import re
from io import StringIO
import yaml
import django_filters
import yaml
from django import forms
from django.conf import settings
from django.contrib.postgres.forms.jsonb import JSONField as _JSONField, InvalidJSONInput
@@ -132,6 +133,13 @@ class SmallTextarea(forms.Textarea):
pass
class SlugWidget(forms.TextInput):
"""
Subclass TextInput and add a slug regeneration button next to the form field.
"""
template_name = 'widgets/sluginput.html'
class ColorSelect(forms.Select):
"""
Extends the built-in Select widget to colorize each <option>.
@@ -534,7 +542,8 @@ class SlugField(forms.SlugField):
def __init__(self, slug_source='name', *args, **kwargs):
label = kwargs.pop('label', "Slug")
help_text = kwargs.pop('help_text', "URL-friendly unique shorthand")
super().__init__(label=label, help_text=help_text, *args, **kwargs)
widget = kwargs.pop('widget', SlugWidget)
super().__init__(label=label, help_text=help_text, widget=widget, *args, **kwargs)
self.widget.attrs['slug-source'] = slug_source
@@ -556,18 +565,17 @@ class TagFilterField(forms.MultipleChoiceField):
class DynamicModelChoiceMixin:
field_modifier = ''
filter = django_filters.ModelChoiceFilter
def get_bound_field(self, form, field_name):
bound_field = BoundField(form, self, field_name)
# Modify the QuerySet of the field before we return it. Limit choices to any data already bound: Options
# will be populated on-demand via the APISelect widget.
field_name = '{}{}'.format(self.to_field_name or 'pk', self.field_modifier)
if bound_field.data:
self.queryset = self.queryset.filter(**{field_name: self.prepare_value(bound_field.data)})
elif bound_field.initial:
self.queryset = self.queryset.filter(**{field_name: self.prepare_value(bound_field.initial)})
data = self.prepare_value(bound_field.data or bound_field.initial)
if data:
filter = self.filter(field_name=self.to_field_name or 'pk', queryset=self.queryset)
self.queryset = filter.filter(self.queryset, data)
else:
self.queryset = self.queryset.none()
@@ -586,7 +594,7 @@ class DynamicModelMultipleChoiceField(DynamicModelChoiceMixin, forms.ModelMultip
"""
A multiple-choice version of DynamicModelChoiceField.
"""
field_modifier = '__in'
filter = django_filters.ModelMultipleChoiceFilter
class LaxURLField(forms.URLField):

4
netbox/utilities/middleware.py
View File

@@ -5,6 +5,7 @@ from django.db import ProgrammingError
from django.http import Http404, HttpResponseRedirect
from django.urls import reverse
from .api import is_api_request
from .views import server_error
@@ -38,9 +39,8 @@ class APIVersionMiddleware(object):
self.get_response = get_response
def __call__(self, request):
api_path = reverse('api-root')
response = self.get_response(request)
if request.path_info.startswith(api_path):
if is_api_request(request):
response['API-Version'] = settings.REST_FRAMEWORK_VERSION
return response

25
netbox/utilities/ordering.py
View File

@@ -7,7 +7,8 @@ INTERFACE_NAME_REGEX = r'(^(?P<type>[^\d\.:]+)?)' \
r'((?P<subposition>\d+)/)?' \
r'((?P<id>\d+))?' \
r'(:(?P<channel>\d+))?' \
r'(.(?P<vc>\d+)$)?'
r'(\.(?P<vc>\d+))?' \
r'(?P<remainder>.*)$'
def naturalize(value, max_length, integer_places=8):
@@ -50,7 +51,7 @@ def naturalize_interface(value, max_length):
:param value: The value to be naturalized
:param max_length: The maximum length of the returned string. Characters beyond this length will be stripped.
"""
output = []
output = ''
match = re.search(INTERFACE_NAME_REGEX, value)
if match is None:
return value
@@ -60,21 +61,25 @@ def naturalize_interface(value, max_length):
for part_name in ('slot', 'subslot', 'position', 'subposition'):
part = match.group(part_name)
if part is not None:
output.append(part.rjust(4, '0'))
output += part.rjust(4, '0')
else:
output.append('9999')
output += '9999'
# Append the type, if any.
if match.group('type') is not None:
output.append(match.group('type'))
output += match.group('type')
# Finally, append any remaining fields, left-padding to six digits each.
# Append any remaining fields, left-padding to six digits each.
for part_name in ('id', 'channel', 'vc'):
part = match.group(part_name)
if part is not None:
output.append(part.rjust(6, '0'))
output += part.rjust(6, '0')
else:
output.append('000000')
output += '000000'
ret = ''.join(output)
return ret[:max_length]
# Finally, naturalize any remaining text and append it
if match.group('remainder') is not None and len(output) < max_length:
remainder = naturalize(match.group('remainder'), max_length - len(output))
output += remainder
return output[:max_length]

8
netbox/utilities/templates/widgets/sluginput.html Normal file
View File

@@ -0,0 +1,8 @@
<div class="input-group">
{% include "django/forms/widgets/input.html" %}
<span class="input-group-btn">
<button class="btn btn-default reslugify" type="button" title="Regenerate slug">
<i class="fa fa-refresh"></i>
</button>
</span>
</div>

91
netbox/utilities/testing/testcases.py
View File

@@ -172,24 +172,29 @@ class ViewTestCases:
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_create_object(self):
# Try GET without permission
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(self._get_url('add')), 403)
# Try GET with permission
self.add_permissions(
'{}.add_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.get(path=self._get_url('add'))
self.assertHttpStatus(response, 200)
# Try POST with permission
initial_count = self.model.objects.count()
request = {
'path': self._get_url('add'),
'data': post_data(self.form_data),
'follow': False, # Do not follow 302 redirects
}
# Attempt to make the request without required permissions
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(**request), 403)
# Assign the required permission and submit again
self.add_permissions(
'{}.add_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
# Validate object creation
self.assertEqual(initial_count + 1, self.model.objects.count())
instance = self.model.objects.order_by('-pk').first()
self.assertInstanceEqual(instance, self.form_data)
@@ -204,23 +209,27 @@ class ViewTestCases:
def test_edit_object(self):
instance = self.model.objects.first()
# Try GET without permission
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(self._get_url('edit', instance)), 403)
# Try GET with permission
self.add_permissions(
'{}.change_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.get(path=self._get_url('edit', instance))
self.assertHttpStatus(response, 200)
# Try POST with permission
request = {
'path': self._get_url('edit', instance),
'data': post_data(self.form_data),
'follow': False, # Do not follow 302 redirects
}
# Attempt to make the request without required permissions
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(**request), 403)
# Assign the required permission and submit again
self.add_permissions(
'{}.change_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
# Validate object modifications
instance = self.model.objects.get(pk=instance.pk)
self.assertInstanceEqual(instance, self.form_data)
@@ -232,23 +241,26 @@ class ViewTestCases:
def test_delete_object(self):
instance = self.model.objects.first()
# Try GET without permissions
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(self._get_url('delete', instance)), 403)
# Try GET with permission
self.add_permissions(
'{}.delete_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.get(path=self._get_url('delete', instance))
self.assertHttpStatus(response, 200)
request = {
'path': self._get_url('delete', instance),
'data': {'confirm': True},
'follow': False, # Do not follow 302 redirects
}
# Attempt to make the request without required permissions
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(**request), 403)
# Assign the required permission and submit again
self.add_permissions(
'{}.delete_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
# Validate object deletion
with self.assertRaises(ObjectDoesNotExist):
self.model.objects.get(pk=instance.pk)
@@ -314,6 +326,20 @@ class ViewTestCases:
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_import_objects(self):
# Test GET without permission
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.get(self._get_url('import')), 403)
# Test GET with permission
self.add_permissions(
'{}.view_{}'.format(self.model._meta.app_label, self.model._meta.model_name),
'{}.add_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.get(self._get_url('import'))
self.assertHttpStatus(response, 200)
# Test POST with permission
initial_count = self.model.objects.count()
request = {
'path': self._get_url('import'),
@@ -321,19 +347,10 @@ class ViewTestCases:
'csv': '\n'.join(self.csv_data)
}
}
# Attempt to make the request without required permissions
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(**request), 403)
# Assign the required permission and submit again
self.add_permissions(
'{}.view_{}'.format(self.model._meta.app_label, self.model._meta.model_name),
'{}.add_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.post(**request)
self.assertHttpStatus(response, 200)
# Validate import of new objects
self.assertEqual(self.model.objects.count(), initial_count + len(self.csv_data) - 1)
class BulkEditObjectsViewTestCase(ModelViewTestCase):

15
netbox/utilities/tests/test_ordering.py
View File

@@ -9,8 +9,8 @@ class NaturalizationTestCase(TestCase):
"""
def test_naturalize(self):
# Original, naturalized
data = (
# Original, naturalized
('abc', 'abc'),
('123', '00000123'),
('abc123', 'abc00000123'),
@@ -21,15 +21,16 @@ class NaturalizationTestCase(TestCase):
)
for origin, naturalized in data:
self.assertEqual(naturalize(origin, max_length=50), naturalized)
self.assertEqual(naturalize(origin, max_length=100), naturalized)
def test_naturalize_max_length(self):
self.assertEqual(naturalize('abc123def456', max_length=10), 'abc0000012')
def test_naturalize_interface(self):
# Original, naturalized
data = (
# Original, naturalized
# IOS/JunOS-style
('Gi', '9999999999999999Gi000000000000000000'),
('Gi1', '9999999999999999Gi000001000000000000'),
('Gi1/2', '0001999999999999Gi000002000000000000'),
@@ -40,10 +41,16 @@ class NaturalizationTestCase(TestCase):
('Gi1/2/3/4/5:6.7', '0001000200030004Gi000005000006000007'),
('Gi1:2', '9999999999999999Gi000001000002000000'),
('Gi1:2.3', '9999999999999999Gi000001000002000003'),
# Generic
('Interface 1', '9999999999999999Interface 000001000000000000'),
('Interface 1 (other)', '9999999999999999Interface 000001000000000000 (other)'),
('Interface 99', '9999999999999999Interface 000099000000000000'),
('PCIe1-p1', '9999999999999999PCIe000001000000000000-p00000001'),
('PCIe1-p99', '9999999999999999PCIe000001000000000000-p00000099'),
)
for origin, naturalized in data:
self.assertEqual(naturalize_interface(origin, max_length=50), naturalized)
self.assertEqual(naturalize_interface(origin, max_length=100), naturalized)
def test_naturalize_interface_max_length(self):
self.assertEqual(naturalize_interface('Gi1/2/3', max_length=20), '0001000299999999Gi00')

35
netbox/utilities/utils.py
View File

@@ -31,8 +31,9 @@ def csv_format(data):
if not isinstance(value, str):
value = '{}'.format(value)
# Double-quote the value if it contains a comma
# Double-quote the value if it contains a comma or line break
if ',' in value or '\n' in value:
value = value.replace('"', '""') # Escape double-quotes
csv.append('"{}"'.format(value))
else:
csv.append('{}'.format(value))
@@ -80,10 +81,12 @@ def get_subquery(model, field):
return subquery
def serialize_object(obj, extra=None):
def serialize_object(obj, extra=None, exclude=None):
"""
Return a generic JSON representation of an object using Django's built-in serializer. (This is used for things like
change logging, not the REST API.) Optionally include a dictionary to supplement the object data.
change logging, not the REST API.) Optionally include a dictionary to supplement the object data. A list of keys
can be provided to exclude them from the returned dictionary. Private fields (prefaced with an underscore) are
implicitly excluded.
"""
json_str = serialize('json', [obj])
data = json.loads(json_str)[0]['fields']
@@ -102,6 +105,16 @@ def serialize_object(obj, extra=None):
if extra is not None:
data.update(extra)
# Copy keys to list to avoid 'dictionary changed size during iteration' exception
for key in list(data):
# Private fields shouldn't be logged in the object change
if isinstance(key, str) and key.startswith('_'):
data.pop(key)
# Explicitly excluded keys
if isinstance(exclude, (list, tuple)) and key in exclude:
data.pop(key)
return data
@@ -222,3 +235,19 @@ def querydict_to_dict(querydict):
key: querydict.get(key) if len(value) == 1 and key != 'pk' else querydict.getlist(key)
for key, value in querydict.lists()
}
def shallow_compare_dict(source_dict, destination_dict, exclude=None):
"""
Return a new dictionary of the different keys. The values of `destination_dict` are returned. Only the equality of
the first layer of keys/values is checked. `exclude` is a list or tuple of keys to be ignored.
"""
difference = {}
for key in destination_dict:
if source_dict.get(key) != destination_dict[key]:
if isinstance(exclude, (list, tuple)) and key in exclude:
continue
difference[key] = destination_dict[key]
return difference

22
netbox/utilities/views.py
View File

@@ -626,12 +626,13 @@ class BulkEditView(GetReturnURLMixin, View):
model = self.queryset.model
# Create a mutable copy of the POST data
post_data = request.POST.copy()
# If we are editing *all* objects in the queryset, replace the PK list with all matched objects.
if post_data.get('_all') and self.filterset is not None:
post_data['pk'] = [obj.pk for obj in self.filterset(request.GET, model.objects.only('pk')).qs]
if request.POST.get('_all') and self.filterset is not None:
pk_list = [
obj.pk for obj in self.filterset(request.GET, model.objects.only('pk')).qs
]
else:
pk_list = request.POST.getlist('pk')
if '_apply' in request.POST:
form = self.form(model, request.POST)
@@ -656,9 +657,8 @@ class BulkEditView(GetReturnURLMixin, View):
try:
model_field = model._meta.get_field(name)
except FieldDoesNotExist:
# The form field is used to modify a field rather than set its value directly,
# so we skip it.
continue
# This form field is used to modify a field rather than set its value directly
model_field = None
# Handle nullification
if name in form.nullable_fields and name in nullified_fields:
@@ -716,12 +716,10 @@ class BulkEditView(GetReturnURLMixin, View):
messages.error(self.request, "{} failed validation: {}".format(obj, e))
else:
# Pass the PK list as initial data to avoid binding the form
initial_data = querydict_to_dict(post_data)
form = self.form(model, initial=initial_data)
form = self.form(model, initial={'pk': pk_list})
# Retrieve objects being edited
table = self.table(self.queryset.filter(pk__in=post_data.getlist('pk')), orderable=False)
table = self.table(self.queryset.filter(pk__in=pk_list), orderable=False)
if not table.rows:
messages.warning(request, "No {} were selected.".format(model._meta.verbose_name_plural))
return redirect(self.get_return_url(request))