Skip to content
This repository has been archived by the owner on May 19, 2020. It is now read-only.

Commit

Permalink
[api] RESTful api for Subnet and IpAddress
Browse files Browse the repository at this point in the history
  • Loading branch information
anurag-ks committed Jul 10, 2018
1 parent 1bc21c1 commit b742fd5
Show file tree
Hide file tree
Showing 8 changed files with 294 additions and 92 deletions.
51 changes: 51 additions & 0 deletions django_ipam/api/generics.py
@@ -0,0 +1,51 @@
from django.shortcuts import get_object_or_404
from rest_framework import status
from rest_framework.authentication import BasicAuthentication, SessionAuthentication
from rest_framework.generics import CreateAPIView, ListCreateAPIView, RetrieveUpdateDestroyAPIView
from rest_framework.permissions import DjangoModelPermissions
from rest_framework.response import Response

from .serializers import IpAddressSerializer, IpRequestSerializer, SubnetSerializer


class BaseIpAddressListCreateView(ListCreateAPIView):
serializer_class = IpAddressSerializer
authentication_classes = (SessionAuthentication, BasicAuthentication)
permission_classes = (DjangoModelPermissions,)

def get_queryset(self):
subnet = get_object_or_404(self.subnet_model, pk=self.kwargs["subnet_id"])
return subnet.ipaddress_set.all()


class BaseSubnetListCreateView(ListCreateAPIView):
serializer_class = SubnetSerializer
authentication_classes = (SessionAuthentication, BasicAuthentication)
permission_classes = (DjangoModelPermissions,)


class BaseSubnetView(RetrieveUpdateDestroyAPIView):
serializer_class = SubnetSerializer
authentication_classes = (SessionAuthentication, BasicAuthentication)
permission_classes = (DjangoModelPermissions,)


class BaseIpAddressView(RetrieveUpdateDestroyAPIView):
serializer_class = IpAddressSerializer
authentication_classes = (SessionAuthentication, BasicAuthentication)
permission_classes = (DjangoModelPermissions,)


class BaseRequestIPView(CreateAPIView):
serializer_class = IpRequestSerializer
authentication_classes = (SessionAuthentication, BasicAuthentication)
permission_classes = (DjangoModelPermissions,)

def post(self, request, *args, **kwargs):
subnet = get_object_or_404(self.subnet_model, pk=kwargs["subnet_id"])
ip_address = subnet.request_ip(dict(description=request.data.get("description")))
if ip_address:
serializer = IpAddressSerializer(ip_address)
headers = self.get_success_headers(serializer.data)
return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
return Response(None)
14 changes: 12 additions & 2 deletions django_ipam/api/serializers.py
@@ -1,16 +1,26 @@
import swapper
from rest_framework import serializers

IpAddress = swapper.load_model("django_ipam", "IpAddress")
IpAddress = swapper.load_model('django_ipam', 'IpAddress')
Subnet = swapper.load_model('django_ipam', 'Subnet')


class IpRequestSerializer(serializers.ModelSerializer):
class Meta:
model = IpAddress
fields = ('subnet', 'description')
read_only_fields = ('created', 'modified')


class IpAddressSerializer(serializers.ModelSerializer):
class Meta:
model = IpAddress
fields = ('ip_address', 'subnet', 'description')
fields = '__all__'
read_only_fields = ('created', 'modified')


class SubnetSerializer(serializers.ModelSerializer):
class Meta:
model = Subnet
fields = '__all__'
read_only_fields = ('created', 'modified')
14 changes: 13 additions & 1 deletion django_ipam/api/urls.py
Expand Up @@ -7,6 +7,18 @@
views.get_first_available_ip,
name='get_first_available_ip'),
url(r'^subnet/(?P<subnet_id>[^/]+)/request-ip/$',
views.RequestIPView.as_view(),
views.request_ip,
name='request_ip'),
url(r'^subnet/(?P<subnet_id>[^/]+)/ip-address$',
views.subnet_list_ipaddress,
name='list_create_ip_address'),
url(r'^subnet/$',
views.subnet_list_create,
name='subnet_list_create'),
url(r'^subnet/(?P<pk>[^/]+)$',
views.subnet,
name='subnet'),
url(r'^ip-address/(?P<pk>[^/]+)/$',
views.ip_address,
name='ip_address')
]
55 changes: 42 additions & 13 deletions django_ipam/api/views.py
@@ -1,12 +1,14 @@
import swapper
from django.shortcuts import get_object_or_404
from rest_framework import status
from rest_framework.decorators import api_view
from rest_framework.generics import CreateAPIView
from rest_framework.response import Response

from .serializers import IpAddressSerializer, IpRequestSerializer
from .generics import (
BaseIpAddressListCreateView, BaseIpAddressView, BaseRequestIPView, BaseSubnetListCreateView,
BaseSubnetView,
)

IpAddress = swapper.load_model('django_ipam', 'IpAddress')
Subnet = swapper.load_model('django_ipam', 'Subnet')


Expand All @@ -19,17 +21,44 @@ def get_first_available_ip(request, subnet_id):
return Response(subnet.get_first_available_ip())


class RequestIPView(CreateAPIView):
class RequestIPView(BaseRequestIPView):
"""
Request and create a record for the next available IP address under a subnet
"""
serializer_class = IpRequestSerializer
subnet_model = Subnet
queryset = IpAddress.objects.none()

def post(self, request, *args, **kwargs):
subnet = get_object_or_404(Subnet, pk=kwargs["subnet_id"])
ip_address = subnet.request_ip(dict(description=request.data.get("description")))
if ip_address:
serializer = IpAddressSerializer(ip_address)
headers = self.get_success_headers(serializer.data)
return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
return Response(None)

