NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-08-22 23:46:44 -06:00
Code Issues Actions 9 Packages Projects Releases Wiki Activity

Implement interface filters for TunnelTermination

Browse Source
This commit is contained in:
Jeremy Stretch 2023-11-27 13:27:29 -05:00
parent dda4071b3a
commit 21f1de4fa0
4 changed files with 87 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
netbox/dcim/models/device_components.py
View File

@ -729,6 +729,12 @@ class Interface(ModularComponentModel, BaseInterface, CabledObjectModel, PathEnd
object_id_field='interface_id', object_id_field='interface_id',
related_query_name='+' related_query_name='+'
) )
tunnel_terminations = GenericRelation(
to='vpn.TunnelTermination',
content_type_field='termination_type',
object_id_field='termination_id',
related_query_name='interface'
)
l2vpn_terminations = GenericRelation( l2vpn_terminations = GenericRelation(
to='ipam.L2VPNTermination', to='ipam.L2VPNTermination',
content_type_field='assigned_object_type', content_type_field='assigned_object_type',

6
netbox/virtualization/models/virtualmachines.py
View File

@ -291,6 +291,12 @@ class VMInterface(NetBoxModel, BaseInterface, TrackingModelMixin):
object_id_field='interface_id', object_id_field='interface_id',
related_query_name='+' related_query_name='+'
) )
tunnel_terminations = GenericRelation(
to='vpn.TunnelTermination',
content_type_field='termination_type',
object_id_field='termination_id',
related_query_name='vminterface',
)
l2vpn_terminations = GenericRelation( l2vpn_terminations = GenericRelation(
to='ipam.L2VPNTermination', to='ipam.L2VPNTermination',
content_type_field='assigned_object_type', content_type_field='assigned_object_type',

44
netbox/vpn/filtersets.py
View File

@ -69,28 +69,28 @@ class TunnelTerminationFilterSet(NetBoxModelFilterSet):
role = django_filters.MultipleChoiceFilter( role = django_filters.MultipleChoiceFilter(
choices=TunnelTerminationRoleChoices choices=TunnelTerminationRoleChoices
) )
# interface = django_filters.ModelMultipleChoiceFilter( interface = django_filters.ModelMultipleChoiceFilter(
# field_name='interface__name', field_name='interface__name',
# queryset=Interface.objects.all(), queryset=Interface.objects.all(),
# to_field_name='name', to_field_name='name',
# label=_('Interface (name)'), label=_('Interface (name)'),
# ) )
# interface_id = django_filters.ModelMultipleChoiceFilter( interface_id = django_filters.ModelMultipleChoiceFilter(
# field_name='interface', field_name='interface',
# queryset=Interface.objects.all(), queryset=Interface.objects.all(),
# label=_('Interface (ID)'), label=_('Interface (ID)'),
# ) )
# vminterface = django_filters.ModelMultipleChoiceFilter( vminterface = django_filters.ModelMultipleChoiceFilter(
# field_name='interface__name', field_name='vminterface__name',
# queryset=VMInterface.objects.all(), queryset=VMInterface.objects.all(),
# to_field_name='name', to_field_name='name',
# label=_('VM interface (name)'), label=_('VM interface (name)'),
# ) )
# vminterface_id = django_filters.ModelMultipleChoiceFilter( vminterface_id = django_filters.ModelMultipleChoiceFilter(
# field_name='vminterface', field_name='vminterface',
# queryset=VMInterface.objects.all(), queryset=VMInterface.objects.all(),
# label=_('VM interface (ID)'), label=_('VM interface (ID)'),
# ) )
outside_ip_id = django_filters.ModelMultipleChoiceFilter( outside_ip_id = django_filters.ModelMultipleChoiceFilter(
field_name='outside_ip', field_name='outside_ip',
queryset=IPAddress.objects.all(), queryset=IPAddress.objects.all(),

59
netbox/vpn/tests/test_filtersets.py
View File

@ -3,10 +3,11 @@ from django.test import TestCase
from dcim.choices import InterfaceTypeChoices from dcim.choices import InterfaceTypeChoices
from dcim.models import Interface from dcim.models import Interface
from ipam.models import IPAddress from ipam.models import IPAddress
from virtualization.models import VMInterface
from vpn.choices import * from vpn.choices import *
from vpn.filtersets import * from vpn.filtersets import *
from vpn.models import * from vpn.models import *
from utilities.testing import ChangeLoggedFilterSetTests, create_test_device from utilities.testing import ChangeLoggedFilterSetTests, create_test_device, create_test_virtualmachine
class TunnelTestCase(TestCase, ChangeLoggedFilterSetTests): class TunnelTestCase(TestCase, ChangeLoggedFilterSetTests):
@ -117,10 +118,21 @@ class TunnelTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
) )
Interface.objects.bulk_create(interfaces) Interface.objects.bulk_create(interfaces)
virtual_machine = create_test_virtualmachine('Virtual Machine 1')
vm_interfaces = (
VMInterface(virtual_machine=virtual_machine, name='Interface 1'),
VMInterface(virtual_machine=virtual_machine, name='Interface 2'),
VMInterface(virtual_machine=virtual_machine, name='Interface 3'),
)
VMInterface.objects.bulk_create(vm_interfaces)
ip_addresses = ( ip_addresses = (
IPAddress(address='192.168.0.1/32'), IPAddress(address='192.168.0.1/32'),
IPAddress(address='192.168.0.2/32'), IPAddress(address='192.168.0.2/32'),
IPAddress(address='192.168.0.3/32'), IPAddress(address='192.168.0.3/32'),
IPAddress(address='192.168.0.4/32'),
IPAddress(address='192.168.0.5/32'),
IPAddress(address='192.168.0.6/32'),
) )
IPAddress.objects.bulk_create(ip_addresses) IPAddress.objects.bulk_create(ip_addresses)
@ -144,6 +156,7 @@ class TunnelTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
Tunnel.objects.bulk_create(tunnels) Tunnel.objects.bulk_create(tunnels)
tunnel_terminations = ( tunnel_terminations = (
# Tunnel 1
TunnelTermination( TunnelTermination(
tunnel=tunnels[0], tunnel=tunnels[0],
role=TunnelTerminationRoleChoices.ROLE_HUB, role=TunnelTerminationRoleChoices.ROLE_HUB,
@ -151,16 +164,36 @@ class TunnelTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
outside_ip=ip_addresses[0] outside_ip=ip_addresses[0]
), ),
TunnelTermination( TunnelTermination(
tunnel=tunnels[1], tunnel=tunnels[0],
role=TunnelTerminationRoleChoices.ROLE_SPOKE, role=TunnelTerminationRoleChoices.ROLE_SPOKE,
termination=interfaces[1], termination=vm_interfaces[0],
outside_ip=ip_addresses[1] outside_ip=ip_addresses[1]
), ),
# Tunnel 2
TunnelTermination(
tunnel=tunnels[1],
role=TunnelTerminationRoleChoices.ROLE_HUB,
termination=interfaces[1],
outside_ip=ip_addresses[2]
),
TunnelTermination(
tunnel=tunnels[1],
role=TunnelTerminationRoleChoices.ROLE_SPOKE,
termination=vm_interfaces[1],
outside_ip=ip_addresses[3]
),
# Tunnel 3
TunnelTermination( TunnelTermination(
tunnel=tunnels[2], tunnel=tunnels[2],
role=TunnelTerminationRoleChoices.ROLE_PEER, role=TunnelTerminationRoleChoices.ROLE_PEER,
termination=interfaces[2], termination=interfaces[2],
outside_ip=ip_addresses[2] outside_ip=ip_addresses[4]
),
TunnelTermination(
tunnel=tunnels[2],
role=TunnelTerminationRoleChoices.ROLE_PEER,
termination=vm_interfaces[2],
outside_ip=ip_addresses[5]
), ),
) )
TunnelTermination.objects.bulk_create(tunnel_terminations) TunnelTermination.objects.bulk_create(tunnel_terminations)
@ -168,12 +201,26 @@ class TunnelTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
def test_tunnel(self): def test_tunnel(self):
tunnels = Tunnel.objects.all()[:2] tunnels = Tunnel.objects.all()[:2]
params = {'tunnel_id': [tunnels[0].pk, tunnels[1].pk]} params = {'tunnel_id': [tunnels[0].pk, tunnels[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
params = {'tunnel': [tunnels[0].name, tunnels[1].name]} params = {'tunnel': [tunnels[0].name, tunnels[1].name]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
def test_role(self): def test_role(self):
params = {'role': [TunnelTerminationRoleChoices.ROLE_HUB, TunnelTerminationRoleChoices.ROLE_SPOKE]} params = {'role': [TunnelTerminationRoleChoices.ROLE_HUB, TunnelTerminationRoleChoices.ROLE_SPOKE]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
def test_interface(self):
interfaces = Interface.objects.all()[:2]
params = {'interface_id': [interfaces[0].pk, interfaces[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'interface': [interfaces[0].name, interfaces[1].name]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_vminterface(self):
vm_interfaces = VMInterface.objects.all()[:2]
params = {'vminterface_id': [vm_interfaces[0].pk, vm_interfaces[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'vminterface': [vm_interfaces[0].name, vm_interfaces[1].name]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_outside_ip(self): def test_outside_ip(self):