Explorar el Código

fix(filtersets): Handle null in several filters

Fourteen model choice filters with a custom method passed the null
value to that method as a string, which raised a server error. They
now reject it as an invalid value. The data source sync_interval
filter now matches null as no interval and rejects None.

Fixes #23574
Martin Hauser hace 1 día
padre
commit
733c5960f7

+ 0 - 1
netbox/core/filtersets.py

@@ -37,7 +37,6 @@ class DataSourceFilterSet(PrimaryModelFilterSet):
     sync_interval = django_filters.MultipleChoiceFilter(
     sync_interval = django_filters.MultipleChoiceFilter(
         choices=JobIntervalChoices,
         choices=JobIntervalChoices,
         distinct=False,
         distinct=False,
-        null_value=None
     )
     )
 
 
     class Meta:
     class Meta:

+ 14 - 1
netbox/core/tests/test_filtersets.py

@@ -46,7 +46,6 @@ class DataSourceTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
                 source_url='https://example.com/git/source3',
                 source_url='https://example.com/git/source3',
                 status=DataSourceStatusChoices.COMPLETED,
                 status=DataSourceStatusChoices.COMPLETED,
                 enabled=False,
                 enabled=False,
-                sync_interval=JobIntervalChoices.INTERVAL_WEEKLY
             ),
             ),
         )
         )
         DataSource.objects.bulk_create(data_sources)
         DataSource.objects.bulk_create(data_sources)
@@ -81,6 +80,20 @@ class DataSourceTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'sync_interval': [JobIntervalChoices.INTERVAL_HOURLY, JobIntervalChoices.INTERVAL_DAILY]}
         params = {'sync_interval': [JobIntervalChoices.INTERVAL_HOURLY, JobIntervalChoices.INTERVAL_DAILY]}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
 
 
+        # The null choice value matches data sources without a sync interval
+        for params, count in (
+            ({'sync_interval': ['null']}, 1),
+            ({'sync_interval__n': ['null']}, 2),
+            ({'sync_interval': [JobIntervalChoices.INTERVAL_HOURLY, 'null']}, 2),
+        ):
+            filterset = self.filterset(params, self.queryset)
+            self.assertTrue(filterset.is_valid(), msg=params)
+            self.assertEqual(filterset.qs.count(), count, msg=params)
+
+        # Reject invalid filter values
+        for params in ({'sync_interval': ['None']}, {'sync_interval__n': ['None']}):
+            self.assertFalse(self.filterset(params, self.queryset).is_valid(), msg=params)
+
 
 
 class DataFileTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
 class DataFileTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
     queryset = DataFile.objects.all()
     queryset = DataFile.objects.all()

+ 1 - 0
netbox/dcim/filtersets.py