class SubnetIpAddressListCreateView(BaseIpAddressListCreateView):
"""
List/Create IP addresses under a specific subnet
"""
subnet_model = Subnet


class SubnetListCreateView(BaseSubnetListCreateView):
"""
List/Create subnets
"""
queryset = Subnet.objects.all()


class SubnetView(BaseSubnetView):
"""
View for retrieving, updating or deleting a subnet instance.
"""
queryset = Subnet.objects.all()


class IpAddressView(BaseIpAddressView):
"""
View for retrieving, updating or deleting a IP address instance.
"""
queryset = IpAddress.objects.all()


request_ip = RequestIPView.as_view()
subnet_list_create = SubnetListCreateView.as_view()
subnet = SubnetView.as_view()
ip_address = IpAddressView.as_view()
subnet_list_ipaddress = SubnetIpAddressListCreateView.as_view()
8 changes: 4 additions & 4 deletions django_ipam/tests/base/base.py
Expand Up @@ -2,14 +2,14 @@


class CreateModelsMixin(object):
def _create_subnet(self, options):
instance = Subnet(**options)
def _create_subnet(self, **kwargs):
instance = Subnet(**dict(kwargs))
instance.full_clean()
instance.save()
return instance

def _create_ipaddress(self, options):
instance = IpAddress(**options)
def _create_ipaddress(self, **kwargs):
instance = IpAddress(**dict(kwargs))
instance.full_clean()
instance.save()
return instance
22 changes: 11 additions & 11 deletions django_ipam/tests/base/test_admin.py
Expand Up @@ -12,7 +12,7 @@ def setUp(self):
self.client.login(username='admin', password='tester')

def test_ipaddress_invalid_entry(self):
subnet = self._create_subnet(dict(subnet="10.0.0.0/24", description="Sample Subnet"))
subnet = self._create_subnet(subnet="10.0.0.0/24", description="Sample Subnet")
post_data = {
'ip_address': "12344",
'subnet': subnet.id,
Expand All @@ -27,26 +27,26 @@ def test_ipaddress_invalid_entry(self):
self.assertContains(response, 'Enter a valid IPv4 or IPv6 address.')

def test_ipaddress_change(self):
subnet = self._create_subnet(dict(subnet="10.0.0.0/24", description="Sample Subnet"))
obj = self._create_ipaddress(dict(ip_address="10.0.0.1", subnet=subnet))
subnet = self._create_subnet(subnet="10.0.0.0/24", description="Sample Subnet")
obj = self._create_ipaddress(ip_address="10.0.0.1", subnet=subnet)

response = self.client.get(reverse('admin:{0}_ipaddress_change'.format(self.app_name), args=[obj.pk]),
follow=True)
self.assertContains(response, 'ok')
self.assertEqual(self.ipaddress_model.objects.get(pk=obj.pk).ip_address, '10.0.0.1')

def test_ipv4_subnet_change(self):
subnet = self._create_subnet(dict(subnet="10.0.0.0/24", description="Sample Subnet"))
self._create_ipaddress(dict(ip_address="10.0.0.1", subnet=subnet))
subnet = self._create_subnet(subnet="10.0.0.0/24", description="Sample Subnet")
self._create_ipaddress(ip_address="10.0.0.1", subnet=subnet)

response = self.client.get(reverse('admin:{0}_subnet_change'.format(self.app_name), args=[subnet.pk]),
follow=True)
self.assertContains(response, 'ok')
self.assertContains(response, '<h3>Subnet Visual Display</h3>')

def test_ipv6_subnet_change(self):
subnet = self._create_subnet(dict(subnet="fdb6:21b:a477::9f7/64", description="Sample Subnet"))
self._create_ipaddress(dict(ip_address="fdb6:21b:a477::9f7", subnet=subnet))
subnet = self._create_subnet(subnet="fdb6:21b:a477::9f7/64", description="Sample Subnet")
self._create_ipaddress(ip_address="fdb6:21b:a477::9f7", subnet=subnet)

response = self.client.get(reverse('admin:{0}_subnet_change'.format(self.app_name), args=[subnet.pk]),
follow=True)
Expand All @@ -69,15 +69,15 @@ def test_subnet_invalid_entry(self):
self.assertContains(response, 'Enter a valid CIDR address.')

def test_subnet_popup_response(self):
subnet = self._create_subnet(dict(subnet="fdb6:21b:a477::9f7/64", description="Sample Subnet"))
self._create_ipaddress(dict(ip_address="fdb6:21b:a477::9f7", subnet=subnet))
subnet = self._create_subnet(subnet="fdb6:21b:a477::9f7/64", description="Sample Subnet")
self._create_ipaddress(ip_address="fdb6:21b:a477::9f7", subnet=subnet)

response = self.client.get('/admin/django_ipam/subnet/{0}/change/?_popup=1'.format(subnet.id),
follow=True)
self.assertContains(response, 'ok')

def test_ipaddress_response(self):
subnet = self._create_subnet(dict(subnet="10.0.0.0/24", description="Sample Subnet"))
subnet = self._create_subnet(subnet="10.0.0.0/24", description="Sample Subnet")
post_data = {
'ip_address': "10.0.0.1",
'subnet': subnet.id,
Expand All @@ -91,7 +91,7 @@ def test_ipaddress_response(self):
self.assertContains(response, 'ok')

def test_ipaddress_popup_response(self):
subnet = self._create_subnet(dict(subnet="10.0.0.0/24", description="Sample Subnet"))
subnet = self._create_subnet(subnet="10.0.0.0/24", description="Sample Subnet")
post_data = {
'ip_address': "10.0.0.1",
'subnet': subnet.id,
Expand Down

0 comments on commit b742fd5

Please sign in to comment.