NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-12-18 11:22:25 -06:00
Code Issues Actions 10 Packages Projects Releases Wiki Activity

Merge branch 'develop' into 2921-tags-select2

Browse Source
This commit is contained in:
Jeremy Stretch
2020-02-04 09:37:31 -05:00
committed by GitHub
parent 7897ebb2ed efb7f151ec
commit 0d41d12267
49 changed files with 2290 additions and 1546 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
netbox/circuits/forms.py
View File

@@ -2,7 +2,9 @@ from django import forms
from taggit.forms import TagField
from dcim.models import Region, Site
from extras.forms import AddRemoveTagsForm, CustomFieldForm, CustomFieldBulkEditForm, CustomFieldFilterForm
from extras.forms import (
AddRemoveTagsForm, CustomFieldBulkEditForm, CustomFieldFilterForm, CustomFieldModelForm, CustomFieldModelCSVForm,
)
from tenancy.forms import TenancyFilterForm, TenancyForm
from tenancy.models import Tenant
from utilities.forms import (
@@ -17,7 +19,7 @@ from .models import Circuit, CircuitTermination, CircuitType, Provider
# Providers
#
class ProviderForm(BootstrapMixin, CustomFieldForm):
class ProviderForm(BootstrapMixin, CustomFieldModelForm):
slug = SlugField()
comments = CommentField()
tags = TagField(
@@ -46,7 +48,7 @@ class ProviderForm(BootstrapMixin, CustomFieldForm):
}
class ProviderCSVForm(forms.ModelForm):
class ProviderCSVForm(CustomFieldModelCSVForm):
slug = SlugField()
class Meta:
@@ -161,7 +163,7 @@ class CircuitTypeCSVForm(forms.ModelForm):
# Circuits
#
class CircuitForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class CircuitForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
comments = CommentField()
tags = TagField(
required=False
@@ -189,7 +191,7 @@ class CircuitForm(BootstrapMixin, TenancyForm, CustomFieldForm):
}
class CircuitCSVForm(forms.ModelForm):
class CircuitCSVForm(CustomFieldModelCSVForm):
provider = forms.ModelChoiceField(
queryset=Provider.objects.all(),
to_field_name='name',

175
netbox/circuits/tests/test_views.py
View File

@@ -1,23 +1,15 @@
import urllib.parse
from django.test import Client, TestCase
from django.urls import reverse
import datetime
from circuits.choices import *
from circuits.models import Circuit, CircuitType, Provider
from utilities.testing import create_test_user
from utilities.testing import StandardTestCases
class ProviderTestCase(TestCase):
class ProviderTestCase(StandardTestCases.Views):
model = Provider
def setUp(self):
user = create_test_user(
permissions=[
'circuits.view_provider',
'circuits.add_provider',
]
)
self.client = Client()
self.client.force_login(user)
@classmethod
def setUpTestData(cls):
Provider.objects.bulk_create([
Provider(name='Provider 1', slug='provider-1', asn=65001),
@@ -25,48 +17,45 @@ class ProviderTestCase(TestCase):
Provider(name='Provider 3', slug='provider-3', asn=65003),
])
def test_provider_list(self):
url = reverse('circuits:provider_list')
params = {
"q": "test",
cls.form_data = {
'name': 'Provider X',
'slug': 'provider-x',
'asn': 65123,
'account': '1234',
'portal_url': 'http://example.com/portal',
'noc_contact': 'noc@example.com',
'admin_contact': 'admin@example.com',
'comments': 'Another provider',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_provider(self):
provider = Provider.objects.first()
response = self.client.get(provider.get_absolute_url())
self.assertEqual(response.status_code, 200)
def test_provider_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"Provider 4,provider-4",
"Provider 5,provider-5",
"Provider 6,provider-6",
)
response = self.client.post(reverse('circuits:provider_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(Provider.objects.count(), 6)
cls.bulk_edit_data = {
'asn': 65009,
'account': '5678',
'portal_url': 'http://example.com/portal2',
'noc_contact': 'noc2@example.com',
'admin_contact': 'admin2@example.com',
'comments': 'New comments',
}
class CircuitTypeTestCase(TestCase):
class CircuitTypeTestCase(StandardTestCases.Views):
model = CircuitType
def setUp(self):
user = create_test_user(
permissions=[
'circuits.view_circuittype',
'circuits.add_circuittype',
]
)
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_get_object = None
test_delete_object = None
test_bulk_edit_objects = None
@classmethod
def setUpTestData(cls):
CircuitType.objects.bulk_create([
CircuitType(name='Circuit Type 1', slug='circuit-type-1'),
@@ -74,79 +63,71 @@ class CircuitTypeTestCase(TestCase):
CircuitType(name='Circuit Type 3', slug='circuit-type-3'),
])
def test_circuittype_list(self):
cls.form_data = {
'name': 'Circuit Type X',
'slug': 'circuit-type-x',
'description': 'A new circuit type',
}
url = reverse('circuits:circuittype_list')
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
def test_circuittype_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"Circuit Type 4,circuit-type-4",
"Circuit Type 5,circuit-type-5",
"Circuit Type 6,circuit-type-6",
)
response = self.client.post(reverse('circuits:circuittype_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(CircuitType.objects.count(), 6)
class CircuitTestCase(StandardTestCases.Views):
model = Circuit
@classmethod
def setUpTestData(cls):
class CircuitTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'circuits.view_circuit',
'circuits.add_circuit',
]
providers = (
Provider(name='Provider 1', slug='provider-1', asn=65001),
Provider(name='Provider 2', slug='provider-2', asn=65002),
)
self.client = Client()
self.client.force_login(user)
Provider.objects.bulk_create(providers)
provider = Provider(name='Provider 1', slug='provider-1', asn=65001)
provider.save()
circuittype = CircuitType(name='Circuit Type 1', slug='circuit-type-1')
circuittype.save()
circuittypes = (
CircuitType(name='Circuit Type 1', slug='circuit-type-1'),
CircuitType(name='Circuit Type 2', slug='circuit-type-2'),
)
CircuitType.objects.bulk_create(circuittypes)
Circuit.objects.bulk_create([
Circuit(cid='Circuit 1', provider=provider, type=circuittype),
Circuit(cid='Circuit 2', provider=provider, type=circuittype),
Circuit(cid='Circuit 3', provider=provider, type=circuittype),
Circuit(cid='Circuit 1', provider=providers[0], type=circuittypes[0]),
Circuit(cid='Circuit 2', provider=providers[0], type=circuittypes[0]),
Circuit(cid='Circuit 3', provider=providers[0], type=circuittypes[0]),
])
def test_circuit_list(self):
url = reverse('circuits:circuit_list')
params = {
"provider": Provider.objects.first().slug,
"type": CircuitType.objects.first().slug,
cls.form_data = {
'cid': 'Circuit X',
'provider': providers[1].pk,
'type': circuittypes[1].pk,
'status': CircuitStatusChoices.STATUS_DECOMMISSIONED,
'tenant': None,
'install_date': datetime.date(2020, 1, 1),
'commit_rate': 1000,
'description': 'A new circuit',
'comments': 'Some comments',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_circuit(self):
circuit = Circuit.objects.first()
response = self.client.get(circuit.get_absolute_url())
self.assertEqual(response.status_code, 200)
def test_circuit_import(self):
csv_data = (
cls.csv_data = (
"cid,provider,type",
"Circuit 4,Provider 1,Circuit Type 1",
"Circuit 5,Provider 1,Circuit Type 1",
"Circuit 6,Provider 1,Circuit Type 1",
)
response = self.client.post(reverse('circuits:circuit_import'), {'csv': '\n'.join(csv_data)})
cls.bulk_edit_data = {
'provider': providers[1].pk,
'type': circuittypes[1].pk,
'status': CircuitStatusChoices.STATUS_DECOMMISSIONED,
'tenant': None,
'commit_rate': 2000,
'description': 'New description',
'comments': 'New comments',
self.assertEqual(response.status_code, 200)
self.assertEqual(Circuit.objects.count(), 6)
}

26
netbox/dcim/forms.py
View File

@@ -13,7 +13,8 @@ from timezone_field import TimeZoneFormField
from circuits.models import Circuit, Provider
from extras.forms import (
AddRemoveTagsForm, CustomFieldForm, CustomFieldBulkEditForm, CustomFieldFilterForm, LocalConfigContextFilterForm
AddRemoveTagsForm, CustomFieldBulkEditForm, CustomFieldModelCSVForm, CustomFieldFilterForm, CustomFieldModelForm,
LocalConfigContextFilterForm,
)
from ipam.constants import BGP_ASN_MAX, BGP_ASN_MIN
from ipam.models import IPAddress, VLAN
@@ -216,7 +217,7 @@ class RegionFilterForm(BootstrapMixin, forms.Form):
# Sites
#
class SiteForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class SiteForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
region = TreeNodeChoiceField(
queryset=Region.objects.all(),
required=False,
@@ -264,7 +265,7 @@ class SiteForm(BootstrapMixin, TenancyForm, CustomFieldForm):
}
class SiteCSVForm(forms.ModelForm):
class SiteCSVForm(CustomFieldModelCSVForm):
status = CSVChoiceField(
choices=SiteStatusChoices,
required=False,
@@ -461,7 +462,7 @@ class RackRoleCSVForm(forms.ModelForm):
# Racks
#
class RackForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class RackForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
group = ChainedModelChoiceField(
queryset=RackGroup.objects.all(),
chains=(
@@ -506,7 +507,7 @@ class RackForm(BootstrapMixin, TenancyForm, CustomFieldForm):
}
class RackCSVForm(forms.ModelForm):
class RackCSVForm(CustomFieldModelCSVForm):
site = forms.ModelChoiceField(
queryset=Site.objects.all(),
to_field_name='name',
@@ -900,7 +901,7 @@ class ManufacturerCSVForm(forms.ModelForm):
# Device types
#
class DeviceTypeForm(BootstrapMixin, CustomFieldForm):
class DeviceTypeForm(BootstrapMixin, CustomFieldModelForm):
slug = SlugField(
slug_source='model'
)
@@ -1520,7 +1521,7 @@ class PlatformCSVForm(forms.ModelForm):
# Devices
#
class DeviceForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class DeviceForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
site = forms.ModelChoiceField(
queryset=Site.objects.all(),
widget=APISelect(
@@ -1552,6 +1553,7 @@ class DeviceForm(BootstrapMixin, TenancyForm, CustomFieldForm):
)
manufacturer = forms.ModelChoiceField(
queryset=Manufacturer.objects.all(),
required=False,
widget=APISelect(
api_url="/api/dcim/manufacturers/",
filter_for={
@@ -1728,7 +1730,7 @@ class DeviceForm(BootstrapMixin, TenancyForm, CustomFieldForm):
self.initial['rack'] = self.instance.parent_bay.device.rack_id
class BaseDeviceCSVForm(forms.ModelForm):
class BaseDeviceCSVForm(CustomFieldModelCSVForm):
device_role = forms.ModelChoiceField(
queryset=DeviceRole.objects.all(),
to_field_name='name',
@@ -2736,7 +2738,7 @@ class InterfaceCSVForm(forms.ModelForm):
super().__init__(*args, **kwargs)
# Limit LAG choices to interfaces belonging to this device (or VC master)
if self.is_bound:
if self.is_bound and 'device' in self.data:
try:
device = self.fields['device'].to_python(self.data['device'])
except forms.ValidationError:
@@ -4256,7 +4258,7 @@ class PowerPanelFilterForm(BootstrapMixin, CustomFieldFilterForm):
# Power feeds
#
class PowerFeedForm(BootstrapMixin, CustomFieldForm):
class PowerFeedForm(BootstrapMixin, CustomFieldModelForm):
site = ChainedModelChoiceField(
queryset=Site.objects.all(),
required=False,
@@ -4301,7 +4303,7 @@ class PowerFeedForm(BootstrapMixin, CustomFieldForm):
self.initial['site'] = self.instance.power_panel.site
class PowerFeedCSVForm(forms.ModelForm):
class PowerFeedCSVForm(CustomFieldModelCSVForm):
site = forms.ModelChoiceField(
queryset=Site.objects.all(),
to_field_name='name',
@@ -4384,7 +4386,7 @@ class PowerFeedBulkEditForm(BootstrapMixin, AddRemoveTagsForm, CustomFieldBulkEd
queryset=PowerFeed.objects.all(),
widget=forms.MultipleHiddenInput
)
powerpanel = forms.ModelChoiceField(
power_panel = forms.ModelChoiceField(
queryset=PowerPanel.objects.all(),
required=False,
widget=APISelect(

3
netbox/dcim/models/__init__.py
View File

@@ -1018,9 +1018,6 @@ class DeviceType(ChangeLoggedModel, CustomFieldModel):
tags = TaggableManager(through=TaggedItem)
csv_headers = [
'manufacturer', 'model', 'slug', 'part_number', 'u_height', 'is_full_depth', 'subdevice_role', 'comments',
]
clone_fields = [
'manufacturer', 'u_height', 'is_full_depth', 'subdevice_role',
]

2
netbox/dcim/models/device_components.py
View File

@@ -676,7 +676,7 @@ class Interface(CableTermination, ComponentModel):
self.untagged_vlan = None
# Only "tagged" interfaces may have tagged VLANs assigned. ("tagged all" implies all VLANs are assigned.)
if self.pk and self.mode is not InterfaceModeChoices.MODE_TAGGED:
if self.pk and self.mode != InterfaceModeChoices.MODE_TAGGED:
self.tagged_vlans.clear()
return super().save(*args, **kwargs)

1112
netbox/dcim/tests/test_views.py
View File

File diff suppressed because it is too large Load Diff

4
netbox/dcim/urls.py
View File

@@ -60,7 +60,7 @@ urlpatterns = [
# Racks
path(r'racks/', views.RackListView.as_view(), name='rack_list'),
path(r'rack-elevations/', views.RackElevationListView.as_view(), name='rack_elevation_list'),
path(r'racks/add/', views.RackEditView.as_view(), name='rack_add'),
path(r'racks/add/', views.RackCreateView.as_view(), name='rack_add'),
path(r'racks/import/', views.RackBulkImportView.as_view(), name='rack_import'),
path(r'racks/edit/', views.RackBulkEditView.as_view(), name='rack_bulk_edit'),
path(r'racks/delete/', views.RackBulkDeleteView.as_view(), name='rack_bulk_delete'),
@@ -318,7 +318,7 @@ urlpatterns = [
# Power feeds
path(r'power-feeds/', views.PowerFeedListView.as_view(), name='powerfeed_list'),
path(r'power-feeds/add/', views.PowerFeedEditView.as_view(), name='powerfeed_add'),
path(r'power-feeds/add/', views.PowerFeedCreateView.as_view(), name='powerfeed_add'),
path(r'power-feeds/import/', views.PowerFeedBulkImportView.as_view(), name='powerfeed_import'),
path(r'power-feeds/edit/', views.PowerFeedBulkEditView.as_view(), name='powerfeed_bulk_edit'),
path(r'power-feeds/delete/', views.PowerFeedBulkDeleteView.as_view(), name='powerfeed_bulk_delete'),

16
netbox/extras/api/serializers.py
View File

@@ -20,6 +20,8 @@ from utilities.api import (
ChoiceField, ContentTypeField, get_serializer_for_model, SerializerNotFound, SerializedPKRelatedField,
ValidatedModelSerializer,
)
from virtualization.api.nested_serializers import NestedClusterGroupSerializer, NestedClusterSerializer
from virtualization.models import Cluster, ClusterGroup
from .nested_serializers import *
@@ -161,6 +163,18 @@ class ConfigContextSerializer(ValidatedModelSerializer):
required=False,
many=True
)
cluster_groups = SerializedPKRelatedField(
queryset=ClusterGroup.objects.all(),
serializer=NestedClusterGroupSerializer,
required=False,
many=True
)
clusters = SerializedPKRelatedField(
queryset=Cluster.objects.all(),
serializer=NestedClusterSerializer,
required=False,
many=True
)
tenant_groups = SerializedPKRelatedField(
queryset=TenantGroup.objects.all(),
serializer=NestedTenantGroupSerializer,
@@ -184,7 +198,7 @@ class ConfigContextSerializer(ValidatedModelSerializer):
model = ConfigContext
fields = [
'id', 'name', 'weight', 'description', 'is_active', 'regions', 'sites', 'roles', 'platforms',
'tenant_groups', 'tenants', 'tags', 'data',
'cluster_groups', 'clusters', 'tenant_groups', 'tenants', 'tags', 'data',
]

17
netbox/extras/filters.py
View File

@@ -4,6 +4,7 @@ from django.db.models import Q
from dcim.models import DeviceRole, Platform, Region, Site
from tenancy.models import Tenant, TenantGroup
from virtualization.models import Cluster, ClusterGroup
from .choices import *
from .models import ConfigContext, CustomField, Graph, ExportTemplate, ObjectChange, Tag
@@ -170,6 +171,22 @@ class ConfigContextFilterSet(django_filters.FilterSet):
to_field_name='slug',
label='Platform (slug)',
)
cluster_group_id = django_filters.ModelMultipleChoiceFilter(
field_name='cluster_groups',
queryset=ClusterGroup.objects.all(),
label='Cluster group',
)
cluster_group = django_filters.ModelMultipleChoiceFilter(
field_name='cluster_groups__slug',
queryset=ClusterGroup.objects.all(),
to_field_name='slug',
label='Cluster group (slug)',
)
cluster_id = django_filters.ModelMultipleChoiceFilter(
field_name='clusters',
queryset=Cluster.objects.all(),
label='Cluster',
)
tenant_group_id = django_filters.ModelMultipleChoiceFilter(
field_name='tenant_groups',
queryset=TenantGroup.objects.all(),

177
netbox/extras/forms.py
View File

@@ -1,18 +1,16 @@
from collections import OrderedDict
from django import forms
from django.contrib.auth.models import User
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from taggit.forms import TagField
from dcim.models import DeviceRole, Platform, Region, Site
from tenancy.models import Tenant, TenantGroup
from utilities.forms import (
add_blank_choice, APISelectMultiple, BootstrapMixin, BulkEditForm, BulkEditNullBooleanSelect, ColorSelect,
CommentField, ContentTypeSelect, DatePicker, DateTimePicker, FilterChoiceField, LaxURLField, JSONField,
SlugField, StaticSelect2, BOOLEAN_WITH_BLANK_CHOICES,
CommentField, ContentTypeSelect, DateTimePicker, FilterChoiceField, JSONField, SlugField, StaticSelect2,
BOOLEAN_WITH_BLANK_CHOICES,
)
from virtualization.models import Cluster, ClusterGroup
from .choices import *
from .models import ConfigContext, CustomField, CustomFieldValue, ImageAttachment, ObjectChange, Tag
@@ -21,102 +19,41 @@ from .models import ConfigContext, CustomField, CustomFieldValue, ImageAttachmen
# Custom fields
#
def get_custom_fields_for_model(content_type, filterable_only=False, bulk_edit=False):
"""
Retrieve all CustomFields applicable to the given ContentType
"""
field_dict = OrderedDict()
custom_fields = CustomField.objects.filter(obj_type=content_type)
if filterable_only:
custom_fields = custom_fields.exclude(filter_logic=CustomFieldFilterLogicChoices.FILTER_DISABLED)
for cf in custom_fields:
field_name = 'cf_{}'.format(str(cf.name))
initial = cf.default if not bulk_edit else None
# Integer
if cf.type == CustomFieldTypeChoices.TYPE_INTEGER:
field = forms.IntegerField(required=cf.required, initial=initial)
# Boolean
elif cf.type == CustomFieldTypeChoices.TYPE_BOOLEAN:
choices = (
(None, '---------'),
(1, 'True'),
(0, 'False'),
)
if initial is not None and initial.lower() in ['true', 'yes', '1']:
initial = 1
elif initial is not None and initial.lower() in ['false', 'no', '0']:
initial = 0
else:
initial = None
field = forms.NullBooleanField(
required=cf.required, initial=initial, widget=StaticSelect2(choices=choices)
)
# Date
elif cf.type == CustomFieldTypeChoices.TYPE_DATE:
field = forms.DateField(required=cf.required, initial=initial, widget=DatePicker())
# Select
elif cf.type == CustomFieldTypeChoices.TYPE_SELECT:
choices = [(cfc.pk, cfc) for cfc in cf.choices.all()]
if not cf.required or bulk_edit or filterable_only:
choices = [(None, '---------')] + choices
# Check for a default choice
default_choice = None
if initial:
try:
default_choice = cf.choices.get(value=initial).pk
except ObjectDoesNotExist:
pass
field = forms.TypedChoiceField(
choices=choices, coerce=int, required=cf.required, initial=default_choice, widget=StaticSelect2()
)
# URL
elif cf.type == CustomFieldTypeChoices.TYPE_URL:
field = LaxURLField(required=cf.required, initial=initial)
# Text
else:
field = forms.CharField(max_length=255, required=cf.required, initial=initial)
field.model = cf
field.label = cf.label if cf.label else cf.name.replace('_', ' ').capitalize()
if cf.description:
field.help_text = cf.description
field_dict[field_name] = field
return field_dict
class CustomFieldForm(forms.ModelForm):
class CustomFieldModelForm(forms.ModelForm):
def __init__(self, *args, **kwargs):
self.custom_fields = []
self.obj_type = ContentType.objects.get_for_model(self._meta.model)
self.custom_fields = []
self.custom_field_values = {}
super().__init__(*args, **kwargs)
# Add all applicable CustomFields to the form
custom_fields = []
for name, field in get_custom_fields_for_model(self.obj_type).items():
self.fields[name] = field
custom_fields.append(name)
self.custom_fields = custom_fields
self._append_customfield_fields()
# If editing an existing object, initialize values for all custom fields
def _append_customfield_fields(self):
"""
Append form fields for all CustomFields assigned to this model.
"""
# Retrieve initial CustomField values for the instance
if self.instance.pk:
existing_values = CustomFieldValue.objects.filter(
for cfv in CustomFieldValue.objects.filter(
obj_type=self.obj_type,
obj_id=self.instance.pk
).prefetch_related('field')
for cfv in existing_values:
self.initial['cf_{}'.format(str(cfv.field.name))] = cfv.serialized_value
).prefetch_related('field'):
self.custom_field_values[cfv.field.name] = cfv.serialized_value
# Append form fields; assign initial values if modifying and existing object
for cf in CustomField.objects.filter(obj_type=self.obj_type):
field_name = 'cf_{}'.format(cf.name)
if self.instance.pk:
self.fields[field_name] = cf.to_form_field(set_initial=False)
self.fields[field_name].initial = self.custom_field_values.get(cf.name)
else:
self.fields[field_name] = cf.to_form_field()
# Annotate the field in the list of CustomField form fields
self.custom_fields.append(field_name)
def _save_custom_fields(self):
@@ -151,6 +88,19 @@ class CustomFieldForm(forms.ModelForm):
return obj
class CustomFieldModelCSVForm(CustomFieldModelForm):
def _append_customfield_fields(self):
# Append form fields
for cf in CustomField.objects.filter(obj_type=self.obj_type):
field_name = 'cf_{}'.format(cf.name)
self.fields[field_name] = cf.to_form_field(for_csv_import=True)
# Annotate the field in the list of CustomField form fields
self.custom_fields.append(field_name)
class CustomFieldBulkEditForm(BulkEditForm):
def __init__(self, *args, **kwargs):
@@ -160,15 +110,14 @@ class CustomFieldBulkEditForm(BulkEditForm):
self.obj_type = ContentType.objects.get_for_model(self.model)
# Add all applicable CustomFields to the form
custom_fields = get_custom_fields_for_model(self.obj_type, bulk_edit=True).items()
for name, field in custom_fields:
custom_fields = CustomField.objects.filter(obj_type=self.obj_type)
for cf in custom_fields:
# Annotate non-required custom fields as nullable
if not field.required:
self.nullable_fields.append(name)
field.required = False
self.fields[name] = field
if not cf.required:
self.nullable_fields.append(cf.name)
self.fields[cf.name] = cf.to_form_field(set_initial=False, enforce_required=False)
# Annotate this as a custom field
self.custom_fields.append(name)
self.custom_fields.append(cf.name)
class CustomFieldFilterForm(forms.Form):
@@ -180,10 +129,11 @@ class CustomFieldFilterForm(forms.Form):
super().__init__(*args, **kwargs)
# Add all applicable CustomFields to the form
custom_fields = get_custom_fields_for_model(self.obj_type, filterable_only=True).items()
for name, field in custom_fields:
field.required = False
self.fields[name] = field
custom_fields = CustomField.objects.filter(obj_type=self.obj_type).exclude(
filter_logic=CustomFieldFilterLogicChoices.FILTER_DISABLED
)
for cf in custom_fields:
self.fields[cf.name] = cf.to_form_field(set_initial=True, enforce_required=False)
#
@@ -254,8 +204,8 @@ class ConfigContextForm(BootstrapMixin, forms.ModelForm):
class Meta:
model = ConfigContext
fields = [
'name', 'weight', 'description', 'is_active', 'regions', 'sites', 'roles', 'platforms', 'tenant_groups',
'tenants', 'tags', 'data',
'name', 'weight', 'description', 'is_active', 'regions', 'sites', 'roles', 'platforms', 'cluster_groups',
'clusters', 'tenant_groups', 'tenants', 'tags', 'data',
]
widgets = {
'regions': APISelectMultiple(
@@ -270,6 +220,12 @@ class ConfigContextForm(BootstrapMixin, forms.ModelForm):
'platforms': APISelectMultiple(
api_url="/api/dcim/platforms/"
),
'cluster_groups': APISelectMultiple(
api_url="/api/virtualization/cluster-groups/"
),
'clusters': APISelectMultiple(
api_url="/api/virtualization/clusters/"
),
'tenant_groups': APISelectMultiple(
api_url="/api/tenancy/tenant-groups/"
),
@@ -340,6 +296,21 @@ class ConfigContextFilterForm(BootstrapMixin, forms.Form):
value_field="slug",
)
)
cluster_group = FilterChoiceField(
queryset=ClusterGroup.objects.all(),
to_field_name='slug',
widget=APISelectMultiple(
api_url="/api/virtualization/cluster-groups/",
value_field="slug",
)
)
cluster_id = FilterChoiceField(
queryset=Cluster.objects.all(),
label='Cluster',
widget=APISelectMultiple(
api_url="/api/virtualization/clusters/",
)
)
tenant_group = FilterChoiceField(
queryset=TenantGroup.objects.all(),
to_field_name='slug',

24
netbox/extras/migrations/0037_configcontexts_clusters.py Normal file
View File

@@ -0,0 +1,24 @@
# Generated by Django 2.2.8 on 2020-01-17 18:11
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('virtualization', '0013_deterministic_ordering'),
('extras', '0036_contenttype_filters_to_q_objects'),
]
operations = [
migrations.AddField(
model_name='configcontext',
name='cluster_groups',
field=models.ManyToManyField(blank=True, related_name='_configcontext_cluster_groups_+', to='virtualization.ClusterGroup'),
),
migrations.AddField(
model_name='configcontext',
name='clusters',
field=models.ManyToManyField(blank=True, related_name='_configcontext_clusters_+', to='virtualization.Cluster'),
),
]

81
netbox/extras/models.py
View File

@@ -1,6 +1,7 @@
from collections import OrderedDict
from datetime import date
from django import forms
from django.contrib.auth.models import User
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
@@ -14,6 +15,7 @@ from django.utils.text import slugify
from taggit.models import TagBase, GenericTaggedItemBase
from utilities.fields import ColorField
from utilities.forms import CSVChoiceField, DatePicker, LaxURLField, StaticSelect2, add_blank_choice
from utilities.utils import deepmerge, render_jinja2
from .choices import *
from .constants import *
@@ -280,6 +282,75 @@ class CustomField(models.Model):
return self.choices.get(pk=int(serialized_value))
return serialized_value
def to_form_field(self, set_initial=True, enforce_required=True, for_csv_import=False):
"""
Return a form field suitable for setting a CustomField's value for an object.
set_initial: Set initial date for the field. This should be False when generating a field for bulk editing.
enforce_required: Honor the value of CustomField.required. Set to False for filtering/bulk editing.
for_csv_import: Return a form field suitable for bulk import of objects in CSV format.
"""
initial = self.default if set_initial else None
required = self.required if enforce_required else False
# Integer
if self.type == CustomFieldTypeChoices.TYPE_INTEGER:
field = forms.IntegerField(required=required, initial=initial)
# Boolean
elif self.type == CustomFieldTypeChoices.TYPE_BOOLEAN:
choices = (
(None, '---------'),
(1, 'True'),
(0, 'False'),
)
if initial is not None and initial.lower() in ['true', 'yes', '1']:
initial = 1
elif initial is not None and initial.lower() in ['false', 'no', '0']:
initial = 0
else:
initial = None
field = forms.NullBooleanField(
required=required, initial=initial, widget=StaticSelect2(choices=choices)
)
# Date
elif self.type == CustomFieldTypeChoices.TYPE_DATE:
field = forms.DateField(required=required, initial=initial, widget=DatePicker())
# Select
elif self.type == CustomFieldTypeChoices.TYPE_SELECT:
choices = [(cfc.pk, cfc.value) for cfc in self.choices.all()]
if not required:
choices = add_blank_choice(choices)
# Set the initial value to the PK of the default choice, if any
if set_initial:
default_choice = self.choices.filter(value=self.default).first()
if default_choice:
initial = default_choice.pk
field_class = CSVChoiceField if for_csv_import else forms.ChoiceField
field = field_class(
choices=choices, required=required, initial=initial, widget=StaticSelect2()
)
# URL
elif self.type == CustomFieldTypeChoices.TYPE_URL:
field = LaxURLField(required=required, initial=initial)
# Text
else:
field = forms.CharField(max_length=255, required=required, initial=initial)
field.model = self
field.label = self.label if self.label else self.name.replace('_', ' ').capitalize()
if self.description:
field.help_text = self.description
return field
class CustomFieldValue(models.Model):
field = models.ForeignKey(
@@ -694,6 +765,16 @@ class ConfigContext(models.Model):
related_name='+',
blank=True
)
cluster_groups = models.ManyToManyField(
to='virtualization.ClusterGroup',
related_name='+',
blank=True
)
clusters = models.ManyToManyField(
to='virtualization.Cluster',
related_name='+',
blank=True
)
tenant_groups = models.ManyToManyField(
to='tenancy.TenantGroup',
related_name='+',

6
netbox/extras/querysets.py
View File

@@ -29,6 +29,10 @@ class ConfigContextQuerySet(QuerySet):
# `device_role` for Device; `role` for VirtualMachine
role = getattr(obj, 'device_role', None) or obj.role
# Virtualization cluster for VirtualMachine
cluster = getattr(obj, 'cluster', None)
cluster_group = getattr(cluster, 'group', None)
# Get the group of the assigned tenant, if any
tenant_group = obj.tenant.group if obj.tenant else None
@@ -44,6 +48,8 @@ class ConfigContextQuerySet(QuerySet):
Q(sites=obj.site) | Q(sites=None),
Q(roles=role) | Q(roles=None),
Q(platforms=obj.platform) | Q(platforms=None),
Q(cluster_groups=cluster_group) | Q(cluster_groups=None),
Q(clusters=cluster) | Q(clusters=None),
Q(tenant_groups=tenant_group) | Q(tenant_groups=None),
Q(tenants=obj.tenant) | Q(tenants=None),
Q(tags__slug__in=obj.tags.slugs()) | Q(tags=None),

115
netbox/extras/tests/test_customfields.py
View File

@@ -1,14 +1,15 @@
from datetime import date
from django.contrib.contenttypes.models import ContentType
from django.test import TestCase
from django.test import Client, TestCase
from django.urls import reverse
from rest_framework import status
from dcim.forms import SiteCSVForm
from dcim.models import Site
from extras.choices import *
from extras.models import CustomField, CustomFieldValue, CustomFieldChoice
from utilities.testing import APITestCase
from utilities.testing import APITestCase, create_test_user
from virtualization.models import VirtualMachine
@@ -364,3 +365,113 @@ class CustomFieldChoiceAPITest(APITestCase):
self.assertEqual(self.cf_choice_1.pk, response.data[self.cf_1.name][self.cf_choice_1.value])
self.assertEqual(self.cf_choice_2.pk, response.data[self.cf_1.name][self.cf_choice_2.value])
self.assertEqual(self.cf_choice_3.pk, response.data[self.cf_2.name][self.cf_choice_3.value])
class CustomFieldImportTest(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'dcim.view_site',
'dcim.add_site',
]
)
self.client = Client()
self.client.force_login(user)
@classmethod
def setUpTestData(cls):
custom_fields = (
CustomField(name='text', type=CustomFieldTypeChoices.TYPE_TEXT),
CustomField(name='integer', type=CustomFieldTypeChoices.TYPE_INTEGER),
CustomField(name='boolean', type=CustomFieldTypeChoices.TYPE_BOOLEAN),
CustomField(name='date', type=CustomFieldTypeChoices.TYPE_DATE),
CustomField(name='url', type=CustomFieldTypeChoices.TYPE_URL),
CustomField(name='select', type=CustomFieldTypeChoices.TYPE_SELECT),
)
for cf in custom_fields:
cf.save()
cf.obj_type.set([ContentType.objects.get_for_model(Site)])
CustomFieldChoice.objects.bulk_create((
CustomFieldChoice(field=custom_fields[5], value='Choice A'),
CustomFieldChoice(field=custom_fields[5], value='Choice B'),
CustomFieldChoice(field=custom_fields[5], value='Choice C'),
))
def test_import(self):
"""
Import a Site in CSV format, including a value for each CustomField.
"""
data = (
('name', 'slug', 'cf_text', 'cf_integer', 'cf_boolean', 'cf_date', 'cf_url', 'cf_select'),
('Site 1', 'site-1', 'ABC', '123', 'True', '2020-01-01', 'http://example.com/1', 'Choice A'),
('Site 2', 'site-2', 'DEF', '456', 'False', '2020-01-02', 'http://example.com/2', 'Choice B'),
('Site 3', 'site-3', '', '', '', '', '', ''),
)
csv_data = '\n'.join(','.join(row) for row in data)
response = self.client.post(reverse('dcim:site_import'), {'csv': csv_data})
self.assertEqual(response.status_code, 200)
# Validate data for site 1
custom_field_values = {
cf.name: value for cf, value in Site.objects.get(name='Site 1').get_custom_fields().items()
}
self.assertEqual(len(custom_field_values), 6)
self.assertEqual(custom_field_values['text'], 'ABC')
self.assertEqual(custom_field_values['integer'], 123)
self.assertEqual(custom_field_values['boolean'], True)
self.assertEqual(custom_field_values['date'], date(2020, 1, 1))
self.assertEqual(custom_field_values['url'], 'http://example.com/1')
self.assertEqual(custom_field_values['select'].value, 'Choice A')
# Validate data for site 2
custom_field_values = {
cf.name: value for cf, value in Site.objects.get(name='Site 2').get_custom_fields().items()
}
self.assertEqual(len(custom_field_values), 6)
self.assertEqual(custom_field_values['text'], 'DEF')
self.assertEqual(custom_field_values['integer'], 456)
self.assertEqual(custom_field_values['boolean'], False)
self.assertEqual(custom_field_values['date'], date(2020, 1, 2))
self.assertEqual(custom_field_values['url'], 'http://example.com/2')
self.assertEqual(custom_field_values['select'].value, 'Choice B')
# No CustomFieldValues should be created for site 3
obj_type = ContentType.objects.get_for_model(Site)
site3 = Site.objects.get(name='Site 3')
self.assertFalse(CustomFieldValue.objects.filter(obj_type=obj_type, obj_id=site3.pk).exists())
self.assertEqual(CustomFieldValue.objects.count(), 12) # Sanity check
def test_import_missing_required(self):
"""
Attempt to import an object missing a required custom field.
"""
# Set one of our CustomFields to required
CustomField.objects.filter(name='text').update(required=True)
form_data = {
'name': 'Site 1',
'slug': 'site-1',
}
form = SiteCSVForm(data=form_data)
self.assertFalse(form.is_valid())
self.assertIn('cf_text', form.errors)
def test_import_invalid_choice(self):
"""
Attempt to import an object with an invalid choice selection.
"""
form_data = {
'name': 'Site 1',
'slug': 'site-1',
'cf_select': 'Choice X'
}
form = SiteCSVForm(data=form_data)
self.assertFalse(form.is_valid())
self.assertIn('cf_select', form.errors)

30
netbox/extras/tests/test_filters.py
View File

@@ -7,6 +7,7 @@ from extras.constants import GRAPH_MODELS
from extras.filters import *
from extras.models import ConfigContext, ExportTemplate, Graph
from tenancy.models import Tenant, TenantGroup
from virtualization.models import Cluster, ClusterGroup, ClusterType
class GraphTestCase(TestCase):
@@ -107,6 +108,21 @@ class ConfigContextTestCase(TestCase):
)
Platform.objects.bulk_create(platforms)
cluster_groups = (
ClusterGroup(name='Cluster Group 1', slug='cluster-group-1'),
ClusterGroup(name='Cluster Group 2', slug='cluster-group-2'),
ClusterGroup(name='Cluster Group 3', slug='cluster-group-3'),
)
ClusterGroup.objects.bulk_create(cluster_groups)
cluster_type = ClusterType.objects.create(name='Cluster Type 1', slug='cluster-type-1')
clusters = (
Cluster(name='Cluster 1', type=cluster_type),
Cluster(name='Cluster 2', type=cluster_type),
Cluster(name='Cluster 3', type=cluster_type),
)
Cluster.objects.bulk_create(clusters)
tenant_groups = (
TenantGroup(name='Tenant Group 1', slug='tenant-group-1'),
TenantGroup(name='Tenant Group 2', slug='tenant-group-2'),
@@ -132,6 +148,8 @@ class ConfigContextTestCase(TestCase):
c.sites.set([sites[i]])
c.roles.set([device_roles[i]])
c.platforms.set([platforms[i]])
c.cluster_groups.set([cluster_groups[i]])
c.clusters.set([clusters[i]])
c.tenant_groups.set([tenant_groups[i]])
c.tenants.set([tenants[i]])
@@ -173,6 +191,18 @@ class ConfigContextTestCase(TestCase):
params = {'platform': [platforms[0].slug, platforms[1].slug]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_cluster_group(self):
cluster_groups = ClusterGroup.objects.all()[:2]
params = {'cluster_group_id': [cluster_groups[0].pk, cluster_groups[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'cluster_group': [cluster_groups[0].slug, cluster_groups[1].slug]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_cluster(self):
clusters = Cluster.objects.all()[:2]
params = {'cluster_id': [clusters[0].pk, clusters[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_tenant_group(self):
tenant_groups = TenantGroup.objects.all()[:2]
params = {'tenant_group_id': [tenant_groups[0].pk, tenant_groups[1].pk]}

102
netbox/extras/tests/test_views.py
View File

@@ -2,86 +2,102 @@ import urllib.parse
import uuid
from django.contrib.auth.models import User
from django.test import Client, TestCase
from django.urls import reverse
from dcim.models import Site
from extras.choices import ObjectChangeActionChoices
from extras.models import ConfigContext, ObjectChange, Tag
from utilities.testing import create_test_user
from utilities.testing import StandardTestCases, TestCase
class TagTestCase(TestCase):
class TagTestCase(StandardTestCases.Views):
model = Tag
def setUp(self):
user = create_test_user(permissions=['extras.view_tag'])
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_create_object = None
test_import_objects = None
Tag.objects.bulk_create([
@classmethod
def setUpTestData(cls):
Tag.objects.bulk_create((
Tag(name='Tag 1', slug='tag-1'),
Tag(name='Tag 2', slug='tag-2'),
Tag(name='Tag 3', slug='tag-3'),
])
))
def test_tag_list(self):
url = reverse('extras:tag_list')
params = {
"q": "tag",
cls.form_data = {
'name': 'Tag X',
'slug': 'tag-x',
'color': 'c0c0c0',
'comments': 'Some comments',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
cls.bulk_edit_data = {
'color': '00ff00',
}
class ConfigContextTestCase(TestCase):
class ConfigContextTestCase(StandardTestCases.Views):
model = ConfigContext
def setUp(self):
user = create_test_user(permissions=['extras.view_configcontext'])
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_import_objects = None
site = Site(name='Site 1', slug='site-1')
site.save()
# TODO: Resolve model discrepancies when creating/editing ConfigContexts
test_create_object = None
test_edit_object = None
@classmethod
def setUpTestData(cls):
site = Site.objects.create(name='Site 1', slug='site-1')
# Create three ConfigContexts
for i in range(1, 4):
configcontext = ConfigContext(
name='Config Context {}'.format(i),
data='{{"foo": {}}}'.format(i)
data={'foo': i}
)
configcontext.save()
configcontext.sites.add(site)
def test_configcontext_list(self):
url = reverse('extras:configcontext_list')
params = {
"q": "foo",
cls.form_data = {
'name': 'Config Context X',
'weight': 200,
'description': 'A new config context',
'is_active': True,
'regions': [],
'sites': [site.pk],
'roles': [],
'platforms': [],
'tenant_groups': [],
'tenants': [],
'tags': [],
'data': '{"foo": 123}',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_configcontext(self):
configcontext = ConfigContext.objects.first()
response = self.client.get(configcontext.get_absolute_url())
self.assertEqual(response.status_code, 200)
cls.bulk_edit_data = {
'weight': 300,
'is_active': False,
'description': 'New description',
}
# TODO: Convert to StandardTestCases.Views
class ObjectChangeTestCase(TestCase):
user_permissions = (
'extras.view_objectchange',
)
def setUp(self):
user = create_test_user(permissions=['extras.view_objectchange'])
self.client = Client()
self.client.force_login(user)
@classmethod
def setUpTestData(cls):
site = Site(name='Site 1', slug='site-1')
site.save()
# Create three ObjectChanges
user = User.objects.create_user(username='testuser2')
for i in range(1, 4):
oc = site.to_objectchange(action=ObjectChangeActionChoices.ACTION_UPDATE)
oc.user = user
@@ -96,10 +112,10 @@ class ObjectChangeTestCase(TestCase):
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
self.assertHttpStatus(response, 200)
def test_objectchange(self):
objectchange = ObjectChange.objects.first()
response = self.client.get(objectchange.get_absolute_url())
self.assertEqual(response.status_code, 200)
self.assertHttpStatus(response, 200)

6
netbox/extras/views.py
View File

@@ -37,7 +37,8 @@ class TagListView(PermissionRequiredMixin, ObjectListView):
template_name = 'extras/tag_list.html'
class TagView(View):
class TagView(PermissionRequiredMixin, View):
permission_required = 'extras.view_tag'
def get(self, request, slug):
@@ -84,10 +85,9 @@ class TagBulkEditView(PermissionRequiredMixin, BulkEditView):
).order_by(
'name'
)
# filter = filters.ProviderFilter
table = TagTable
form = forms.TagBulkEditForm
default_return_url = 'circuits:provider_list'
default_return_url = 'extras:tag_list'
class TagBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):

30
netbox/ipam/forms.py
View File

@@ -4,7 +4,9 @@ from django.core.validators import MaxValueValidator, MinValueValidator
from taggit.forms import TagField
from dcim.models import Device, Interface, Rack, Region, Site
from extras.forms import AddRemoveTagsForm, CustomFieldForm, CustomFieldBulkEditForm, CustomFieldFilterForm
from extras.forms import (
AddRemoveTagsForm, CustomFieldBulkEditForm, CustomFieldModelCSVForm, CustomFieldModelForm, CustomFieldFilterForm,
)
from tenancy.forms import TenancyFilterForm, TenancyForm
from tenancy.models import Tenant
from utilities.forms import (
@@ -31,7 +33,7 @@ IPADDRESS_MASK_LENGTH_CHOICES = add_blank_choice([
# VRFs
#
class VRFForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class VRFForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
tags = TagField(
required=False
)
@@ -49,7 +51,7 @@ class VRFForm(BootstrapMixin, TenancyForm, CustomFieldForm):
}
class VRFCSVForm(forms.ModelForm):
class VRFCSVForm(CustomFieldModelCSVForm):
tenant = forms.ModelChoiceField(
queryset=Tenant.objects.all(),
required=False,
@@ -145,7 +147,7 @@ class RIRFilterForm(BootstrapMixin, forms.Form):
# Aggregates
#
class AggregateForm(BootstrapMixin, CustomFieldForm):
class AggregateForm(BootstrapMixin, CustomFieldModelForm):
tags = TagField(
required=False
)
@@ -167,7 +169,7 @@ class AggregateForm(BootstrapMixin, CustomFieldForm):
}
class AggregateCSVForm(forms.ModelForm):
class AggregateCSVForm(CustomFieldModelCSVForm):
rir = forms.ModelChoiceField(
queryset=RIR.objects.all(),
to_field_name='name',
@@ -265,7 +267,7 @@ class RoleCSVForm(forms.ModelForm):
# Prefixes
#
class PrefixForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class PrefixForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
site = forms.ModelChoiceField(
queryset=Site.objects.all(),
required=False,
@@ -343,7 +345,7 @@ class PrefixForm(BootstrapMixin, TenancyForm, CustomFieldForm):
self.fields['vrf'].empty_label = 'Global'
class PrefixCSVForm(forms.ModelForm):
class PrefixCSVForm(CustomFieldModelCSVForm):
vrf = FlexibleModelChoiceField(
queryset=VRF.objects.all(),
to_field_name='rd',
@@ -587,7 +589,7 @@ class PrefixFilterForm(BootstrapMixin, TenancyFilterForm, CustomFieldFilterForm)
# IP addresses
#
class IPAddressForm(BootstrapMixin, TenancyForm, ReturnURLForm, CustomFieldForm):
class IPAddressForm(BootstrapMixin, TenancyForm, ReturnURLForm, CustomFieldModelForm):
interface = forms.ModelChoiceField(
queryset=Interface.objects.all(),
required=False
@@ -754,7 +756,7 @@ class IPAddressBulkCreateForm(BootstrapMixin, forms.Form):
)
class IPAddressBulkAddForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class IPAddressBulkAddForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
class Meta:
model = IPAddress
@@ -774,7 +776,7 @@ class IPAddressBulkAddForm(BootstrapMixin, TenancyForm, CustomFieldForm):
self.fields['vrf'].empty_label = 'Global'
class IPAddressCSVForm(forms.ModelForm):
class IPAddressCSVForm(CustomFieldModelCSVForm):
vrf = FlexibleModelChoiceField(
queryset=VRF.objects.all(),
to_field_name='rd',
@@ -1091,7 +1093,7 @@ class VLANGroupFilterForm(BootstrapMixin, forms.Form):
# VLANs
#
class VLANForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class VLANForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
site = forms.ModelChoiceField(
queryset=Site.objects.all(),
required=False,
@@ -1139,7 +1141,7 @@ class VLANForm(BootstrapMixin, TenancyForm, CustomFieldForm):
}
class VLANCSVForm(forms.ModelForm):
class VLANCSVForm(CustomFieldModelCSVForm):
site = forms.ModelChoiceField(
queryset=Site.objects.all(),
required=False,
@@ -1315,7 +1317,7 @@ class VLANFilterForm(BootstrapMixin, TenancyFilterForm, CustomFieldFilterForm):
# Services
#
class ServiceForm(BootstrapMixin, CustomFieldForm):
class ServiceForm(BootstrapMixin, CustomFieldModelForm):
port = forms.IntegerField(
min_value=SERVICE_PORT_MIN,
max_value=SERVICE_PORT_MAX
@@ -1396,5 +1398,5 @@ class ServiceBulkEditForm(BootstrapMixin, CustomFieldBulkEditForm):
class Meta:
nullable_fields = [
'site', 'tenant', 'role', 'description',
'description',
]

522
netbox/ipam/tests/test_views.py
View File

@@ -1,26 +1,18 @@
from netaddr import IPNetwork
import urllib.parse
import datetime
from django.test import Client, TestCase
from django.urls import reverse
from netaddr import IPNetwork
from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site
from ipam.choices import ServiceProtocolChoices
from ipam.choices import *
from ipam.models import Aggregate, IPAddress, Prefix, RIR, Role, Service, VLAN, VLANGroup, VRF
from utilities.testing import create_test_user
from utilities.testing import StandardTestCases
class VRFTestCase(TestCase):
class VRFTestCase(StandardTestCases.Views):
model = VRF
def setUp(self):
user = create_test_user(
permissions=[
'ipam.view_vrf',
'ipam.add_vrf',
]
)
self.client = Client()
self.client.force_login(user)
@classmethod
def setUpTestData(cls):
VRF.objects.bulk_create([
VRF(name='VRF 1', rd='65000:1'),
@@ -28,48 +20,39 @@ class VRFTestCase(TestCase):
VRF(name='VRF 3', rd='65000:3'),
])
def test_vrf_list(self):
url = reverse('ipam:vrf_list')
params = {
"q": "65000",
cls.form_data = {
'name': 'VRF X',
'rd': '65000:999',
'tenant': None,
'enforce_unique': True,
'description': 'A new VRF',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_vrf(self):
vrf = VRF.objects.first()
response = self.client.get(vrf.get_absolute_url())
self.assertEqual(response.status_code, 200)
def test_vrf_import(self):
csv_data = (
cls.csv_data = (
"name",
"VRF 4",
"VRF 5",
"VRF 6",
)
response = self.client.post(reverse('ipam:vrf_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(VRF.objects.count(), 6)
cls.bulk_edit_data = {
'tenant': None,
'enforce_unique': False,
'description': 'New description',
}
class RIRTestCase(TestCase):
class RIRTestCase(StandardTestCases.Views):
model = RIR
def setUp(self):
user = create_test_user(
permissions=[
'ipam.view_rir',
'ipam.add_rir',
]
)
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_get_object = None
test_delete_object = None
test_bulk_edit_objects = None
@classmethod
def setUpTestData(cls):
RIR.objects.bulk_create([
RIR(name='RIR 1', slug='rir-1'),
@@ -77,91 +60,71 @@ class RIRTestCase(TestCase):
RIR(name='RIR 3', slug='rir-3'),
])
def test_rir_list(self):
cls.form_data = {
'name': 'RIR X',
'slug': 'rir-x',
'is_private': True,
}
url = reverse('ipam:rir_list')
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
def test_rir_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"RIR 4,rir-4",
"RIR 5,rir-5",
"RIR 6,rir-6",
)
response = self.client.post(reverse('ipam:rir_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(RIR.objects.count(), 6)
class AggregateTestCase(StandardTestCases.Views):
model = Aggregate
@classmethod
def setUpTestData(cls):
class AggregateTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'ipam.view_aggregate',
'ipam.add_aggregate',
]
rirs = (
RIR(name='RIR 1', slug='rir-1'),
RIR(name='RIR 2', slug='rir-2'),
)
self.client = Client()
self.client.force_login(user)
rir = RIR(name='RIR 1', slug='rir-1')
rir.save()
RIR.objects.bulk_create(rirs)
Aggregate.objects.bulk_create([
Aggregate(family=4, prefix=IPNetwork('10.1.0.0/16'), rir=rir),
Aggregate(family=4, prefix=IPNetwork('10.2.0.0/16'), rir=rir),
Aggregate(family=4, prefix=IPNetwork('10.3.0.0/16'), rir=rir),
Aggregate(family=4, prefix=IPNetwork('10.1.0.0/16'), rir=rirs[0]),
Aggregate(family=4, prefix=IPNetwork('10.2.0.0/16'), rir=rirs[0]),
Aggregate(family=4, prefix=IPNetwork('10.3.0.0/16'), rir=rirs[0]),
])
def test_aggregate_list(self):
url = reverse('ipam:aggregate_list')
params = {
"rir": RIR.objects.first().slug,
cls.form_data = {
'family': 4,
'prefix': IPNetwork('10.99.0.0/16'),
'rir': rirs[1].pk,
'date_added': datetime.date(2020, 1, 1),
'description': 'A new aggregate',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_aggregate(self):
aggregate = Aggregate.objects.first()
response = self.client.get(aggregate.get_absolute_url())
self.assertEqual(response.status_code, 200)
def test_aggregate_import(self):
csv_data = (
cls.csv_data = (
"prefix,rir",
"10.4.0.0/16,RIR 1",
"10.5.0.0/16,RIR 1",
"10.6.0.0/16,RIR 1",
)
response = self.client.post(reverse('ipam:aggregate_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(Aggregate.objects.count(), 6)
cls.bulk_edit_data = {
'rir': rirs[1].pk,
'date_added': datetime.date(2020, 1, 1),
'description': 'New description',
}
class RoleTestCase(TestCase):
class RoleTestCase(StandardTestCases.Views):
model = Role
def setUp(self):
user = create_test_user(
permissions=[
'ipam.view_role',
'ipam.add_role',
]
)
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_get_object = None
test_delete_object = None
test_bulk_edit_objects = None
@classmethod
def setUpTestData(cls):
Role.objects.bulk_create([
Role(name='Role 1', slug='role-1'),
@@ -169,146 +132,140 @@ class RoleTestCase(TestCase):
Role(name='Role 3', slug='role-3'),
])
def test_role_list(self):
cls.form_data = {
'name': 'Role X',
'slug': 'role-x',
'weight': 200,
'description': 'A new role',
}
url = reverse('ipam:role_list')
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
def test_role_import(self):
csv_data = (
cls.csv_data = (
"name,slug,weight",
"Role 4,role-4,1000",
"Role 5,role-5,1000",
"Role 6,role-6,1000",
)
response = self.client.post(reverse('ipam:role_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(Role.objects.count(), 6)
class PrefixTestCase(StandardTestCases.Views):
model = Prefix
@classmethod
def setUpTestData(cls):
class PrefixTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'ipam.view_prefix',
'ipam.add_prefix',
]
sites = (
Site(name='Site 1', slug='site-1'),
Site(name='Site 2', slug='site-2'),
)
self.client = Client()
self.client.force_login(user)
Site.objects.bulk_create(sites)
site = Site(name='Site 1', slug='site-1')
site.save()
vrfs = (
VRF(name='VRF 1', rd='65000:1'),
VRF(name='VRF 2', rd='65000:2'),
)
VRF.objects.bulk_create(vrfs)
roles = (
Role(name='Role 1', slug='role-1'),
Role(name='Role 2', slug='role-2'),
)
Prefix.objects.bulk_create([
Prefix(family=4, prefix=IPNetwork('10.1.0.0/16'), site=site),
Prefix(family=4, prefix=IPNetwork('10.2.0.0/16'), site=site),
Prefix(family=4, prefix=IPNetwork('10.3.0.0/16'), site=site),
Prefix(family=4, prefix=IPNetwork('10.1.0.0/16'), vrf=vrfs[0], site=sites[0], role=roles[0]),
Prefix(family=4, prefix=IPNetwork('10.2.0.0/16'), vrf=vrfs[0], site=sites[0], role=roles[0]),
Prefix(family=4, prefix=IPNetwork('10.3.0.0/16'), vrf=vrfs[0], site=sites[0], role=roles[0]),
])
def test_prefix_list(self):
url = reverse('ipam:prefix_list')
params = {
"site": Site.objects.first().slug,
cls.form_data = {
'prefix': IPNetwork('192.0.2.0/24'),
'site': sites[1].pk,
'vrf': vrfs[1].pk,
'tenant': None,
'vlan': None,
'status': PrefixStatusChoices.STATUS_RESERVED,
'role': roles[1].pk,
'is_pool': True,
'description': 'A new prefix',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_prefix(self):
prefix = Prefix.objects.first()
response = self.client.get(prefix.get_absolute_url())
self.assertEqual(response.status_code, 200)
def test_prefix_import(self):
csv_data = (
cls.csv_data = (
"prefix,status",
"10.4.0.0/16,Active",
"10.5.0.0/16,Active",
"10.6.0.0/16,Active",
)
response = self.client.post(reverse('ipam:prefix_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(Prefix.objects.count(), 6)
class IPAddressTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'ipam.view_ipaddress',
'ipam.add_ipaddress',
]
)
self.client = Client()
self.client.force_login(user)
vrf = VRF(name='VRF 1', rd='65000:1')
vrf.save()
IPAddress.objects.bulk_create([
IPAddress(family=4, address=IPNetwork('192.0.2.1/24'), vrf=vrf),
IPAddress(family=4, address=IPNetwork('192.0.2.2/24'), vrf=vrf),
IPAddress(family=4, address=IPNetwork('192.0.2.3/24'), vrf=vrf),
])
def test_ipaddress_list(self):
url = reverse('ipam:ipaddress_list')
params = {
"vrf": VRF.objects.first().rd,
cls.bulk_edit_data = {
'site': sites[1].pk,
'vrf': vrfs[1].pk,
'tenant': None,
'status': PrefixStatusChoices.STATUS_RESERVED,
'role': roles[1].pk,
'is_pool': False,
'description': 'New description',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_ipaddress(self):
class IPAddressTestCase(StandardTestCases.Views):
model = IPAddress
ipaddress = IPAddress.objects.first()
response = self.client.get(ipaddress.get_absolute_url())
self.assertEqual(response.status_code, 200)
@classmethod
def setUpTestData(cls):
def test_ipaddress_import(self):
vrfs = (
VRF(name='VRF 1', rd='65000:1'),
VRF(name='VRF 2', rd='65000:2'),
)
csv_data = (
IPAddress.objects.bulk_create([
IPAddress(family=4, address=IPNetwork('192.0.2.1/24'), vrf=vrfs[0]),
IPAddress(family=4, address=IPNetwork('192.0.2.2/24'), vrf=vrfs[0]),
IPAddress(family=4, address=IPNetwork('192.0.2.3/24'), vrf=vrfs[0]),
])
cls.form_data = {
'vrf': vrfs[1].pk,
'address': IPNetwork('192.0.2.99/24'),
'tenant': None,
'status': IPAddressStatusChoices.STATUS_RESERVED,
'role': IPAddressRoleChoices.ROLE_ANYCAST,
'interface': None,
'nat_inside': None,
'dns_name': 'example',
'description': 'A new IP address',
'tags': 'Alpha,Bravo,Charlie',
}
cls.csv_data = (
"address,status",
"192.0.2.4/24,Active",
"192.0.2.5/24,Active",
"192.0.2.6/24,Active",
)
response = self.client.post(reverse('ipam:ipaddress_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(IPAddress.objects.count(), 6)
cls.bulk_edit_data = {
'vrf': vrfs[1].pk,
'tenant': None,
'status': IPAddressStatusChoices.STATUS_RESERVED,
'role': IPAddressRoleChoices.ROLE_ANYCAST,
'dns_name': 'example',
'description': 'New description',
}
class VLANGroupTestCase(TestCase):
class VLANGroupTestCase(StandardTestCases.Views):
model = VLANGroup
def setUp(self):
user = create_test_user(
permissions=[
'ipam.view_vlangroup',
'ipam.add_vlangroup',
]
)
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_get_object = None
test_delete_object = None
test_bulk_edit_objects = None
site = Site(name='Site 1', slug='site-1')
site.save()
@classmethod
def setUpTestData(cls):
site = Site.objects.create(name='Site 1', slug='site-1')
VLANGroup.objects.bulk_create([
VLANGroup(name='VLAN Group 1', slug='vlan-group-1', site=site),
@@ -316,104 +273,96 @@ class VLANGroupTestCase(TestCase):
VLANGroup(name='VLAN Group 3', slug='vlan-group-3', site=site),
])
def test_vlangroup_list(self):
url = reverse('ipam:vlangroup_list')
params = {
"site": Site.objects.first().slug,
cls.form_data = {
'name': 'VLAN Group X',
'slug': 'vlan-group-x',
'site': site.pk,
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_vlangroup_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"VLAN Group 4,vlan-group-4",
"VLAN Group 5,vlan-group-5",
"VLAN Group 6,vlan-group-6",
)
response = self.client.post(reverse('ipam:vlangroup_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(VLANGroup.objects.count(), 6)
class VLANTestCase(StandardTestCases.Views):
model = VLAN
@classmethod
def setUpTestData(cls):
class VLANTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'ipam.view_vlan',
'ipam.add_vlan',
]
sites = (
Site(name='Site 1', slug='site-1'),
Site(name='Site 2', slug='site-2'),
)
self.client = Client()
self.client.force_login(user)
Site.objects.bulk_create(sites)
vlangroup = VLANGroup(name='VLAN Group 1', slug='vlan-group-1')
vlangroup.save()
vlangroups = (
VLANGroup(name='VLAN Group 1', slug='vlan-group-1', site=sites[0]),
VLANGroup(name='VLAN Group 2', slug='vlan-group-2', site=sites[1]),
)
VLANGroup.objects.bulk_create(vlangroups)
roles = (
Role(name='Role 1', slug='role-1'),
Role(name='Role 2', slug='role-2'),
)
Role.objects.bulk_create(roles)
VLAN.objects.bulk_create([
VLAN(group=vlangroup, vid=101, name='VLAN101'),
VLAN(group=vlangroup, vid=102, name='VLAN102'),
VLAN(group=vlangroup, vid=103, name='VLAN103'),
VLAN(group=vlangroups[0], vid=101, name='VLAN101', site=sites[0], role=roles[0]),
VLAN(group=vlangroups[0], vid=102, name='VLAN102', site=sites[0], role=roles[0]),
VLAN(group=vlangroups[0], vid=103, name='VLAN103', site=sites[0], role=roles[0]),
])
def test_vlan_list(self):
url = reverse('ipam:vlan_list')
params = {
"group": VLANGroup.objects.first().slug,
cls.form_data = {
'site': sites[1].pk,
'group': vlangroups[1].pk,
'vid': 999,
'name': 'VLAN999',
'tenant': None,
'status': VLANStatusChoices.STATUS_RESERVED,
'role': roles[1].pk,
'description': 'A new VLAN',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_vlan(self):
vlan = VLAN.objects.first()
response = self.client.get(vlan.get_absolute_url())
self.assertEqual(response.status_code, 200)
def test_vlan_import(self):
csv_data = (
cls.csv_data = (
"vid,name,status",
"104,VLAN104,Active",
"105,VLAN105,Active",
"106,VLAN106,Active",
)
response = self.client.post(reverse('ipam:vlan_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(VLAN.objects.count(), 6)
cls.bulk_edit_data = {
'site': sites[1].pk,
'group': vlangroups[1].pk,
'tenant': None,
'status': VLANStatusChoices.STATUS_RESERVED,
'role': roles[1].pk,
'description': 'New description',
}
class ServiceTestCase(TestCase):
class ServiceTestCase(StandardTestCases.Views):
model = Service
def setUp(self):
user = create_test_user(permissions=['ipam.view_service'])
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_import_objects = None
site = Site(name='Site 1', slug='site-1')
site.save()
# TODO: Resolve URL for Service creation
test_create_object = None
manufacturer = Manufacturer(name='Manufacturer 1', slug='manufacturer-1')
manufacturer.save()
@classmethod
def setUpTestData(cls):
devicetype = DeviceType(manufacturer=manufacturer, model='Device Type 1')
devicetype.save()
devicerole = DeviceRole(name='Device Role 1', slug='device-role-1')
devicerole.save()
device = Device(name='Device 1', site=site, device_type=devicetype, device_role=devicerole)
device.save()
site = Site.objects.create(name='Site 1', slug='site-1')
manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1')
devicerole = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
device = Device.objects.create(name='Device 1', site=site, device_type=devicetype, device_role=devicerole)
Service.objects.bulk_create([
Service(device=device, name='Service 1', protocol=ServiceProtocolChoices.PROTOCOL_TCP, port=101),
@@ -421,18 +370,19 @@ class ServiceTestCase(TestCase):
Service(device=device, name='Service 3', protocol=ServiceProtocolChoices.PROTOCOL_TCP, port=103),
])
def test_service_list(self):
url = reverse('ipam:service_list')
params = {
"device_id": Device.objects.first(),
cls.form_data = {
'device': device.pk,
'virtual_machine': None,
'name': 'Service X',
'protocol': ServiceProtocolChoices.PROTOCOL_TCP,
'port': 999,
'ipaddresses': [],
'description': 'A new service',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_service(self):
service = Service.objects.first()
response = self.client.get(service.get_absolute_url())
self.assertEqual(response.status_code, 200)
cls.bulk_edit_data = {
'protocol': ServiceProtocolChoices.PROTOCOL_UDP,
'port': 888,
'description': 'New description',
}

1
netbox/netbox/settings.py
View File

@@ -74,6 +74,7 @@ CORS_ORIGIN_WHITELIST = getattr(configuration, 'CORS_ORIGIN_WHITELIST', [])
DATE_FORMAT = getattr(configuration, 'DATE_FORMAT', 'N j, Y')
DATETIME_FORMAT = getattr(configuration, 'DATETIME_FORMAT', 'N j, Y g:i a')
DEBUG = getattr(configuration, 'DEBUG', False)
DEVELOPER = getattr(configuration, 'DEVELOPER', False)
EMAIL = getattr(configuration, 'EMAIL', {})
ENFORCE_GLOBAL_UNIQUE = getattr(configuration, 'ENFORCE_GLOBAL_UNIQUE', False)
EXEMPT_VIEW_PERMISSIONS = getattr(configuration, 'EXEMPT_VIEW_PERMISSIONS', [])

6
netbox/netbox/tests/test_views.py
View File

@@ -1,6 +1,6 @@
import urllib.parse
from django.test import TestCase
from utilities.testing import TestCase
from django.urls import reverse
@@ -11,7 +11,7 @@ class HomeViewTestCase(TestCase):
url = reverse('home')
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertHttpStatus(response, 200)
def test_search(self):
@@ -21,4 +21,4 @@ class HomeViewTestCase(TestCase):
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
self.assertHttpStatus(response, 200)

11
netbox/project-static/js/configcontext.js Normal file
View File

@@ -0,0 +1,11 @@
$('.rendered-context-format').on('click', function() {
if (!$(this).hasClass('active')) {
// Update selection in the button group
$('span.rendered-context-format').removeClass('active');
$('span.rendered-context-format[data-format=' + $(this).data('format') + ']').addClass('active');
// Hide all rendered contexts and only show the selected one
$('div.rendered-context-data').hide();
$('div.rendered-context-data[data-format=' + $(this).data('format') + ']').show();
}
});

8
netbox/secrets/forms.py
View File

@@ -4,7 +4,9 @@ from django import forms
from taggit.forms import TagField
from dcim.models import Device
from extras.forms import AddRemoveTagsForm, CustomFieldBulkEditForm, CustomFieldFilterForm, CustomFieldForm
from extras.forms import (
AddRemoveTagsForm, CustomFieldBulkEditForm, CustomFieldFilterForm, CustomFieldModelForm, CustomFieldModelCSVForm,
)
from utilities.forms import (
APISelect, APISelectMultiple, BootstrapMixin, FilterChoiceField, FlexibleModelChoiceField, SlugField,
StaticSelect2Multiple, TagFilterField
@@ -68,7 +70,7 @@ class SecretRoleCSVForm(forms.ModelForm):
# Secrets
#
class SecretForm(BootstrapMixin, CustomFieldForm):
class SecretForm(BootstrapMixin, CustomFieldModelForm):
plaintext = forms.CharField(
max_length=SECRET_PLAINTEXT_MAX_LENGTH,
required=False,
@@ -116,7 +118,7 @@ class SecretForm(BootstrapMixin, CustomFieldForm):
})
class SecretCSVForm(forms.ModelForm):
class SecretCSVForm(CustomFieldModelCSVForm):
device = FlexibleModelChoiceField(
queryset=Device.objects.all(),
to_field_name='name',

149
netbox/secrets/tests/test_views.py
View File

@@ -1,26 +1,23 @@
import base64
import urllib.parse
from django.test import Client, TestCase
from django.urls import reverse
from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site
from secrets.models import Secret, SecretRole, SessionKey, UserKey
from utilities.testing import create_test_user
from utilities.testing import StandardTestCases
from .constants import PRIVATE_KEY, PUBLIC_KEY
class SecretRoleTestCase(TestCase):
class SecretRoleTestCase(StandardTestCases.Views):
model = SecretRole
def setUp(self):
user = create_test_user(
permissions=[
'secrets.view_secretrole',
'secrets.add_secretrole',
]
)
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_get_object = None
test_delete_object = None
test_bulk_edit_objects = None
@classmethod
def setUpTestData(cls):
SecretRole.objects.bulk_create([
SecretRole(name='Secret Role 1', slug='secret-role-1'),
@@ -28,89 +25,83 @@ class SecretRoleTestCase(TestCase):
SecretRole(name='Secret Role 3', slug='secret-role-3'),
])
def test_secretrole_list(self):
cls.form_data = {
'name': 'Secret Role X',
'slug': 'secret-role-x',
'description': 'A secret role',
'users': [],
'groups': [],
}
url = reverse('secrets:secretrole_list')
response = self.client.get(url, follow=True)
self.assertEqual(response.status_code, 200)
def test_secretrole_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"Secret Role 4,secret-role-4",
"Secret Role 5,secret-role-5",
"Secret Role 6,secret-role-6",
)
response = self.client.post(reverse('secrets:secretrole_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(SecretRole.objects.count(), 6)
class SecretTestCase(StandardTestCases.Views):
model = Secret
# Disable inapplicable tests
test_create_object = None
class SecretTestCase(TestCase):
# TODO: Check permissions enforcement on secrets.views.secret_edit
test_edit_object = None
@classmethod
def setUpTestData(cls):
site = Site.objects.create(name='Site 1', slug='site-1')
manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1')
devicerole = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
devices = (
Device(name='Device 1', site=site, device_type=devicetype, device_role=devicerole),
Device(name='Device 2', site=site, device_type=devicetype, device_role=devicerole),
Device(name='Device 3', site=site, device_type=devicetype, device_role=devicerole),
)
Device.objects.bulk_create(devices)
secretroles = (
SecretRole(name='Secret Role 1', slug='secret-role-1'),
SecretRole(name='Secret Role 2', slug='secret-role-2'),
)
SecretRole.objects.bulk_create(secretroles)
# Create one secret per device to allow bulk-editing of names (which must be unique per device/role)
Secret.objects.bulk_create((
Secret(device=devices[0], role=secretroles[0], name='Secret 1', ciphertext=b'1234567890'),
Secret(device=devices[1], role=secretroles[0], name='Secret 2', ciphertext=b'1234567890'),
Secret(device=devices[2], role=secretroles[0], name='Secret 3', ciphertext=b'1234567890'),
))
cls.form_data = {
'device': devices[1].pk,
'role': secretroles[1].pk,
'name': 'Secret X',
}
cls.bulk_edit_data = {
'role': secretroles[1].pk,
'name': 'New name',
}
def setUp(self):
user = create_test_user(
permissions=[
'secrets.view_secret',
'secrets.add_secret',
]
)
# Set up a master key
userkey = UserKey(user=user, public_key=PUBLIC_KEY)
super().setUp()
# Set up a master key for the test user
userkey = UserKey(user=self.user, public_key=PUBLIC_KEY)
userkey.save()
master_key = userkey.get_master_key(PRIVATE_KEY)
self.session_key = SessionKey(userkey=userkey)
self.session_key.save(master_key)
self.client = Client()
self.client.force_login(user)
site = Site(name='Site 1', slug='site-1')
site.save()
manufacturer = Manufacturer(name='Manufacturer 1', slug='manufacturer-1')
manufacturer.save()
devicetype = DeviceType(manufacturer=manufacturer, model='Device Type 1')
devicetype.save()
devicerole = DeviceRole(name='Device Role 1', slug='device-role-1')
devicerole.save()
device = Device(name='Device 1', site=site, device_type=devicetype, device_role=devicerole)
device.save()
secretrole = SecretRole(name='Secret Role 1', slug='secret-role-1')
secretrole.save()
Secret.objects.bulk_create([
Secret(device=device, role=secretrole, name='Secret 1', ciphertext=b'1234567890'),
Secret(device=device, role=secretrole, name='Secret 2', ciphertext=b'1234567890'),
Secret(device=device, role=secretrole, name='Secret 3', ciphertext=b'1234567890'),
])
def test_secret_list(self):
url = reverse('secrets:secret_list')
params = {
"role": SecretRole.objects.first().slug,
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)), follow=True)
self.assertEqual(response.status_code, 200)
def test_secret(self):
secret = Secret.objects.first()
response = self.client.get(secret.get_absolute_url(), follow=True)
self.assertEqual(response.status_code, 200)
def test_secret_import(self):
def test_import_objects(self):
self.add_permissions('secrets.add_secret')
csv_data = (
"device,role,name,plaintext",
@@ -125,5 +116,5 @@ class SecretTestCase(TestCase):
response = self.client.post(reverse('secrets:secret_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertHttpStatus(response, 200)
self.assertEqual(Secret.objects.count(), 6)

36
netbox/templates/extras/configcontext.html
View File

@@ -1,5 +1,6 @@
{% extends '_base.html' %}
{% load helpers %}
{% load static %}
{% block header %}
<div class="row noprint">
@@ -134,6 +135,34 @@
{% endif %}
</td>
</tr>
<tr>
<td>Cluster Groups</td>
<td>
{% if configcontext.cluster_groups.all %}
<ul>
{% for cluster_group in configcontext.cluster_groups.all %}
<li><a href="{{ cluster_group.get_absolute_url }}">{{ cluster_group }}</a></li>
{% endfor %}
</ul>
{% else %}
<span class="text-muted">None</span>
{% endif %}
</td>
</tr>
<tr>
<td>Clusters</td>
<td>
{% if configcontext.clusters.all %}
<ul>
{% for cluster in configcontext.clusters.all %}
<li><a href="{{ cluster.get_absolute_url }}">{{ cluster }}</a></li>
{% endfor %}
</ul>
{% else %}
<span class="text-muted">None</span>
{% endif %}
</td>
</tr>
<tr>
<td>Tenant Groups</td>
<td>
@@ -183,11 +212,16 @@
<div class="panel panel-default">
<div class="panel-heading">
<strong>Data</strong>
{% include 'extras/inc/configcontext_format.html' %}
</div>
<div class="panel-body">
<pre>{{ configcontext.data|render_json }}</pre>
{% include 'extras/inc/configcontext_data.html' with data=configcontext.data %}
</div>
</div>
</div>
</div>
{% endblock %}
{% block javascript %}
<script src="{% static 'js/configcontext.js' %}?v{{ settings.VERSION }}"></script>
{% endblock %}

2
netbox/templates/extras/configcontext_edit.html
View File

@@ -18,6 +18,8 @@
{% render_field form.sites %}
{% render_field form.roles %}
{% render_field form.platforms %}
{% render_field form.cluster_groups %}
{% render_field form.clusters %}
{% render_field form.tenant_groups %}
{% render_field form.tenants %}
{% render_field form.tags %}

8
netbox/templates/extras/inc/configcontext_data.html Normal file
View File

@@ -0,0 +1,8 @@
{% load helpers %}
<div class="rendered-context-data" data-format="json">
<pre>{{ data|render_json }}</pre>
</div>
<div class="rendered-context-data" data-format="yaml" style="display: none;">
<pre>{{ data|render_yaml }}</pre>
</div>

6
netbox/templates/extras/inc/configcontext_format.html Normal file
View File

@@ -0,0 +1,6 @@
<div class="pull-right">
<div class="btn-group btn-group-xs" role="group">
<span class="btn btn-default rendered-context-format active" data-format="json">JSON</span>
<span class="btn btn-default rendered-context-format" data-format="yaml">YAML</span>
</div>
</div>

12
netbox/templates/extras/object_configcontext.html
View File

@@ -1,5 +1,6 @@
{% extends base_template %}
{% load helpers %}
{% load static %}
{% block title %}{{ block.super }} - Config Context{% endblock %}
@@ -9,9 +10,10 @@
<div class="panel panel-default">
<div class="panel-heading">
<strong>Rendered Context</strong>
{% include 'extras/inc/configcontext_format.html' %}
</div>
<div class="panel-body">
<pre>{{ rendered_context|render_json }}</pre>
{% include 'extras/inc/configcontext_data.html' with data=rendered_context %}
</div>
</div>
</div>
@@ -22,7 +24,7 @@
</div>
<div class="panel-body">
{% if obj.local_context_data %}
<pre>{{ obj.local_context_data|render_json }}</pre>
{% include 'extras/inc/configcontext_data.html' with data=obj.local_context_data %}
{% else %}
<span class="text-muted">None</span>
{% endif %}
@@ -47,7 +49,7 @@
{% if context.description %}
<br /><small>{{ context.description }}</small>
{% endif %}
<pre>{{ context.data|render_json }}</pre>
{% include 'extras/inc/configcontext_data.html' with data=context.data %}
</div>
{% empty %}
<div class="panel-body">
@@ -58,3 +60,7 @@
</div>
</div>
{% endblock %}
{% block javascript %}
<script src="{% static 'js/configcontext.js' %}?v{{ settings.VERSION }}"></script>
{% endblock %}

2
netbox/templates/virtualization/virtualmachine.html
View File

@@ -265,7 +265,9 @@
<th>Name</th>
<th>LAG</th>
<th>Description</th>
<th>MTU</th>
<th>Mode</th>
<th>Cable</th>
<th colspan="2">Connection</th>
<th></th>
</tr>

8
netbox/tenancy/forms.py
View File

@@ -1,7 +1,9 @@
from django import forms
from taggit.forms import TagField
from extras.forms import AddRemoveTagsForm, CustomFieldForm, CustomFieldBulkEditForm, CustomFieldFilterForm
from extras.forms import (
AddRemoveTagsForm, CustomFieldModelForm, CustomFieldBulkEditForm, CustomFieldModelCSVForm, CustomFieldFilterForm,
)
from utilities.forms import (
APISelect, APISelectMultiple, BootstrapMixin, ChainedFieldsMixin, ChainedModelChoiceField, CommentField,
FilterChoiceField, SlugField, TagFilterField
@@ -38,7 +40,7 @@ class TenantGroupCSVForm(forms.ModelForm):
# Tenants
#
class TenantForm(BootstrapMixin, CustomFieldForm):
class TenantForm(BootstrapMixin, CustomFieldModelForm):
slug = SlugField()
comments = CommentField()
tags = TagField(
@@ -57,7 +59,7 @@ class TenantForm(BootstrapMixin, CustomFieldForm):
}
class TenantCSVForm(forms.ModelForm):
class TenantCSVForm(CustomFieldModelForm):
slug = SlugField()
group = forms.ModelChoiceField(
queryset=TenantGroup.objects.all(),

102
netbox/tenancy/tests/test_views.py
View File

@@ -1,23 +1,17 @@
import urllib.parse
from django.test import Client, TestCase
from django.urls import reverse
from tenancy.models import Tenant, TenantGroup
from utilities.testing import create_test_user
from utilities.testing import StandardTestCases
class TenantGroupTestCase(TestCase):
class TenantGroupTestCase(StandardTestCases.Views):
model = TenantGroup
def setUp(self):
user = create_test_user(
permissions=[
'tenancy.view_tenantgroup',
'tenancy.add_tenantgroup',
]
)
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_get_object = None
test_delete_object = None
test_bulk_edit_objects = None
@classmethod
def setUpTestData(cls):
TenantGroup.objects.bulk_create([
TenantGroup(name='Tenant Group 1', slug='tenant-group-1'),
@@ -25,75 +19,53 @@ class TenantGroupTestCase(TestCase):
TenantGroup(name='Tenant Group 3', slug='tenant-group-3'),
])
def test_tenantgroup_list(self):
cls.form_data = {
'name': 'Tenant Group X',
'slug': 'tenant-group-x',
}
url = reverse('tenancy:tenantgroup_list')
response = self.client.get(url, follow=True)
self.assertEqual(response.status_code, 200)
def test_tenantgroup_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"Tenant Group 4,tenant-group-4",
"Tenant Group 5,tenant-group-5",
"Tenant Group 6,tenant-group-6",
)
response = self.client.post(reverse('tenancy:tenantgroup_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(TenantGroup.objects.count(), 6)
class TenantTestCase(StandardTestCases.Views):
model = Tenant
@classmethod
def setUpTestData(cls):
class TenantTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'tenancy.view_tenant',
'tenancy.add_tenant',
]
tenantgroups = (
TenantGroup(name='Tenant Group 1', slug='tenant-group-1'),
TenantGroup(name='Tenant Group 2', slug='tenant-group-2'),
)
self.client = Client()
self.client.force_login(user)
tenantgroup = TenantGroup(name='Tenant Group 1', slug='tenant-group-1')
tenantgroup.save()
TenantGroup.objects.bulk_create(tenantgroups)
Tenant.objects.bulk_create([
Tenant(name='Tenant 1', slug='tenant-1', group=tenantgroup),
Tenant(name='Tenant 2', slug='tenant-2', group=tenantgroup),
Tenant(name='Tenant 3', slug='tenant-3', group=tenantgroup),
Tenant(name='Tenant 1', slug='tenant-1', group=tenantgroups[0]),
Tenant(name='Tenant 2', slug='tenant-2', group=tenantgroups[0]),
Tenant(name='Tenant 3', slug='tenant-3', group=tenantgroups[0]),
])
def test_tenant_list(self):
url = reverse('tenancy:tenant_list')
params = {
"group": TenantGroup.objects.first().slug,
cls.form_data = {
'name': 'Tenant X',
'slug': 'tenant-x',
'group': tenantgroups[1].pk,
'description': 'A new tenant',
'comments': 'Some comments',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)), follow=True)
self.assertEqual(response.status_code, 200)
def test_tenant(self):
tenant = Tenant.objects.first()
response = self.client.get(tenant.get_absolute_url(), follow=True)
self.assertEqual(response.status_code, 200)
def test_tenant_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"Tenant 4,tenant-4",
"Tenant 5,tenant-5",
"Tenant 6,tenant-6",
)
response = self.client.post(reverse('tenancy:tenant_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(Tenant.objects.count(), 6)
cls.bulk_edit_data = {
'group': tenantgroups[1].pk,
}

23
netbox/utilities/management/commands/makemigrations.py
View File

@@ -1,7 +1,28 @@
# noinspection PyUnresolvedReferences
from django.core.management.commands.makemigrations import Command
from django.conf import settings
from django.core.management.base import CommandError
from django.core.management.commands.makemigrations import Command as _Command
from django.db import models
from . import custom_deconstruct
models.Field.deconstruct = custom_deconstruct
class Command(_Command):
def handle(self, *args, **kwargs):
"""
This built-in management command enables the creation of new database schema migration files, which should
never be required by and ordinary user. We prevent this command from executing unless the configuration
indicates that the user is a developer (i.e. configuration.DEVELOPER == True).
"""
if not settings.DEVELOPER:
raise CommandError(
"This command is available for development purposes only. It will\n"
"NOT resolve any issues with missing or unapplied migrations. For assistance,\n"
"please post to the NetBox mailing list:\n"
" https://groups.google.com/forum/#!forum/netbox-discuss"
)
super().handle(*args, **kwargs)

9
netbox/utilities/templatetags/helpers.py
View File

@@ -1,6 +1,7 @@
import datetime
import json
import re
import yaml
from django import template
from django.utils.html import strip_tags
@@ -76,6 +77,14 @@ def render_json(value):
return json.dumps(value, indent=4, sort_keys=True)
@register.filter()
def render_yaml(value):
"""
Render a dictionary as formatted YAML.
"""
return yaml.dump(dict(value))
@register.filter()
def model_name(obj):
"""

79
netbox/utilities/testing.py
View File

@@ -1,79 +0,0 @@
import logging
from contextlib import contextmanager
from django.contrib.auth.models import Permission, User
from rest_framework.test import APITestCase as _APITestCase
from users.models import Token
class APITestCase(_APITestCase):
def setUp(self):
"""
Create a superuser and token for API calls.
"""
self.user = User.objects.create(username='testuser', is_superuser=True)
self.token = Token.objects.create(user=self.user)
self.header = {'HTTP_AUTHORIZATION': 'Token {}'.format(self.token.key)}
def assertHttpStatus(self, response, expected_status):
"""
Provide more detail in the event of an unexpected HTTP response.
"""
err_message = "Expected HTTP status {}; received {}: {}"
self.assertEqual(response.status_code, expected_status, err_message.format(
expected_status, response.status_code, getattr(response, 'data', 'No data')
))
def create_test_user(username='testuser', permissions=list()):
"""
Create a User with the given permissions.
"""
user = User.objects.create_user(username=username)
for perm_name in permissions:
app, codename = perm_name.split('.')
perm = Permission.objects.get(content_type__app_label=app, codename=codename)
user.user_permissions.add(perm)
return user
def choices_to_dict(choices_list):
"""
Convert a list of field choices to a dictionary suitable for direct comparison with a ChoiceSet. For example:
[
{
"value": "choice-1",
"label": "First Choice"
},
{
"value": "choice-2",
"label": "Second Choice"
}
]
Becomes:
{
"choice-1": "First Choice",
"choice-2": "Second Choice
}
"""
return {
choice['value']: choice['label'] for choice in choices_list
}
@contextmanager
def disable_warnings(logger_name):
"""
Temporarily suppress expected warning messages to keep the test output clean.
"""
logger = logging.getLogger(logger_name)
current_level = logger.level
logger.setLevel(logging.ERROR)
yield
logger.setLevel(current_level)

2
netbox/utilities/testing/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from .testcases import *
from .utils import *

325
netbox/utilities/testing/testcases.py Normal file
View File

@@ -0,0 +1,325 @@
from django.contrib.auth.models import Permission, User
from django.core.exceptions import ObjectDoesNotExist
from django.test import Client, TestCase as _TestCase, override_settings
from django.urls import reverse, NoReverseMatch
from rest_framework.test import APIClient
from users.models import Token
from .utils import disable_warnings, model_to_dict, post_data
class TestCase(_TestCase):
user_permissions = ()
def setUp(self):
# Create the test user and assign permissions
self.user = User.objects.create_user(username='testuser')
self.add_permissions(*self.user_permissions)
# Initialize the test client
self.client = Client()
self.client.force_login(self.user)
#
# Permissions management
#
def add_permissions(self, *names):
"""
Assign a set of permissions to the test user. Accepts permission names in the form <app>.<action>_<model>.
"""
for name in names:
app, codename = name.split('.')
perm = Permission.objects.get(content_type__app_label=app, codename=codename)
self.user.user_permissions.add(perm)
def remove_permissions(self, *names):
"""
Remove a set of permissions from the test user, if assigned.
"""
for name in names:
app, codename = name.split('.')
perm = Permission.objects.get(content_type__app_label=app, codename=codename)
self.user.user_permissions.remove(perm)
#
# Convenience methods
#
def assertHttpStatus(self, response, expected_status):
"""
TestCase method. Provide more detail in the event of an unexpected HTTP response.
"""
err_message = "Expected HTTP status {}; received {}: {}"
self.assertEqual(response.status_code, expected_status, err_message.format(
expected_status, response.status_code, getattr(response, 'data', 'No data')
))
class APITestCase(TestCase):
client_class = APIClient
def setUp(self):
"""
Create a superuser and token for API calls.
"""
self.user = User.objects.create(username='testuser', is_superuser=True)
self.token = Token.objects.create(user=self.user)
self.header = {'HTTP_AUTHORIZATION': 'Token {}'.format(self.token.key)}
class StandardTestCases:
"""
We keep any TestCases with test_* methods inside a class to prevent unittest from trying to run them.
"""
class Views(TestCase):
"""
Stock TestCase suitable for testing all standard View functions:
- List objects
- View single object
- Create new object
- Modify existing object
- Delete existing object
- Import multiple new objects
"""
model = None
# Data to be sent when creating/editing individual objects
form_data = {}
# CSV lines used for bulk import of new objects
csv_data = ()
# Form data to be used when editing multiple objects at once
bulk_edit_data = {}
maxDiff = None
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if self.model is None:
raise Exception("Test case requires model to be defined")
def _get_url(self, action, instance=None):
"""
Return the URL name for a specific action. An instance must be specified for
get/edit/delete views.
"""
url_format = '{}:{}_{{}}'.format(
self.model._meta.app_label,
self.model._meta.model_name
)
if action in ('list', 'add', 'import', 'bulk_edit', 'bulk_delete'):
return reverse(url_format.format(action))
elif action in ('get', 'edit', 'delete'):
if instance is None:
raise Exception("Resolving {} URL requires specifying an instance".format(action))
# Attempt to resolve using slug first
if hasattr(self.model, 'slug'):
try:
return reverse(url_format.format(action), kwargs={'slug': instance.slug})
except NoReverseMatch:
pass
return reverse(url_format.format(action), kwargs={'pk': instance.pk})
else:
raise Exception("Invalid action for URL resolution: {}".format(action))
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects(self):
# Attempt to make the request without required permissions
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.get(self._get_url('list')), 403)
# Assign the required permission and submit again
self.add_permissions(
'{}.view_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.get(self._get_url('list'))
self.assertHttpStatus(response, 200)
# Built-in CSV export
if hasattr(self.model, 'csv_headers'):
response = self.client.get('{}?export'.format(self._get_url('list')))
self.assertHttpStatus(response, 200)
self.assertEqual(response.get('Content-Type'), 'text/csv')
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_get_object(self):
instance = self.model.objects.first()
# Attempt to make the request without required permissions
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.get(instance.get_absolute_url()), 403)
# Assign the required permission and submit again
self.add_permissions(
'{}.view_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.get(instance.get_absolute_url())
self.assertHttpStatus(response, 200)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_create_object(self):
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)
self.assertEqual(initial_count + 1, self.model.objects.count())
instance = self.model.objects.order_by('-pk').first()
self.assertDictEqual(model_to_dict(instance), self.form_data)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_edit_object(self):
instance = self.model.objects.first()
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)
instance = self.model.objects.get(pk=instance.pk)
self.assertDictEqual(model_to_dict(instance), self.form_data)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_delete_object(self):
instance = self.model.objects.first()
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)
with self.assertRaises(ObjectDoesNotExist):
self.model.objects.get(pk=instance.pk)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_import_objects(self):
initial_count = self.model.objects.count()
request = {
'path': self._get_url('import'),
'data': {
'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)
self.assertEqual(self.model.objects.count(), initial_count + len(self.csv_data) - 1)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_bulk_edit_objects(self):
pk_list = self.model.objects.values_list('pk', flat=True)
request = {
'path': self._get_url('bulk_edit'),
'data': {
'pk': pk_list,
'_apply': True, # Form button
},
'follow': False, # Do not follow 302 redirects
}
# Append the form data to the request
request['data'].update(post_data(self.bulk_edit_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(
'{}.change_{}'.format(self.model._meta.app_label, self.model._meta.model_name)
)
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
bulk_edit_fields = self.bulk_edit_data.keys()
for i, instance in enumerate(self.model.objects.filter(pk__in=pk_list)):
self.assertDictEqual(
model_to_dict(instance, fields=bulk_edit_fields),
self.bulk_edit_data,
msg="Instance {} failed to validate after bulk edit: {}".format(i, instance)
)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_bulk_delete_objects(self):
pk_list = self.model.objects.values_list('pk', flat=True)
request = {
'path': self._get_url('bulk_delete'),
'data': {
'pk': pk_list,
'confirm': True,
'_confirm': True, # Form button
},
'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)
# Check that all objects were deleted
self.assertEqual(self.model.objects.count(), 0)

102
netbox/utilities/testing/utils.py Normal file
View File

@@ -0,0 +1,102 @@
import logging
from contextlib import contextmanager
from django.contrib.auth.models import Permission, User
from django.forms.models import model_to_dict as _model_to_dict
def model_to_dict(instance, fields=None, exclude=None):
"""
Customized wrapper for Django's built-in model_to_dict(). Does the following:
- Excludes the instance ID field
- Exclude any fields prepended with an underscore
- Convert any assigned tags to a comma-separated string
"""
_exclude = ['id']
if exclude is not None:
_exclude += exclude
model_dict = _model_to_dict(instance, fields=fields, exclude=_exclude)
for key in list(model_dict.keys()):
if key.startswith('_'):
del model_dict[key]
# TODO: Differentiate between tags assigned to the instance and a M2M field for tags (ex: ConfigContext)
elif key == 'tags':
model_dict[key] = ','.join(sorted([tag.name for tag in model_dict['tags']]))
# Convert ManyToManyField to list of instance PKs
elif model_dict[key] and type(model_dict[key]) in (list, tuple) and hasattr(model_dict[key][0], 'pk'):
model_dict[key] = [obj.pk for obj in model_dict[key]]
return model_dict
def post_data(data):
"""
Take a dictionary of test data (suitable for comparison to an instance) and return a dict suitable for POSTing.
"""
ret = {}
for key, value in data.items():
if value is None:
ret[key] = ''
elif type(value) in (list, tuple):
ret[key] = value
else:
ret[key] = str(value)
return ret
def create_test_user(username='testuser', permissions=list()):
"""
Create a User with the given permissions.
"""
user = User.objects.create_user(username=username)
for perm_name in permissions:
app, codename = perm_name.split('.')
perm = Permission.objects.get(content_type__app_label=app, codename=codename)
user.user_permissions.add(perm)
return user
def choices_to_dict(choices_list):
"""
Convert a list of field choices to a dictionary suitable for direct comparison with a ChoiceSet. For example:
[
{
"value": "choice-1",
"label": "First Choice"
},
{
"value": "choice-2",
"label": "Second Choice"
}
]
Becomes:
{
"choice-1": "First Choice",
"choice-2": "Second Choice
}
"""
return {
choice['value']: choice['label'] for choice in choices_list
}
@contextmanager
def disable_warnings(logger_name):
"""
Temporarily suppress expected warning messages to keep the test output clean.
"""
logger = logging.getLogger(logger_name)
current_level = logger.level
logger.setLevel(logging.ERROR)
yield
logger.setLevel(current_level)

55
netbox/utilities/views.py
View File

@@ -4,9 +4,9 @@ from copy import deepcopy
from django.conf import settings
from django.contrib import messages
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.core.exceptions import FieldDoesNotExist, ValidationError
from django.db import transaction, IntegrityError
from django.db.models import ProtectedError
from django.db.models import Count, ManyToManyField, ProtectedError
from django.db.models.query import QuerySet
from django.forms import CharField, Form, ModelMultipleChoiceField, MultipleHiddenInput, Textarea
from django.http import HttpResponse, HttpResponseServerError
@@ -88,15 +88,27 @@ class ObjectListView(View):
Export the queryset of objects as comma-separated value (CSV), using the model's to_csv() method.
"""
csv_data = []
custom_fields = []
# Start with the column headers
headers = ','.join(self.queryset.model.csv_headers)
csv_data.append(headers)
headers = self.queryset.model.csv_headers.copy()
# Add custom field headers, if any
if hasattr(self.queryset.model, 'get_custom_fields'):
for custom_field in self.queryset.model().get_custom_fields():
headers.append(custom_field.name)
custom_fields.append(custom_field.name)
csv_data.append(','.join(headers))
# Iterate through the queryset appending each object
for obj in self.queryset:
data = csv_format(obj.to_csv())
csv_data.append(data)
data = obj.to_csv()
for custom_field in custom_fields:
data += (obj.cf.get(custom_field, ''),)
csv_data.append(csv_format(data))
return '\n'.join(csv_data)
@@ -631,7 +643,9 @@ class BulkEditView(GetReturnURLMixin, View):
if form.is_valid():
custom_fields = form.custom_fields if hasattr(form, 'custom_fields') else []
standard_fields = [field for field in form.fields if field not in custom_fields and field != 'pk']
standard_fields = [
field for field in form.fields if field not in custom_fields + ['pk']
]
nullified_fields = request.POST.getlist('_nullify')
try:
@@ -643,14 +657,29 @@ class BulkEditView(GetReturnURLMixin, View):
# Update standard fields. If a field is listed in _nullify, delete its value.
for name in standard_fields:
if name in form.nullable_fields and name in nullified_fields and isinstance(form.cleaned_data[name], QuerySet):
getattr(obj, name).set([])
elif name in form.nullable_fields and name in nullified_fields:
setattr(obj, name, '' if isinstance(form.fields[name], CharField) else None)
elif isinstance(form.cleaned_data[name], QuerySet) and form.cleaned_data[name]:
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
# Handle nullification
if name in form.nullable_fields and name in nullified_fields:
if isinstance(model_field, ManyToManyField):
getattr(obj, name).set([])
else:
setattr(obj, name, None if model_field.null else '')
# ManyToManyFields
elif isinstance(model_field, ManyToManyField):
getattr(obj, name).set(form.cleaned_data[name])
elif form.cleaned_data[name] not in (None, '') and not isinstance(form.cleaned_data[name], QuerySet):
# Normal fields
elif form.cleaned_data[name] not in (None, ''):
setattr(obj, name, form.cleaned_data[name])
obj.full_clean()
obj.save()

12
netbox/virtualization/forms.py
View File

@@ -6,7 +6,9 @@ from dcim.choices import InterfaceModeChoices
from dcim.constants import INTERFACE_MTU_MAX, INTERFACE_MTU_MIN
from dcim.forms import INTERFACE_MODE_HELP_TEXT
from dcim.models import Device, DeviceRole, Interface, Platform, Rack, Region, Site
from extras.forms import AddRemoveTagsForm, CustomFieldBulkEditForm, CustomFieldForm, CustomFieldFilterForm
from extras.forms import (
AddRemoveTagsForm, CustomFieldBulkEditForm, CustomFieldModelCSVForm, CustomFieldModelForm, CustomFieldFilterForm,
)
from ipam.models import IPAddress, VLANGroup, VLAN
from tenancy.forms import TenancyFilterForm, TenancyForm
from tenancy.models import Tenant
@@ -74,7 +76,7 @@ class ClusterGroupCSVForm(forms.ModelForm):
# Clusters
#
class ClusterForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class ClusterForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
comments = CommentField()
tags = TagField(
required=False
@@ -98,7 +100,7 @@ class ClusterForm(BootstrapMixin, TenancyForm, CustomFieldForm):
}
class ClusterCSVForm(forms.ModelForm):
class ClusterCSVForm(CustomFieldModelCSVForm):
type = forms.ModelChoiceField(
queryset=ClusterType.objects.all(),
to_field_name='name',
@@ -328,7 +330,7 @@ class ClusterRemoveDevicesForm(ConfirmationForm):
# Virtual Machines
#
class VirtualMachineForm(BootstrapMixin, TenancyForm, CustomFieldForm):
class VirtualMachineForm(BootstrapMixin, TenancyForm, CustomFieldModelForm):
cluster_group = forms.ModelChoiceField(
queryset=ClusterGroup.objects.all(),
required=False,
@@ -431,7 +433,7 @@ class VirtualMachineForm(BootstrapMixin, TenancyForm, CustomFieldForm):
self.fields['primary_ip6'].widget.attrs['readonly'] = True
class VirtualMachineCSVForm(forms.ModelForm):
class VirtualMachineCSVForm(CustomFieldModelCSVForm):
status = CSVChoiceField(
choices=VirtualMachineStatusChoices,
required=False,

257
netbox/virtualization/tests/test_views.py
View File

@@ -1,23 +1,19 @@
import urllib.parse
from django.test import Client, TestCase
from django.urls import reverse
from utilities.testing import create_test_user
from dcim.models import DeviceRole, Platform, Site
from utilities.testing import StandardTestCases
from virtualization.choices import *
from virtualization.models import Cluster, ClusterGroup, ClusterType, VirtualMachine
class ClusterGroupTestCase(TestCase):
class ClusterGroupTestCase(StandardTestCases.Views):
model = ClusterGroup
def setUp(self):
user = create_test_user(
permissions=[
'virtualization.view_clustergroup',
'virtualization.add_clustergroup',
]
)
self.client = Client()
self.client.force_login(user)
# Disable inapplicable tests
test_get_object = None
test_delete_object = None
test_bulk_edit_objects = None
@classmethod
def setUpTestData(cls):
ClusterGroup.objects.bulk_create([
ClusterGroup(name='Cluster Group 1', slug='cluster-group-1'),
@@ -25,39 +21,29 @@ class ClusterGroupTestCase(TestCase):
ClusterGroup(name='Cluster Group 3', slug='cluster-group-3'),
])
def test_clustergroup_list(self):
cls.form_data = {
'name': 'Cluster Group X',
'slug': 'cluster-group-x',
}
url = reverse('virtualization:clustergroup_list')
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
def test_clustergroup_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"Cluster Group 4,cluster-group-4",
"Cluster Group 5,cluster-group-5",
"Cluster Group 6,cluster-group-6",
)
response = self.client.post(reverse('virtualization:clustergroup_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(ClusterGroup.objects.count(), 6)
class ClusterTypeTestCase(StandardTestCases.Views):
model = ClusterType
# Disable inapplicable tests
test_get_object = None
test_delete_object = None
test_bulk_edit_objects = None
class ClusterTypeTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'virtualization.view_clustertype',
'virtualization.add_clustertype',
]
)
self.client = Client()
self.client.force_login(user)
@classmethod
def setUpTestData(cls):
ClusterType.objects.bulk_create([
ClusterType(name='Cluster Type 1', slug='cluster-type-1'),
@@ -65,134 +51,139 @@ class ClusterTypeTestCase(TestCase):
ClusterType(name='Cluster Type 3', slug='cluster-type-3'),
])
def test_clustertype_list(self):
cls.form_data = {
'name': 'Cluster Type X',
'slug': 'cluster-type-x',
}
url = reverse('virtualization:clustertype_list')
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
def test_clustertype_import(self):
csv_data = (
cls.csv_data = (
"name,slug",
"Cluster Type 4,cluster-type-4",
"Cluster Type 5,cluster-type-5",
"Cluster Type 6,cluster-type-6",
)
response = self.client.post(reverse('virtualization:clustertype_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(ClusterType.objects.count(), 6)
class ClusterTestCase(StandardTestCases.Views):
model = Cluster
@classmethod
def setUpTestData(cls):
class ClusterTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'virtualization.view_cluster',
'virtualization.add_cluster',
]
sites = (
Site(name='Site 1', slug='site-1'),
Site(name='Site 2', slug='site-2'),
)
self.client = Client()
self.client.force_login(user)
Site.objects.bulk_create(sites)
clustergroup = ClusterGroup(name='Cluster Group 1', slug='cluster-group-1')
clustergroup.save()
clustergroups = (
ClusterGroup(name='Cluster Group 1', slug='cluster-group-1'),
ClusterGroup(name='Cluster Group 2', slug='cluster-group-2'),
)
ClusterGroup.objects.bulk_create(clustergroups)
clustertype = ClusterType(name='Cluster Type 1', slug='cluster-type-1')
clustertype.save()
clustertypes = (
ClusterType(name='Cluster Type 1', slug='cluster-type-1'),
ClusterType(name='Cluster Type 2', slug='cluster-type-2'),
)
ClusterType.objects.bulk_create(clustertypes)
Cluster.objects.bulk_create([
Cluster(name='Cluster 1', group=clustergroup, type=clustertype),
Cluster(name='Cluster 2', group=clustergroup, type=clustertype),
Cluster(name='Cluster 3', group=clustergroup, type=clustertype),
Cluster(name='Cluster 1', group=clustergroups[0], type=clustertypes[0], site=sites[0]),
Cluster(name='Cluster 2', group=clustergroups[0], type=clustertypes[0], site=sites[0]),
Cluster(name='Cluster 3', group=clustergroups[0], type=clustertypes[0], site=sites[0]),
])
def test_cluster_list(self):
url = reverse('virtualization:cluster_list')
params = {
"group": ClusterGroup.objects.first().slug,
"type": ClusterType.objects.first().slug,
cls.form_data = {
'name': 'Cluster X',
'group': clustergroups[1].pk,
'type': clustertypes[1].pk,
'tenant': None,
'site': sites[1].pk,
'comments': 'Some comments',
'tags': 'Alpha,Bravo,Charlie',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_cluster(self):
cluster = Cluster.objects.first()
response = self.client.get(cluster.get_absolute_url())
self.assertEqual(response.status_code, 200)
def test_cluster_import(self):
csv_data = (
cls.csv_data = (
"name,type",
"Cluster 4,Cluster Type 1",
"Cluster 5,Cluster Type 1",
"Cluster 6,Cluster Type 1",
)
response = self.client.post(reverse('virtualization:cluster_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(Cluster.objects.count(), 6)
class VirtualMachineTestCase(TestCase):
def setUp(self):
user = create_test_user(
permissions=[
'virtualization.view_virtualmachine',
'virtualization.add_virtualmachine',
]
)
self.client = Client()
self.client.force_login(user)
clustertype = ClusterType(name='Cluster Type 1', slug='cluster-type-1')
clustertype.save()
cluster = Cluster(name='Cluster 1', type=clustertype)
cluster.save()
VirtualMachine.objects.bulk_create([
VirtualMachine(name='Virtual Machine 1', cluster=cluster),
VirtualMachine(name='Virtual Machine 2', cluster=cluster),
VirtualMachine(name='Virtual Machine 3', cluster=cluster),
])
def test_virtualmachine_list(self):
url = reverse('virtualization:virtualmachine_list')
params = {
"cluster_id": Cluster.objects.first().pk,
cls.bulk_edit_data = {
'group': clustergroups[1].pk,
'type': clustertypes[1].pk,
'tenant': None,
'site': sites[1].pk,
'comments': 'New comments',
}
response = self.client.get('{}?{}'.format(url, urllib.parse.urlencode(params)))
self.assertEqual(response.status_code, 200)
def test_virtualmachine(self):
class VirtualMachineTestCase(StandardTestCases.Views):
model = VirtualMachine
virtualmachine = VirtualMachine.objects.first()
response = self.client.get(virtualmachine.get_absolute_url())
self.assertEqual(response.status_code, 200)
@classmethod
def setUpTestData(cls):
def test_virtualmachine_import(self):
deviceroles = (
DeviceRole(name='Device Role 1', slug='device-role-1'),
DeviceRole(name='Device Role 2', slug='device-role-2'),
)
DeviceRole.objects.bulk_create(deviceroles)
csv_data = (
platforms = (
Platform(name='Platform 1', slug='platform-1'),
Platform(name='Platform 2', slug='platform-2'),
)
Platform.objects.bulk_create(platforms)
clustertype = ClusterType.objects.create(name='Cluster Type 1', slug='cluster-type-1')
clusters = (
Cluster(name='Cluster 1', type=clustertype),
Cluster(name='Cluster 2', type=clustertype),
)
Cluster.objects.bulk_create(clusters)
VirtualMachine.objects.bulk_create([
VirtualMachine(name='Virtual Machine 1', cluster=clusters[0], role=deviceroles[0], platform=platforms[0]),
VirtualMachine(name='Virtual Machine 2', cluster=clusters[0], role=deviceroles[0], platform=platforms[0]),
VirtualMachine(name='Virtual Machine 3', cluster=clusters[0], role=deviceroles[0], platform=platforms[0]),
])
cls.form_data = {
'cluster': clusters[1].pk,
'tenant': None,
'platform': platforms[1].pk,
'name': 'Virtual Machine X',
'status': VirtualMachineStatusChoices.STATUS_STAGED,
'role': deviceroles[1].pk,
'primary_ip4': None,
'primary_ip6': None,
'vcpus': 4,
'memory': 32768,
'disk': 4000,
'comments': 'Some comments',
'tags': 'Alpha,Bravo,Charlie',
'local_context_data': None,
}
cls.csv_data = (
"name,cluster",
"Virtual Machine 4,Cluster 1",
"Virtual Machine 5,Cluster 1",
"Virtual Machine 6,Cluster 1",
)
response = self.client.post(reverse('virtualization:virtualmachine_import'), {'csv': '\n'.join(csv_data)})
self.assertEqual(response.status_code, 200)
self.assertEqual(VirtualMachine.objects.count(), 6)
cls.bulk_edit_data = {
'cluster': clusters[1].pk,
'tenant': None,
'platform': platforms[1].pk,
'status': VirtualMachineStatusChoices.STATUS_STAGED,
'role': deviceroles[1].pk,
'vcpus': 8,
'memory': 65535,
'disk': 8000,
'comments': 'New comments',
}