Merge pull request #91 from k01ek/develop

Develop

Author: k01ek

netbox_bgp/forms.py

@@ -11,7 +11,7 @@
 from ipam.models import IPAddress
 from ipam.formfields import IPNetworkFormField
 from utilities.forms import (
-    BootstrapMixin, DynamicModelChoiceField, BulkEditForm,
+    DynamicModelChoiceField,
     DynamicModelMultipleChoiceField, StaticSelect,
     APISelect, APISelectMultiple, StaticSelectMultiple, TagFilterField
 )
@@ -33,7 +33,7 @@ class ASNField(forms.CharField):
 
     def to_python(self, value):
         if not re.match(r'^\d+(\.\d+)?$', value):
-            raise ValidationError('Invalid AS Number: {}'.format(value))       
+            raise ValidationError('Invalid AS Number: {}'.format(value))
         if '.' in value:
             if int(value.split('.')[0]) > 65535 or int(value.split('.')[1]) > 65535:
                 raise ValidationError('Invalid AS Number: {}'.format(value))
@@ -157,7 +157,7 @@ class ASNBulkEditForm(NetBoxModelBulkEditForm):
     )
 
 
-class CommunityForm(BootstrapMixin, forms.ModelForm):
+class CommunityForm(NetBoxModelForm):
     tags = DynamicModelMultipleChoiceField(
         queryset=Tag.objects.all(),
         required=False
@@ -179,7 +179,7 @@ class Meta:
         ]
 
 
