Add available IPs REST API endpoint for IP ranges

This commit is contained in:
jeremystretch 2021-07-19 09:32:04 -04:00
parent 747c065213
commit 0fe1a426c0
4 changed files with 78 additions and 7 deletions

View File

@ -98,6 +98,7 @@ class AvailablePrefixesMixin:
class AvailableIPsMixin:
parent_model = Prefix
@swagger_auto_schema(method='get', responses={200: serializers.AvailableIPSerializer(many=True)})
@swagger_auto_schema(method='post', responses={201: serializers.AvailableIPSerializer(many=True)},
@ -113,7 +114,7 @@ class AvailableIPsMixin:
The advisory lock decorator uses a PostgreSQL advisory lock to prevent this API from being
invoked in parallel, which results in a race condition where multiple insertions can occur.
"""
parent = get_object_or_404(Prefix.objects.restrict(request.user), pk=pk)
parent = get_object_or_404(self.parent_model.objects.restrict(request.user), pk=pk)
# Create the next available IP
if request.method == 'POST':
@ -174,8 +175,7 @@ class AvailableIPsMixin:
break
serializer = serializers.AvailableIPSerializer(ip_list, many=True, context={
'request': request,
'prefix': parent.prefix,
'vrf': parent.vrf,
'parent': parent,
})
return Response(serializer.data)

View File

@ -329,9 +329,9 @@ class AvailableIPSerializer(serializers.Serializer):
else:
vrf = None
return OrderedDict([
('family', self.context['prefix'].version),
('address', '{}/{}'.format(instance, self.context['prefix'].prefixlen)),
('vrf', vrf),
('family', self.context['parent'].family),
('address', f"{instance}/{self.context['parent'].mask_length}"),
('vrf', self.context['parent'].vrf),
])

View File

@ -87,6 +87,8 @@ class PrefixViewSet(mixins.AvailableIPsMixin, mixins.AvailablePrefixesMixin, Cus
serializer_class = serializers.PrefixSerializer
filterset_class = filtersets.PrefixFilterSet
parent_model = Prefix # AvailableIPsMixin
def get_serializer_class(self):
if self.action == "available_prefixes" and self.request.method == "POST":
return serializers.PrefixLengthSerializer
@ -97,11 +99,13 @@ class PrefixViewSet(mixins.AvailableIPsMixin, mixins.AvailablePrefixesMixin, Cus
# IP ranges
#
class IPRangeViewSet(CustomFieldModelViewSet):
class IPRangeViewSet(mixins.AvailableIPsMixin, CustomFieldModelViewSet):
queryset = IPRange.objects.prefetch_related('vrf', 'role', 'tenant', 'tags')
serializer_class = serializers.IPRangeSerializer
filterset_class = filtersets.IPRangeFilterSet
parent_model = IPRange # AvailableIPsMixin
#
# IP addresses

View File

@ -390,6 +390,73 @@ class IPRangeTest(APIViewTestCases.APIViewTestCase):
)
IPRange.objects.bulk_create(ip_ranges)
def test_list_available_ips(self):
"""
Test retrieval of all available IP addresses within a parent IP range.
"""
iprange = IPRange.objects.create(
start_address=IPNetwork('192.0.2.10/24'),
end_address=IPNetwork('192.0.2.19/24')
)
url = reverse('ipam-api:iprange-available-ips', kwargs={'pk': iprange.pk})
self.add_permissions('ipam.view_iprange', 'ipam.view_ipaddress')
# Retrieve all available IPs
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data), 10)
def test_create_single_available_ip(self):
"""
Test retrieval of the first available IP address within a parent IP range.
"""
vrf = VRF.objects.create(name='Test VRF 1', rd='1234')
iprange = IPRange.objects.create(
start_address=IPNetwork('192.0.2.1/24'),
end_address=IPNetwork('192.0.2.3/24'),
vrf=vrf
)
url = reverse('ipam-api:iprange-available-ips', kwargs={'pk': iprange.pk})
self.add_permissions('ipam.view_iprange', 'ipam.add_ipaddress')
# Create all three available IPs with individual requests
for i in range(1, 4):
data = {
'description': f'Test IP #{i}'
}
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(response.data['vrf']['id'], vrf.pk)
self.assertEqual(response.data['description'], data['description'])
# Try to create one more IP
response = self.client.post(url, {}, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertIn('detail', response.data)
def test_create_multiple_available_ips(self):
"""
Test the creation of available IP addresses within a parent IP range.
"""
iprange = IPRange.objects.create(
start_address=IPNetwork('192.0.2.1/24'),
end_address=IPNetwork('192.0.2.8/24')
)
url = reverse('ipam-api:iprange-available-ips', kwargs={'pk': iprange.pk})
self.add_permissions('ipam.view_iprange', 'ipam.add_ipaddress')
# Try to create nine IPs (only eight are available)
data = [{'description': f'Test IP #{i}'} for i in range(1, 10)] # 9 IPs
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertIn('detail', response.data)
# Create all eight available IPs in a single request
data = [{'description': f'Test IP #{i}'} for i in range(1, 9)] # 8 IPs
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(len(response.data), 8)
class IPAddressTest(APIViewTestCases.APIViewTestCase):
model = IPAddress