@@ -1353,6 +1353,7 @@ class PlatformFilterSet(NestedGroupModelFilterSet):
     )
     )
     available_for_device_type = django_filters.ModelChoiceFilter(
     available_for_device_type = django_filters.ModelChoiceFilter(
         queryset=DeviceType.objects.all(),
         queryset=DeviceType.objects.all(),
+        null_label=None,
         method='get_for_device_type'
         method='get_for_device_type'
     )
     )
     config_template_id = django_filters.ModelMultipleChoiceFilter(
     config_template_id = django_filters.ModelMultipleChoiceFilter(

+ 4 - 0
netbox/dcim/tests/test_filtersets.py

@@ -3009,6 +3009,10 @@ class PlatformTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'available_for_device_type': device_type.pk}
         params = {'available_for_device_type': device_type.pk}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
 
+        # Reject invalid filter values
+        params = {'available_for_device_type': 'null'}
+        self.assertFalse(self.filterset(params, self.queryset).is_valid())
+
 
 
 class DeviceTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
 class DeviceTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
     queryset = Device.objects.all()
     queryset = Device.objects.all()

+ 13 - 0
netbox/ipam/filtersets.py

@@ -382,11 +382,13 @@ class PrefixFilterSet(PrimaryModelFilterSet, ScopedFilterSet, TenancyFilterSet,
     )
     )
     present_in_vrf_id = django_filters.ModelChoiceFilter(
     present_in_vrf_id = django_filters.ModelChoiceFilter(
         queryset=VRF.objects.all(),
         queryset=VRF.objects.all(),
+        null_label=None,
         method='filter_present_in_vrf_id',
         method='filter_present_in_vrf_id',
         label=_('VRF')
         label=_('VRF')
     )
     )
     present_in_vrf = django_filters.ModelChoiceFilter(
     present_in_vrf = django_filters.ModelChoiceFilter(
         queryset=VRF.objects.all(),
         queryset=VRF.objects.all(),
+        null_label=None,
         method='filter_present_in_vrf',
         method='filter_present_in_vrf',
         to_field_name='rd',
         to_field_name='rd',
         label=_('VRF (RD)'),
         label=_('VRF (RD)'),
@@ -646,11 +648,13 @@ class IPAddressFilterSet(PrimaryModelFilterSet, TenancyFilterSet, ContactModelFi
     )
     )
     present_in_vrf_id = django_filters.ModelChoiceFilter(
     present_in_vrf_id = django_filters.ModelChoiceFilter(
         queryset=VRF.objects.all(),
         queryset=VRF.objects.all(),
+        null_label=None,
         method='filter_present_in_vrf_id',
         method='filter_present_in_vrf_id',
         label=_('VRF')
         label=_('VRF')
     )
     )
     present_in_vrf = django_filters.ModelChoiceFilter(
     present_in_vrf = django_filters.ModelChoiceFilter(
         queryset=VRF.objects.all(),
         queryset=VRF.objects.all(),
+        null_label=None,
         method='filter_present_in_vrf',
         method='filter_present_in_vrf',
         to_field_name='rd',
         to_field_name='rd',
         label=_('VRF (RD)'),
         label=_('VRF (RD)'),
@@ -844,6 +848,7 @@ class FHRPGroupFilterSet(PrimaryModelFilterSet):
     )
     )
     related_ip = django_filters.ModelMultipleChoiceFilter(
     related_ip = django_filters.ModelMultipleChoiceFilter(
         queryset=IPAddress.objects.all(),
         queryset=IPAddress.objects.all(),
+        null_label=None,
         method='filter_related_ip'
         method='filter_related_ip'
     )
     )
 
 
@@ -1033,12 +1038,14 @@ class VLANFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
     )
     )
     related_to_site = django_filters.ModelMultipleChoiceFilter(
     related_to_site = django_filters.ModelMultipleChoiceFilter(
         queryset=Site.objects.all(),
         queryset=Site.objects.all(),
+        null_label=None,
         method='filter_related_to_site',
         method='filter_related_to_site',
         label=_('Related to site (ID)'),
         label=_('Related to site (ID)'),
     )
     )
     # get_additional_lookups() skips filters with a method.
     # get_additional_lookups() skips filters with a method.
     related_to_site__n = django_filters.ModelMultipleChoiceFilter(
     related_to_site__n = django_filters.ModelMultipleChoiceFilter(
         queryset=Site.objects.all(),
         queryset=Site.objects.all(),
+        null_label=None,
         method='filter_related_to_site_negated',
         method='filter_related_to_site_negated',
         label=_('Related to site (ID)'),
         label=_('Related to site (ID)'),
     )
     )
@@ -1073,18 +1080,22 @@ class VLANFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
     )
     )
     available_at_site = django_filters.ModelChoiceFilter(
     available_at_site = django_filters.ModelChoiceFilter(
         queryset=Site.objects.all(),
         queryset=Site.objects.all(),
+        null_label=None,
         method='get_for_site'
         method='get_for_site'
     )
     )
     available_at_site_group = django_filters.ModelChoiceFilter(
     available_at_site_group = django_filters.ModelChoiceFilter(
         queryset=SiteGroup.objects.all(),
         queryset=SiteGroup.objects.all(),
+        null_label=None,
         method='get_for_site_group'
         method='get_for_site_group'
     )
     )
     available_on_device = django_filters.ModelChoiceFilter(
     available_on_device = django_filters.ModelChoiceFilter(
         queryset=Device.objects.select_related('cluster'),
         queryset=Device.objects.select_related('cluster'),
+        null_label=None,
         method='get_for_device'
         method='get_for_device'
     )
     )
     available_on_virtualmachine = django_filters.ModelChoiceFilter(
     available_on_virtualmachine = django_filters.ModelChoiceFilter(
         queryset=VirtualMachine.objects.all(),
         queryset=VirtualMachine.objects.all(),
+        null_label=None,
         method='get_for_virtualmachine'
         method='get_for_virtualmachine'
     )
     )
     qinq_role = django_filters.MultipleChoiceFilter(
     qinq_role = django_filters.MultipleChoiceFilter(
@@ -1113,11 +1124,13 @@ class VLANFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
     )
     )
     interface_id = django_filters.ModelChoiceFilter(
     interface_id = django_filters.ModelChoiceFilter(
         queryset=Interface.objects.all(),
         queryset=Interface.objects.all(),
+        null_label=None,
         method='filter_interface_id',
         method='filter_interface_id',
         label=_('Assigned interface')
         label=_('Assigned interface')
     )
     )
     vminterface_id = django_filters.ModelChoiceFilter(
     vminterface_id = django_filters.ModelChoiceFilter(
         queryset=VMInterface.objects.all(),
         queryset=VMInterface.objects.all(),
+        null_label=None,
         method='filter_vminterface_id',
         method='filter_vminterface_id',
         label=_('Assigned VM interface')
         label=_('Assigned VM interface')
     )
     )