-class CommunityFilterForm(BootstrapMixin, forms.ModelForm):
+class CommunityFilterForm(NetBoxModelFilterSetForm):
     q = forms.CharField(
         required=False,
         label='Search'
@@ -200,12 +200,10 @@ class CommunityFilterForm(BootstrapMixin, forms.ModelForm):
 
     tag = TagFilterField(Community)
 
-    class Meta:
-        model = Community
-        fields = ['q', 'status', 'tenant']
+    model = Community
 
 
-class CommunityBulkEditForm(BulkEditForm):
+class CommunityBulkEditForm(NetBoxModelBulkEditForm):
     pk = forms.ModelMultipleChoiceField(
         queryset=Community.objects.all(),
         widget=forms.MultipleHiddenInput
@@ -224,10 +222,10 @@ class CommunityBulkEditForm(BulkEditForm):
         widget=StaticSelect()
     )
 
-    class Meta:
-        nullable_fields = [
-            'tenant', 'description',
-        ]
+    model = Community
+    nullable_fields = [
+       'tenant', 'description',
+    ]
 
 
 class BGPSessionForm(NetBoxModelForm):

netbox_bgp/graphql.py

@@ -0,0 +1,64 @@
+from graphene import ObjectType, Field
+
+from netbox.graphql.scalars import BigInt
+from netbox.graphql.types import NetBoxObjectType
+from netbox.graphql.fields import ObjectField, ObjectListField
+
+from . import models, filters
+
+
+class CommunityType(NetBoxObjectType):
+    class Meta:
+        model = models.Community
+        fields = '__all__'
+        filterset_class = filters.CommunityFilterSet
+
+
+class AsnType(NetBoxObjectType):
+    number = Field(BigInt)
+
+    class Meta:
+        model = models.ASN
+        fields = '__all__'
+        filterset_class = filters.ASNFilterSet
+
+
+class BgpSessionType(NetBoxObjectType):
+    class Meta:
+        model = models.BGPSession
+        fields = '__all__'
+        filterset_class = filters.BGPSessionFilterSet
+
+
+class PeerGroupType(NetBoxObjectType):
+    class Meta:
+        model = models.BGPPeerGroup
+        fields = '__all__'
+        filterset_class = filters.BGPPeerGroupFilterSet
+
+
+class RoutingPolicyType(NetBoxObjectType):
+    class Meta:
+        model = models.RoutingPolicy
+        fields = '__all__'
+        filterset_class = filters.RoutingPolicyFilterSet
+
+
+class BGPQuery(ObjectType):
+    community = ObjectField(CommunityType)
+    community_list = ObjectListField(CommunityType)
+
+    bgp_asn = ObjectField(AsnType)
+    bgp_asn_list = ObjectListField(AsnType)
+
+    bgp_session = ObjectField(BgpSessionType)
+    bgp_session_list = ObjectListField(BgpSessionType)
+
+    peer_group = ObjectField(PeerGroupType)
+    peer_group_list = ObjectListField(PeerGroupType)
+
+    routing_policy = ObjectField(RoutingPolicyType)
+    routing_policy_list = ObjectListField(RoutingPolicyType)
+
+
+schema = BGPQuery

netbox_bgp/tests/test_api.py

@@ -1,5 +1,8 @@
+import json
+
 from django.contrib.auth.models import User
 from django.test import TestCase
+from django.test.client import Client
 from django.urls import reverse
 from rest_framework import status
 from rest_framework.test import APIClient, APITestCase
@@ -18,8 +21,10 @@ class BaseTestCase(TestCase):
     def setUp(self):
         self.user = User.objects.create(username='testuser', is_superuser=True)
         self.token = Token.objects.create(user=self.user)
+        # todo change to Client
         self.client = APIClient()
         self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
+        self.gql_client = Client(HTTP_AUTHORIZATION=f'Token {self.token.key}')
 
 
 class ASNTestCase(BaseTestCase):
@@ -92,6 +97,27 @@ def test_uniqueconstraint_asn(self):
         response = self.client.post(url, data, format='json')
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
 
+    def test_graphql(self):
+        url = reverse('graphql')
+        query = 'query bgp_asn($id: Int!){bgp_asn(id: $id){number}}'
+        response = self.gql_client.post(
+            url,
+            json.dumps({'query': query, 'variables': {'id': self.asn1.pk}}),
+            content_type='application/json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(json.loads(response.content)['data']['bgp_asn']['number'], self.asn1.number)
+
+    def test_graphql_list(self):
+        url = reverse('graphql')
+        query = '{bgp_asn_list{number}}'
+        response = self.gql_client.post(
+            url,
+            json.dumps({'query': query}),
+            content_type='application/json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
 
 class CommunityTestCase(BaseTestCase):
     def setUp(self):
@@ -126,6 +152,27 @@ def test_update_community(self):
     def test_delete_community(self):
         pass
 
+    def test_graphql(self):
+        url = reverse('graphql')
+        query = 'query community($id: Int!){community(id: $id){value}}'
+        response = self.gql_client.post(
+            url,
+            json.dumps({'query': query, 'variables': {'id': self.community1.pk}}),
+            content_type='application/json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(json.loads(response.content)['data']['community']['value'], self.community1.value)
+
+    def test_graphql_list(self):
+        url = reverse('graphql')
+        query = '{community_list{value}}'
+        response = self.gql_client.post(
+            url,
+            json.dumps({'query': query}),
+            content_type='application/json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
 
 class PeerGroupTestCase(BaseTestCase):
     def setUp(self):
@@ -160,6 +207,27 @@ def test_update_peer_group(self):
     def test_delete_peer_group(self):
         pass
 
+    def test_graphql(self):
+        url = reverse('graphql')
+        query = 'query peer_group($id: Int!){peer_group(id: $id){name}}'
+        response = self.gql_client.post(
+            url,
+            json.dumps({'query': query, 'variables': {'id': self.peer_group.pk}}),
+            content_type='application/json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(json.loads(response.content)['data']['peer_group']['name'], self.peer_group.name)
+
+    def test_graphql_list(self):
+        url = reverse('graphql')
+        query = '{peer_group_list{name}}'
+        response = self.gql_client.post(
+            url,
+            json.dumps({'query': query}),
+            content_type='application/json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
 
 class SessionTestCase(BaseTestCase):
     def setUp(self):
@@ -251,3 +319,24 @@ def test_session_no_device(self):
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
         self.assertEqual(BGPSession.objects.get(pk=response.data['id']).name, 'test_session')
         self.assertEqual(BGPSession.objects.get(pk=response.data['id']).description, 'session_descr')
+
+    def test_graphql(self):
+        url = reverse('graphql')
+        query = 'query bgp_session($id: Int!){bgp_session(id: $id){name}}'
+        response = self.gql_client.post(
+            url,
+            json.dumps({'query': query, 'variables': {'id': self.session.pk}}),
+            content_type='application/json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(json.loads(response.content)['data']['bgp_session']['name'], self.session.name)
+
+    def test_graphql_list(self):
+        url = reverse('graphql')
+        query = '{bgp_session_list{name}}'
+        response = self.gql_client.post(
+            url,
+            json.dumps({'query': query}),
+            content_type='application/json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)

netbox_bgp/version.py

@@ -1 +1 @@
-__version__ = "0.6.0"
+__version__ = "0.6.1"