Closes #14538 - Add available_at_site filter (#14541)

* Closes #14538 - Add available_at_site filter

* Add tests

* Fix tests

netbox/ipam/filtersets.py

@@ -950,6 +950,10 @@ class VLANFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
         choices=VLANStatusChoices,
         null_value=None
     )
+    available_at_site = django_filters.ModelChoiceFilter(
+        queryset=Site.objects.all(),
+        method='get_for_site'
+    )
     available_on_device = django_filters.ModelChoiceFilter(
         queryset=Device.objects.all(),
         method='get_for_device'
@@ -984,6 +988,10 @@ class VLANFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
             pass
         return queryset.filter(qs_filter)
 
+    @extend_schema_field(OpenApiTypes.STR)
+    def get_for_site(self, queryset, name, value):
+        return queryset.get_for_site(value)
+
     @extend_schema_field(OpenApiTypes.STR)
     def get_for_device(self, queryset, name, value):
         return queryset.get_for_device(value)

netbox/ipam/querysets.py

@@ -69,6 +69,35 @@ class VLANGroupQuerySet(RestrictedQuerySet):
 
 class VLANQuerySet(RestrictedQuerySet):
 
+    def get_for_site(self, site):
+        """
+        Return all VLANs in the specified site
+        """
+        from .models import VLANGroup
+        q = Q()
+        q |= Q(
+            scope_type=ContentType.objects.get_by_natural_key('dcim', 'site'),
+            scope_id=site.pk
+        )
+
+        if site.region:
+            q |= Q(
+                scope_type=ContentType.objects.get_by_natural_key('dcim', 'region'),
+                scope_id__in=site.region.get_ancestors(include_self=True)
+            )
+        if site.group:
+            q |= Q(
+                scope_type=ContentType.objects.get_by_natural_key('dcim', 'sitegroup'),
+                scope_id__in=site.group.get_ancestors(include_self=True)
+            )
+
+        return self.filter(
+            Q(group__in=VLANGroup.objects.filter(q)) |
+            Q(site=site) |
+            Q(group__scope_id__isnull=True, site__isnull=True) |  # Global group VLANs
+            Q(group__isnull=True, site__isnull=True)  # Global VLANs
+        )
+
     def get_for_device(self, device):
         """
         Return all VLANs available to the specified Device.

netbox/ipam/tests/test_filtersets.py

@@ -1359,6 +1359,7 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
             VLANGroup(name='VLAN Group 1', slug='vlan-group-1'),
             VLANGroup(name='VLAN Group 2', slug='vlan-group-2'),
             VLANGroup(name='VLAN Group 3', slug='vlan-group-3'),
+            VLANGroup(name='VLAN Group 4', slug='vlan-group-4'),
         )
         VLANGroup.objects.bulk_create(groups)
 
@@ -1415,6 +1416,9 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
             VLAN(vid=301, name='VLAN 301', site=sites[5], group=groups[23], role=roles[2], tenant=tenants[2], status=VLANStatusChoices.STATUS_RESERVED),
             VLAN(vid=302, name='VLAN 302', site=sites[5], group=groups[23], role=roles[2], tenant=tenants[2], status=VLANStatusChoices.STATUS_RESERVED),
 
+            # Create one globally available VLAN on a VLAN group
+            VLAN(vid=500, name='VLAN Group 1', group=groups[24]),
+
             # Create one globally available VLAN
             VLAN(vid=1000, name='Global VLAN'),
         )
@@ -1488,12 +1492,17 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
     def test_available_on_device(self):
         device_id = Device.objects.first().pk
         params = {'available_on_device': device_id}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6)  # 5 scoped + 1 global
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 7)  # 5 scoped + 1 global group + 1 global
 
     def test_available_on_virtualmachine(self):
         vm_id = VirtualMachine.objects.first().pk
         params = {'available_on_virtualmachine': vm_id}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6)  # 5 scoped + 1 global
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 7)  # 5 scoped + 1 global group + 1 global
+
+    def test_available_at_site(self):
+        site_id = Site.objects.first().pk
+        params = {'available_at_site': site_id}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 5)  # 4 scoped + 1 global group + 1 global
 
 
 class ServiceTemplateTestCase(TestCase, ChangeLoggedFilterSetTests):