+ 56 - 0
netbox/ipam/tests/test_filtersets.py

@@ -862,6 +862,10 @@ class PrefixTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         self.assertEqual(self.filterset({'present_in_vrf': vrf1.rd}, self.queryset).qs.count(), 6)
         self.assertEqual(self.filterset({'present_in_vrf': vrf1.rd}, self.queryset).qs.count(), 6)
         self.assertEqual(self.filterset({'present_in_vrf': vrf2.rd}, self.queryset).qs.count(), 2)
         self.assertEqual(self.filterset({'present_in_vrf': vrf2.rd}, self.queryset).qs.count(), 2)
 
 
+        # Reject invalid filter values
+        for params in ({'present_in_vrf_id': 'null'}, {'present_in_vrf': 'null'}):
+            self.assertFalse(self.filterset(params, self.queryset).is_valid(), msg=params)
+
     def test_region(self):
     def test_region(self):
         regions = Region.objects.all()[:2]
         regions = Region.objects.all()[:2]
         params = {'region_id': [regions[0].pk, regions[1].pk]}
         params = {'region_id': [regions[0].pk, regions[1].pk]}
@@ -1448,6 +1452,22 @@ class IPAddressTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'vrf': [vrfs[0].rd, vrfs[1].rd]}
         params = {'vrf': [vrfs[0].rd, vrfs[1].rd]}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
 
+    def test_present_in_vrf(self):
+        """Match addresses in the VRF or in a VRF exporting one of its import targets."""
+        vrf1 = VRF.objects.get(name='VRF 1')
+        vrf2 = VRF.objects.get(name='VRF 2')
+        route_target = RouteTarget.objects.create(name='65000:200')
+        vrf1.import_targets.add(route_target)
+        vrf2.export_targets.add(route_target)
+        self.assertEqual(self.filterset({'present_in_vrf_id': vrf1.pk}, self.queryset).qs.count(), 4)
+        self.assertEqual(self.filterset({'present_in_vrf_id': vrf2.pk}, self.queryset).qs.count(), 2)
+        self.assertEqual(self.filterset({'present_in_vrf': vrf1.rd}, self.queryset).qs.count(), 4)
+        self.assertEqual(self.filterset({'present_in_vrf': vrf2.rd}, self.queryset).qs.count(), 2)
+
+        # Reject invalid filter values
+        for params in ({'present_in_vrf_id': 'null'}, {'present_in_vrf': 'null'}):
+            self.assertFalse(self.filterset(params, self.queryset).is_valid(), msg=params)
+
     def test_device(self):
     def test_device(self):
         devices = Device.objects.all()[:2]
         devices = Device.objects.all()[:2]
         params = {'device_id': [devices[0].pk, devices[1].pk]}
         params = {'device_id': [devices[0].pk, devices[1].pk]}
@@ -1620,6 +1640,10 @@ class FHRPGroupTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'related_ip': [ipaddresses[0].pk, ipaddresses[1].pk]}
         params = {'related_ip': [ipaddresses[0].pk, ipaddresses[1].pk]}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
 
 
+        # Reject invalid filter values
+        for params in ({'related_ip': ['null']}, {'related_ip': [ipaddresses[0].pk, 'null']}):
+            self.assertFalse(self.filterset(params, self.queryset).is_valid(), msg=params)
+
 
 
 class FHRPGroupAssignmentTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
 class FHRPGroupAssignmentTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
     queryset = FHRPGroupAssignment.objects.all()
     queryset = FHRPGroupAssignment.objects.all()
@@ -2296,6 +2320,10 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
             [2001, 2002, 2003, 3001, 3002, 3003]
             [2001, 2002, 2003, 3001, 3002, 3003]
         )
         )
 
 
+        # Reject invalid filter values
+        for params in ({'related_to_site': ['null']}, {'related_to_site': [site.pk, 'null']}):
+            self.assertFalse(self.filterset(params, self.queryset).is_valid(), msg=params)
+
     def test_related_to_site_matches_any_of_several_sites(self):
     def test_related_to_site_matches_any_of_several_sites(self):
         """Several sites spanning two site groups are ORed together, and a shared VLAN comes back once."""
         """Several sites spanning two site groups are ORed together, and a shared VLAN comes back once."""
         sites = (
         sites = (
@@ -2365,6 +2393,10 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'related_to_site__n': [site.pk]}
         params = {'related_to_site__n': [site.pk]}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), self.queryset.count() - 2)
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), self.queryset.count() - 2)
 
 
+        # Reject invalid filter values
+        for params in ({'related_to_site__n': ['null']}, {'related_to_site__n': [site.pk, 'null']}):
+            self.assertFalse(self.filterset(params, self.queryset).is_valid(), msg=params)
+
     def test_related_to_site_resolves_site_groups_in_constant_queries(self):
     def test_related_to_site_resolves_site_groups_in_constant_queries(self):
         """The query count is constant in the number of sites, and drops to one without groups."""
         """The query count is constant in the number of sites, and drops to one without groups."""
         sites = list(Site.objects.filter(slug__in=('site-1', 'site-2', 'site-3')))
         sites = list(Site.objects.filter(slug__in=('site-1', 'site-2', 'site-3')))
