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

Closes #2658: Avalable VLANs API endpoint for VLAN groups

Browse Source
This commit is contained in:
jeremystretch
2021-12-23 10:14:28 -05:00
parent 2dd165bbef
commit e0cfd5e49b
7 changed files with 202 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/release-notes/version-3.2.md
View File

@@ -14,6 +14,10 @@
### New Features
#### Automatic Provisioning of Next Available VLANs ([#2658](https://github.com/netbox-community/netbox/issues/2658))
A new REST API endpoint has been added at `/api/ipam/vlan-groups/<pk>/available-vlans/`. A GET request to this endpoint will return a list of available VLANs within the group. A POST request can be made to this endpoint specifying the name(s) of one or more VLANs to create within the group, and their VLAN IDs will be assigned automatically.
#### Modules & Module Types ([#7844](https://github.com/netbox-community/netbox/issues/7844))
Several new models have been added to support field-replaceable device modules, such as those within a chassis-based switch or router. Similar to devices, each module is instantiated from a user-defined module type, and can have components associated with it. These components become available to the parent device once the module has been installed within a module bay. This makes it very convenient to replicate the addition and deletion of device components as modules are installed and removed.

34
netbox/ipam/api/serializers.py
View File

@@ -213,6 +213,40 @@ class VLANSerializer(PrimaryModelSerializer):
]
class AvailableVLANSerializer(serializers.Serializer):
"""
Representation of a VLAN which does not exist in the database.
"""
vid = serializers.IntegerField(read_only=True)
group = NestedVLANGroupSerializer(read_only=True)
def to_representation(self, instance):
return OrderedDict([
('vid', instance),
('group', NestedVLANGroupSerializer(
self.context['group'],
context={'request': self.context['request']}
).data),
])
class CreateAvailableVLANSerializer(PrimaryModelSerializer):
site = NestedSiteSerializer(required=False, allow_null=True)
tenant = NestedTenantSerializer(required=False, allow_null=True)
status = ChoiceField(choices=VLANStatusChoices, required=False)
role = NestedRoleSerializer(required=False, allow_null=True)
class Meta:
model = VLAN
fields = [
'name', 'site', 'tenant', 'status', 'role', 'description', 'tags', 'custom_fields',
]
def validate(self, data):
# Bypass model validation since we don't have a VID yet
return data
#
# Prefixes
#

5
netbox/ipam/api/urls.py
View File

@@ -62,6 +62,11 @@ urlpatterns = [
views.PrefixAvailableIPAddressesView.as_view(),
name='prefix-available-ips'
),
path(
'vlan-groups/<int:pk>/available-vlans/',
views.AvailableVLANsView.as_view(),
name='vlangroup-available-vlans'
),
]
urlpatterns += router.urls

72
netbox/ipam/api/views.py
View File

