Merge pull request #49 from k01ek/develop

0.3.7

Author: k01ek

netbox_bgp/api/serializers.py

@@ -1,4 +1,4 @@
-from rest_framework.serializers import Serializer, HyperlinkedIdentityField
+from rest_framework.serializers import Serializer, HyperlinkedIdentityField, ValidationError
 from rest_framework.relations import PrimaryKeyRelatedField
 
 from netbox.api import ChoiceField, WritableNestedSerializer, ValidatedModelSerializer
@@ -63,6 +63,13 @@ class ASNSerializer(TaggedObjectSerializer, CustomFieldModelSerializer):
     site = NestedSiteSerializer(required=False, allow_null=True)
     tenant = NestedTenantSerializer(required=False, allow_null=True)
 
+    def validate(self, attrs):
+        if ASN.objects.filter(number=attrs['number'], tenant=attrs.get('tenant')).exists():
+            raise ValidationError(
+                {'error': 'Asn with this Number and Tenant already exists.'}
+            )
+        return attrs
+
     class Meta:
         model = ASN
         fields = ['number', 'id', 'status', 'description', 'custom_fields', 'site', 'tenant', 'tags']

netbox_bgp/filters.py

@@ -1,6 +1,7 @@
 import django_filters
+import netaddr
 from django.db.models import Q
-
+from netaddr.core import AddrFormatError
 # With netbox v2.11.3 TagFilter was moved to extras
 try:
     from utilities.filters import TagFilter
@@ -122,7 +123,6 @@ class BGPSessionFilterSet(django_filters.FilterSet):
         to_field_name='address',
         label='Remote Address',
     )
-
     device_id = django_filters.ModelMultipleChoiceFilter(
         field_name='device__id',
         queryset=Device.objects.all(),
@@ -135,6 +135,14 @@ class BGPSessionFilterSet(django_filters.FilterSet):
         to_field_name='name',
         label='Device (name)',
     )
+    by_remote_address = django_filters.CharFilter(
+        method='search_by_remote_ip',
+        label='Remote Address',
+    )
+    by_local_address = django_filters.CharFilter(
+        method='search_by_local_ip',
+        label='Local Address',
+    )
 
     class Meta:
         model = BGPSession
@@ -152,6 +160,24 @@ def search(self, queryset, name, value):
         )
         return queryset.filter(qs_filter)
 
+    def search_by_remote_ip(self, queryset, name, value):
+        if not value.strip():
+            return queryset
+        try:
+            query = str(netaddr.IPNetwork(value).cidr)
+            return queryset.filter(remote_address__address=query)
+        except (AddrFormatError, ValueError):
+            return queryset.none()
+
+    def search_by_local_ip(self, queryset, name, value):
+        if not value.strip():
+            return queryset
+        try:
+            query = str(netaddr.IPNetwork(value).cidr)
+            return queryset.filter(local_address__address=query)
+        except (AddrFormatError, ValueError):
+            return queryset.none()
+
 
 class RoutingPolicyFilterSet(django_filters.FilterSet):
     q = django_filters.CharFilter(

netbox_bgp/forms.py

@@ -1,5 +1,6 @@
 from django import forms
 from django.core.exceptions import MultipleObjectsReturned, ObjectDoesNotExist
+from django.utils.translation import gettext as _
 
 from extras.models import Tag
 from tenancy.models import Tenant
@@ -294,9 +295,18 @@ class BGPSessionFilterForm(BootstrapMixin, CustomFieldModelForm):
             api_url='/api/plugins/bgp/asn/',
         )
     )
-    tenant = DynamicModelChoiceField(
-        queryset=Tenant.objects.all(),
-        required=False
+    by_local_address = forms.CharField(
+        required=False,
+        label='Local Address'
+    )
+    by_remote_address = forms.CharField(
+        required=False,
+        label='Remote Address'
+    )
+    device_id = DynamicModelMultipleChoiceField(
+        queryset=Device.objects.all(),
+        required=False,
+        label=_('Device')
     )
     status = forms.MultipleChoiceField(
         choices=SessionStatusChoices,
@@ -324,12 +334,16 @@ class BGPSessionFilterForm(BootstrapMixin, CustomFieldModelForm):
             api_url='/api/plugins/bgp/routing-policy/'
         )
     )
+    tenant = DynamicModelChoiceField(
+        queryset=Tenant.objects.all(),
+        required=False
+    )
 
     tag = TagFilterField(BGPSession)
 
     class Meta:
         model = BGPSession
-        fields = ['q', 'status', 'tenant', 'remote_as', 'local_as']
+        fields = ['q', 'status', 'device_id', 'remote_as', 'local_as']
 
 
 class RoutingPolicyFilterForm(BootstrapMixin, CustomFieldModelForm):

netbox_bgp/tables.py

@@ -14,6 +14,14 @@
  {% endif %}
  """
 
+POLICIES = """
+{% for rp in value.all %}
+    <a href="{{ rp.get_absolute_url }}">{{ rp }}</a>{% if not forloop.last %}<br />{% endif %}
+{% empty %}
+    &mdash;
+{% endfor %}
+"""
+
 
 class ASNTable(BaseTable):
     pk = ToggleColumn()
@@ -95,7 +103,24 @@ class Meta(BaseTable.Meta):
 class BGPPeerGroupTable(BaseTable):
     pk = ToggleColumn()
     name = tables.LinkColumn()
+    import_policies = tables.TemplateColumn(
+        template_code=POLICIES,
+        orderable=False
+    )
+    export_policies = tables.TemplateColumn(
+        template_code=POLICIES,
+        orderable=False
+    )
+    tags = TagColumn(
+        url_name='plugins:netbox_bgp:peer_group_list'
+    )
 
     class Meta(BaseTable.Meta):
         model = BGPPeerGroup
-        fields = ('pk', 'name', 'description')
+        fields = (
+            'pk', 'name', 'description', 'tags',
+            'import_policies', 'export_policies'
+        )
+        default_columns = (
+            'pk', 'name', 'description'
+        )

netbox_bgp/tests/test_api.py

@@ -7,6 +7,7 @@
 
 from users.models import Token
 
+from tenancy.models import Tenant
 from netbox_bgp.models import ASN, Community
 
 
@@ -24,6 +25,7 @@ def setUp(self):
         self.base_url_lookup = 'plugins-api:netbox_bgp-api:asn'
         self.asn1 = ASN.objects.create(number=65000, description='test_asn1')
         self.asn2 = ASN.objects.create(number=65001, description='test_asn2')
+        self.tenant = Tenant.objects.create(name='tenant')
 
     def test_list_asn(self):
         url = reverse(f'{self.base_url_lookup}-list')
@@ -72,6 +74,16 @@ def test_delete_task(self):
         with self.assertRaises(ASN.DoesNotExist):
             ASN.objects.get(pk=self.asn1.pk)
 
+    def test_uniqueconstraint_asn(self):
+        url = reverse(f'{self.base_url_lookup}-list')
+        data = {'number': 65001, 'description': 'test_asn3'}
+        response = self.client.post(url, data, format='json')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+        data = {'number': 65001, 'description': 'test_asn3', 'tenant': self.tenant.pk}
+        response = self.client.post(url, data, format='json')
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
 
 class CommunityTestCase(BaseTestCase):
     def setUp(self):

netbox_bgp/version.py

@@ -1 +1 @@
-__version__ = "0.3.6"
+__version__ = "0.3.7"