@@ -2459,6 +2491,10 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'available_on_device': device_id}
         params = {'available_on_device': device_id}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 7)  # 5 scoped + 1 global group + 1 global
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 7)  # 5 scoped + 1 global group + 1 global
 
 
+        # Reject invalid filter values
+        params = {'available_on_device': 'null'}
+        self.assertFalse(self.filterset(params, self.queryset).is_valid())
+
     def test_available_on_device_cluster_scopes(self):
     def test_available_on_device_cluster_scopes(self):
         device = Device.objects.get(name='Device 1')
         device = Device.objects.get(name='Device 1')
         device.cluster = Cluster.objects.get(name='Cluster 1')
         device.cluster = Cluster.objects.get(name='Cluster 1')
@@ -2487,16 +2523,28 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'available_on_virtualmachine': vm_id}
         params = {'available_on_virtualmachine': vm_id}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 8)  # 6 scoped + 1 global group + 1 global
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 8)  # 6 scoped + 1 global group + 1 global
 
 
+        # Reject invalid filter values
+        params = {'available_on_virtualmachine': 'null'}
+        self.assertFalse(self.filterset(params, self.queryset).is_valid())
+
     def test_available_at_site(self):
     def test_available_at_site(self):
         site_id = Site.objects.first().pk
         site_id = Site.objects.first().pk
         params = {'available_at_site': site_id}
         params = {'available_at_site': site_id}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 5)  # 4 scoped + 1 global group + 1 global
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 5)  # 4 scoped + 1 global group + 1 global
 
 
+        # Reject invalid filter values
+        params = {'available_at_site': 'null'}
+        self.assertFalse(self.filterset(params, self.queryset).is_valid())
+
     def test_available_at_site_group(self):
     def test_available_at_site_group(self):
         site_group = SiteGroup.objects.get(name='Site Group 1')
         site_group = SiteGroup.objects.get(name='Site Group 1')
         params = {'available_at_site_group': site_group.pk}
         params = {'available_at_site_group': site_group.pk}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)  # 1 scoped + 1 global group + 1 global
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)  # 1 scoped + 1 global group + 1 global
 
 
+        # Reject invalid filter values
+        params = {'available_at_site_group': 'null'}
+        self.assertFalse(self.filterset(params, self.queryset).is_valid())
+
     def test_interface(self):
     def test_interface(self):
         interface_id = Interface.objects.first().pk
         interface_id = Interface.objects.first().pk
         params = {'interface_id': interface_id}
         params = {'interface_id': interface_id}
@@ -2522,6 +2570,10 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         self.assertEqual(qs.count(), 2)
         self.assertEqual(qs.count(), 2)
         self.assertEqual(len(qs), len(set(qs.values_list('pk', flat=True))))
         self.assertEqual(len(qs), len(set(qs.values_list('pk', flat=True))))
 
 
+        # Reject invalid filter values
+        params = {'interface_id': 'null'}
+        self.assertFalse(self.filterset(params, self.queryset).is_valid())
+
     def test_vminterface(self):
     def test_vminterface(self):
         vminterface_id = VMInterface.objects.first().pk
         vminterface_id = VMInterface.objects.first().pk
         params = {'vminterface_id': vminterface_id}
         params = {'vminterface_id': vminterface_id}
@@ -2546,6 +2598,10 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         self.assertEqual(qs.count(), 2)
         self.assertEqual(qs.count(), 2)
         self.assertEqual(len(qs), len(set(qs.values_list('pk', flat=True))))
         self.assertEqual(len(qs), len(set(qs.values_list('pk', flat=True))))
 
 
+        # Reject invalid filter values
+        params = {'vminterface_id': 'null'}
+        self.assertFalse(self.filterset(params, self.queryset).is_valid())
+
     def test_qinq_role(self):
     def test_qinq_role(self):
         params = {'qinq_role': [VLANQinQRoleChoices.ROLE_SERVICE, VLANQinQRoleChoices.ROLE_CUSTOMER]}
         params = {'qinq_role': [VLANQinQRoleChoices.ROLE_SERVICE, VLANQinQRoleChoices.ROLE_CUSTOMER]}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6)
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6)