@@ -327,3 +327,75 @@ class IPRangeAvailableIPAddressesView(AvailableIPAddressesView):
def get_parent(self, request, pk):
return get_object_or_404(IPRange.objects.restrict(request.user), pk=pk)
class AvailableVLANsView(ObjectValidationMixin, APIView):
queryset = VLAN.objects.all()
@swagger_auto_schema(responses={200: serializers.AvailableVLANSerializer(many=True)})
def get(self, request, pk):
vlangroup = get_object_or_404(VLANGroup.objects.restrict(request.user), pk=pk)
available_vlans = vlangroup.get_available_vids()
serializer = serializers.AvailableVLANSerializer(available_vlans, many=True, context={
'request': request,
'group': vlangroup,
})
return Response(serializer.data)
@swagger_auto_schema(
request_body=serializers.CreateAvailableVLANSerializer,
responses={201: serializers.VLANSerializer(many=True)}
)
@advisory_lock(ADVISORY_LOCK_KEYS['available-vlans'])
def post(self, request, pk):
self.queryset = self.queryset.restrict(request.user, 'add')
vlangroup = get_object_or_404(VLANGroup.objects.restrict(request.user), pk=pk)
available_vlans = vlangroup.get_available_vids()
many = isinstance(request.data, list)
# Validate requested VLANs
serializer = serializers.CreateAvailableVLANSerializer(
data=request.data if many else [request.data],
many=True,
context={
'request': request,
'group': vlangroup,
}
)
if not serializer.is_valid():
return Response(
serializer.errors,
status=status.HTTP_400_BAD_REQUEST
)
requested_vlans = serializer.validated_data
for i, requested_vlan in enumerate(requested_vlans):
try:
requested_vlan['vid'] = available_vlans.pop(0)
requested_vlan['group'] = vlangroup.pk
except IndexError:
return Response({
"detail": "The requested number of VLANs is not available"
}, status=status.HTTP_409_CONFLICT)
# Initialize the serializer with a list or a single object depending on what was requested
context = {'request': request}
if many:
serializer = serializers.VLANSerializer(data=requested_vlans, many=True, context=context)
else:
serializer = serializers.VLANSerializer(data=requested_vlans[0], context=context)
# Create the new VLAN(s)
if serializer.is_valid():
try:
with transaction.atomic():
created = serializer.save()
self._validate_objects(created)
except ObjectDoesNotExist:
raise PermissionDenied()
return Response(serializer.data, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

10
netbox/ipam/models/vlans.py
View File

@@ -75,6 +75,16 @@ class VLANGroup(OrganizationalModel):
if self.scope_id and not self.scope_type:
raise ValidationError("Cannot set scope_id without scope_type.")
def get_available_vids(self):
"""
Return all available VLANs within this group.
"""
available_vlans = {vid for vid in range(VLAN_VID_MIN, VLAN_VID_MAX + 1)}
available_vlans -= set(VLAN.objects.filter(group=self).values_list('vid', flat=True))
# TODO: Check ordering
return list(available_vlans)
def get_next_available_vid(self):
"""
Return the first available VLAN ID (1-4094) in the group.

76
netbox/ipam/tests/test_api.py
View File

@@ -695,6 +695,82 @@ class VLANGroupTest(APIViewTestCases.APIViewTestCase):
)
VLANGroup.objects.bulk_create(vlan_groups)
def test_list_available_vlans(self):
"""
Test retrieval of all available VLANs within a group.
"""
self.add_permissions('ipam.view_vlan')
vlangroup = VLANGroup.objects.first()
vlans = (
VLAN(vid=10, name='VLAN 10', group=vlangroup),
VLAN(vid=20, name='VLAN 20', group=vlangroup),
VLAN(vid=30, name='VLAN 30', group=vlangroup),
)
VLAN.objects.bulk_create(vlans)
# Retrieve all available VLANs
url = reverse('ipam-api:vlangroup-available-vlans', kwargs={'pk': vlangroup.pk})
response = self.client.get(url, **self.header)
self.assertEqual(len(response.data), 4094 - len(vlans))
available_vlans = {vlan['vid'] for vlan in response.data}
for vlan in vlans:
self.assertNotIn(vlan.vid, available_vlans)
def test_create_single_available_vlan(self):
"""
Test the creation of a single available VLAN.
"""
self.add_permissions('ipam.view_vlan', 'ipam.add_vlan')
vlangroup = VLANGroup.objects.first()
VLAN.objects.create(vid=1, name='VLAN 1', group=vlangroup)
data = {
"name": "First VLAN",
}
url = reverse('ipam-api:vlangroup-available-vlans', kwargs={'pk': vlangroup.pk})
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(response.data['name'], data['name'])
self.assertEqual(response.data['group']['id'], vlangroup.pk)
self.assertEqual(response.data['vid'], 2)
def test_create_multiple_available_vlans(self):
"""
Test the creation of multiple available VLANs.
"""
self.add_permissions('ipam.view_vlan', 'ipam.add_vlan')
vlangroup = VLANGroup.objects.first()
vlans = (
VLAN(vid=1, name='VLAN 1', group=vlangroup),
VLAN(vid=3, name='VLAN 3', group=vlangroup),
VLAN(vid=5, name='VLAN 5', group=vlangroup),
)
VLAN.objects.bulk_create(vlans)
data = (
{"name": "First VLAN"},
{"name": "Second VLAN"},
{"name": "Third VLAN"},
)
url = reverse('ipam-api:vlangroup-available-vlans', kwargs={'pk': vlangroup.pk})
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(len(response.data), 3)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[0]['group']['id'], vlangroup.pk)
self.assertEqual(response.data[0]['vid'], 2)
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[1]['group']['id'], vlangroup.pk)
self.assertEqual(response.data[1]['vid'], 4)
self.assertEqual(response.data[2]['name'], data[2]['name'])
self.assertEqual(response.data[2]['group']['id'], vlangroup.pk)
self.assertEqual(response.data[2]['vid'], 6)
class VLANTest(APIViewTestCases.APIViewTestCase):
model = VLAN

1
netbox/utilities/constants.py
View File

@@ -42,6 +42,7 @@ FILTER_TREENODE_NEGATION_LOOKUP_MAP = dict(
ADVISORY_LOCK_KEYS = {
'available-prefixes': 100100,
'available-ips': 100200,
'available-vlans': 100300,
}
#