#1694: Initial work on "next available" prefix provisioning

netbox/ipam/api/serializers.py

@@ -237,6 +237,20 @@ class WritablePrefixSerializer(CustomFieldModelSerializer):
         ]
 
 
+class AvailablePrefixSerializer(serializers.Serializer):
+
+    def to_representation(self, instance):
+        if self.context.get('vrf'):
+            vrf = NestedVRFSerializer(self.context['vrf'], context={'request': self.context['request']}).data
+        else:
+            vrf = None
+        return OrderedDict([
+            ('family', instance.version),
+            ('prefix', str(instance)),
+            ('vrf', vrf),
+        ])
+
+
 #
 # IP addresses
 #

netbox/ipam/api/views.py

@@ -80,6 +80,65 @@ class PrefixViewSet(CustomFieldModelViewSet):
     write_serializer_class = serializers.WritablePrefixSerializer
     filter_class = filters.PrefixFilter
 
+    @detail_route(url_path='available-prefixes', methods=['get', 'post'])
+    def available_prefixes(self, request, pk=None):
+        """
+        A convenience method for returning available child prefixes within a parent.
+        """
+        prefix = get_object_or_404(Prefix, pk=pk)
+        available_prefixes = prefix.get_available_prefixes()
+
+        if request.method == 'POST':
+
+            # Permissions check
+            if not request.user.has_perm('ipam.add_prefix'):
+                raise PermissionDenied()
+
+            requested_prefixes = request.data if isinstance(request.data, list) else [request.data]
+
+            # Allocate prefixes to the requested objects based on availability within the parent
+            for requested_prefix in requested_prefixes:
+
+                # Find the first available prefix equal to or larger than the requested size
+                for available_prefix in available_prefixes.iter_cidrs():
+                    if requested_prefix['prefix_length'] >= available_prefix.prefixlen:
+                        allocated_prefix = '{}/{}'.format(available_prefix.network, requested_prefix['prefix_length'])
+                        requested_prefix['prefix'] = allocated_prefix
+                        requested_prefix['vrf'] = prefix.vrf.pk if prefix.vrf else None
+                        break
+                else:
+                    return Response(
+                        {
+                            "detail": "Insufficient space is available to accommodate the requested prefix size(s)"
+                        },
+                        status=status.HTTP_400_BAD_REQUEST
+                    )
+
+                # Remove the allocated prefix from the list of available prefixes
+                available_prefixes.remove(allocated_prefix)
+
+            # Initialize the serializer with a list or a single object depending on what was requested
+            if isinstance(request.data, list):
+                serializer = serializers.WritablePrefixSerializer(data=requested_prefixes, many=True)
+            else:
+                serializer = serializers.WritablePrefixSerializer(data=requested_prefixes[0])
+
+            # Create the new Prefix(es)
+            if serializer.is_valid():
+                serializer.save()
+                return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        else:
+
+            serializer = serializers.AvailablePrefixSerializer(available_prefixes.iter_cidrs(), many=True, context={
+                'request': request,
+                'vrf': prefix.vrf,
+            })
+
+            return Response(serializer.data)
+
     @detail_route(url_path='available-ips', methods=['get', 'post'])
     def available_ips(self, request, pk=None):
         """

netbox/ipam/models.py

@@ -281,6 +281,21 @@ class Prefix(CreatedUpdatedModel, CustomFieldModel):
     def get_duplicates(self):
         return Prefix.objects.filter(vrf=self.vrf, prefix=str(self.prefix)).exclude(pk=self.pk)
 
+    def get_child_prefixes(self):
+        """
+        Return all child Prefixes within this Prefix.
+        """
+        return Prefix.objects.filter(prefix__net_contained=str(self.prefix), vrf=self.vrf)
+
+    def get_available_prefixes(self):
+        """
+        Return all available prefixes within this Prefix.
+        """
+        prefix = netaddr.IPSet(self.prefix)
+        child_prefixes = netaddr.IPSet([p.prefix for p in self.get_child_prefixes()])
+        available_prefixes = prefix - child_prefixes
+        return available_prefixes
+
     def get_child_ips(self):
         """
         Return all IPAddresses within this Prefix.

netbox/ipam/tests/test_api.py

@@ -365,6 +365,72 @@ class PrefixTest(HttpStatusMixin, APITestCase):
         self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
         self.assertEqual(Prefix.objects.count(), 2)
 
+    def test_list_available_prefixes(self):
+
+        prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
+        Prefix.objects.create(prefix=IPNetwork('192.0.2.64/26'))
+        Prefix.objects.create(prefix=IPNetwork('192.0.2.192/27'))
+        url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
+
+        # Retrieve all available IPs
+        response = self.client.get(url, **self.header)
+        available_prefixes = ['192.0.2.0/26', '192.0.2.128/26', '192.0.2.224/27']
+        for i, p in enumerate(response.data):
+            self.assertEqual(p['prefix'], available_prefixes[i])
+
+    def test_create_single_available_prefix(self):
+
+        prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/28'), is_pool=True)
+        url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
+
+        # Create four available prefixes with individual requests
+        prefixes_to_be_created = [
+            '192.0.2.0/30',
+            '192.0.2.4/30',
+            '192.0.2.8/30',
+            '192.0.2.12/30',
+        ]
+        for i in range(4):
+            data = {
+                'prefix_length': 30,
+                'description': 'Test Prefix {}'.format(i + 1)
+            }
+            response = self.client.post(url, data, format='json', **self.header)
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            self.assertEqual(response.data['prefix'], prefixes_to_be_created[i])
+            self.assertEqual(response.data['description'], data['description'])
+
+        # Try to create one more prefix
+        response = self.client.post(url, {'prefix_length': 30}, **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('detail', response.data)
+
+    def test_create_multiple_available_prefixes(self):
+
+        prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/28'), is_pool=True)
+        url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
+
+        # Try to create five /30s (only four are available)
+        data = [
+            {'prefix_length': 30, 'description': 'Test Prefix 1'},
+            {'prefix_length': 30, 'description': 'Test Prefix 2'},
+            {'prefix_length': 30, 'description': 'Test Prefix 3'},
+            {'prefix_length': 30, 'description': 'Test Prefix 4'},
+            {'prefix_length': 30, 'description': 'Test Prefix 5'},
+        ]
+        response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('detail', response.data)
+
+        # Verify that no prefixes were created (the entire /28 is still available)
+        response = self.client.get(url, **self.header)
+        self.assertEqual(response.data[0]['prefix'], '192.0.2.0/28')
+
+        # Create four /30s in a single request
+        response = self.client.post(url, data[:4], format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+        self.assertEqual(len(response.data), 4)
+
     def test_list_available_ips(self):
 
         prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'), is_pool=True)
@@ -391,8 +457,6 @@ class PrefixTest(HttpStatusMixin, APITestCase):
                 'description': 'Test IP {}'.format(i)
             }
             response = self.client.post(url, data, format='json', **self.header)
-            if response.status_code != status.HTTP_201_CREATED:
-                assert False, response.content
             self.assertHttpStatus(response, status.HTTP_201_CREATED)
             self.assertEqual(response.data['description'], data['description'])