Merge pull request #23189 from netbox-community/22886-display-vlans-assigned-to-a-site-via-vlan-groups

Closes #22886: Include VLANs assigned through VLAN Groups in Site related objects

netbox/dcim/tests/test_views.py

@@ -22,7 +22,7 @@ from dcim.constants import *
 from dcim.models import *
 from dcim.views import DeviceTypeListView, ModuleTypeListView
 from extras.models import ConfigContext, ConfigTemplate
-from ipam.models import ASN, RIR, VLAN, VRF
+from ipam.models import ASN, RIR, VLAN, VRF, VLANGroup
 from netbox.choices import (
     CSVDelimiterChoices,
     DiameterUnitChoices,
@@ -232,6 +232,84 @@ class SiteTestCase(ViewTestCases.PrimaryObjectViewTestCase):
             with self.subTest(panel=panel):
                 self.assertNotContains(response, url)
 
+    def test_get_object_vlan_row_covers_group_scopes(self):
+        """The VLANs row counts direct, site-scoped and site-group-scoped VLANs and links to the filter."""
+        self.add_permissions('dcim.view_site', 'ipam.view_vlan')
+        site = Site.objects.get(slug='site-1')
+
+        direct_vlan = VLAN.objects.create(vid=100, name='Direct', site=site)
+        site_scoped_group = VLANGroup.objects.create(name='Site scope', slug='site-scope', scope=site)
+        site_vlan = VLAN.objects.create(vid=200, name='Site scoped', group=site_scoped_group)
+        group_scoped_group = VLANGroup.objects.create(
+            name='Site group scope', slug='site-group-scope', scope=site.group
+        )
+        group_vlan = VLAN.objects.create(vid=300, name='Site group scoped', group=group_scoped_group)
+        VLAN.objects.create(vid=400, name='Unrelated')
+
+        response = self.client.get(site.get_absolute_url())
+        self.assertHttpStatus(response, 200)
+
+        rows = [row for row in response.context['related_models'] if row.queryset.model is VLAN]
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0].filter_param, 'related_to_site')
+        self.assertEqual(set(rows[0].queryset), {direct_vlan, site_vlan, group_vlan})
+
+        list_url = f"{reverse('ipam:vlan_list')}?related_to_site={site.pk}"
+        self.assertContains(response, list_url)
+
+        list_response = self.client.get(list_url)
+        self.assertHttpStatus(list_response, 200)
+        self.assertEqual(
+            {vlan.pk for vlan in list_response.context['table'].data},
+            {direct_vlan.pk, site_vlan.pk, group_vlan.pk}
+        )
+
+    def test_get_object_vlan_row_without_direct_site_assignment(self):
+        """The VLANs row appears when a site's only VLANs arrive through a group scope."""
+        self.add_permissions('dcim.view_site', 'ipam.view_vlan')
+        site = Site.objects.get(slug='site-3')
+
+        group = VLANGroup.objects.create(name='Site 3 scope', slug='site-3-scope', scope=site)
+        vlan = VLAN.objects.create(vid=500, name='Group only', group=group)
+
+        response = self.client.get(site.get_absolute_url())
+        self.assertHttpStatus(response, 200)
+
+        rows = [row for row in response.context['related_models'] if row.queryset.model is VLAN]
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(list(rows[0].queryset), [vlan])
+
+    def test_get_object_vlan_row_respects_constrained_permissions(self):
+        """A VLAN constraint narrows both the related objects count and the linked list."""
+        self.add_permissions('dcim.view_site')
+        site = Site.objects.get(slug='site-1')
+        tenant = Tenant.objects.create(name='Visible', slug='visible')
+
+        group = VLANGroup.objects.create(name='Site scope', slug='site-scope', scope=site)
+        visible_vlan = VLAN.objects.create(vid=200, name='Visible', group=group, tenant=tenant)
+        VLAN.objects.create(vid=201, name='Hidden', group=group)
+
+        obj_perm = ObjectPermission(
+            name='Visible VLANs',
+            actions=['view'],
+            constraints={'tenant__slug': 'visible'}
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(VLAN))
+
+        response = self.client.get(site.get_absolute_url())
+        self.assertHttpStatus(response, 200)
+
+        rows = [row for row in response.context['related_models'] if row.queryset.model is VLAN]
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(list(rows[0].queryset), [visible_vlan])
+
+        list_url = f"{reverse('ipam:vlan_list')}?related_to_site={site.pk}"
+        list_response = self.client.get(list_url)
+        self.assertHttpStatus(list_response, 200)
+        self.assertEqual([vlan.pk for vlan in list_response.context['table'].data], [visible_vlan.pk])
+
 
 class LocationTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
     model = Location

netbox/dcim/views.py

@@ -589,12 +589,13 @@ class SiteView(GetRelatedModelsMixin, generic.ObjectView):
             'related_models': self.get_related_models(
                 request,
                 instance,
-                omit=(CableTermination, CircuitTermination, Cluster, Prefix, WirelessLAN),
+                omit=(CableTermination, CircuitTermination, Cluster, Prefix, VLAN, WirelessLAN),
                 extra=(
                     (VLANGroup.objects.restrict(request.user, 'view').filter(
                         scope_type=ContentType.objects.get_for_model(Site),
                         scope_id=instance.pk
                     ), 'site'),
+                    (VLAN.objects.restrict(request.user, 'view').get_related_to_sites([instance]), 'related_to_site'),
                     (ASN.objects.restrict(request.user, 'view').filter(sites=instance), 'site_id'),
                     (
                         Circuit.objects.restrict(request.user, 'view').filter(terminations___site=instance).distinct(),

netbox/ipam/filtersets.py

@@ -1036,6 +1036,17 @@ class VLANFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
         to_field_name='slug',
         label=_('Site (slug)'),
     )
+    related_to_site = django_filters.ModelMultipleChoiceFilter(
+        queryset=Site.objects.all(),
+        method='filter_related_to_site',
+        label=_('Related to site (ID)'),
+    )
+    # get_additional_lookups() skips filters with a method.
+    related_to_site__n = django_filters.ModelMultipleChoiceFilter(
+        queryset=Site.objects.all(),
+        method='filter_related_to_site_negated',
+        label=_('Related to site (ID)'),
+    )
     group_id = django_filters.ModelMultipleChoiceFilter(
         queryset=VLANGroup.objects.all(),
         distinct=False,
@@ -1146,6 +1157,18 @@ class VLANFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
     def get_for_virtualmachine(self, queryset, name, value):
         return queryset.get_for_virtualmachine(value)
 
+    @extend_schema_field(OpenApiTypes.INT)
+    def filter_related_to_site(self, queryset, name, value):
+        if not value:
+            return queryset
+        return queryset.get_related_to_sites(value)
+
+    @extend_schema_field(OpenApiTypes.INT)
+    def filter_related_to_site_negated(self, queryset, name, value):
+        if not value:
+            return queryset
+        return queryset.get_related_to_sites(value, negate=True)
+
     @extend_schema_field(OpenApiTypes.INT)
     def filter_interface_id(self, queryset, name, value):
         if value is None:

netbox/ipam/forms/filtersets.py

@@ -567,7 +567,7 @@ class VLANFilterForm(TenancyFilterForm, PrimaryModelFilterSetForm):
     model = VLAN
     fieldsets = (
         FieldSet('q', 'filter_id', 'tag'),
-        FieldSet('region_id', 'site_group_id', 'site_id', name=_('Location')),
+        FieldSet('region_id', 'site_group_id', 'site_id', 'related_to_site', name=_('Location')),
         FieldSet('group_id', 'status', 'role_id', 'vid', 'l2vpn_id', name=_('Attributes')),
         FieldSet('qinq_role', 'qinq_svlan_id', name=_('Q-in-Q/802.1ad')),
         FieldSet('tenant_group_id', 'tenant_id', name=_('Tenant')),
@@ -593,6 +593,11 @@ class VLANFilterForm(TenancyFilterForm, PrimaryModelFilterSetForm):
         },
         label=_('Site')
     )
+    related_to_site = DynamicModelMultipleChoiceField(
+        queryset=Site.objects.all(),
+        required=False,
+        label=_('Related to site')
+    )
     group_id = DynamicModelMultipleChoiceField(
         queryset=VLANGroup.objects.all(),
         required=False,

netbox/ipam/querysets.py

@@ -432,3 +432,39 @@ class VLANQuerySet(RestrictedQuerySet):
             q |= Q(site=site)
 
         return self.filter(q)
+
+    def get_related_to_sites(self, sites, *, negate=False):
+        """
+        Return VLANs related to any of the given sites, directly or through a scoped group.
+        Unlike get_for_site(), region-scoped and globally available VLANs are excluded.
+        Pass negate=True to exclude the related VLANs instead.
+        """
+        from dcim.models import SiteGroup
+
+        site_ids = set()
+        site_group_ids = set()
+        for site in sites:
+            site_ids.add(site.pk)
+            if site.group_id:
+                site_group_ids.add(site.group_id)
+
+        if not site_ids:
+            return self if negate else self.none()
+
+        q = Q(site_id__in=site_ids) | Q(
+            group__scope_type=ContentType.objects.get_by_natural_key('dcim', 'site'),
+            group__scope_id__in=site_ids
+        )
+
+        if site_group_ids:
+            # A site group scope reaches its descendants, so match the selected groups' ancestors.
+            ancestor_groups = SiteGroup.objects.none()
+            for site_group in SiteGroup.objects.filter(pk__in=site_group_ids):
+                ancestor_groups |= site_group.get_ancestors(include_self=True)
+
+            q |= Q(
+                group__scope_type=ContentType.objects.get_by_natural_key('dcim', 'sitegroup'),
+                group__scope_id__in=ancestor_groups.values('pk')
+            )
+
+        return self.exclude(q) if negate else self.filter(q)

netbox/ipam/tests/test_filtersets.py

@@ -2216,6 +2216,143 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'site': [sites[3].slug, sites[4].slug]}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
+    def test_related_to_site(self):
+        """Site-related VLANs cover direct assignment plus site and site group scopes."""
+        # site-1: one site-scoped VLAN and one scoped to its site group
+        site = Site.objects.get(slug='site-1')
+        params = {'related_to_site': [site.pk]}
+        self.assertEqual(
+            list(self.filterset(params, self.queryset).qs.order_by('vid').values_list('vid', flat=True)),
+            [4, 7]
+        )
+        # site-4: two directly assigned VLANs plus the same site-group-scoped VLAN
+        site = Site.objects.get(slug='site-4')
+        params = {'related_to_site': [site.pk]}
+        self.assertEqual(
+            list(self.filterset(params, self.queryset).qs.order_by('vid').values_list('vid', flat=True)),
+            [4, 101, 102]
+        )
+        # site-7 has no site group
+        site = Site.objects.get(slug='site-7')
+        params = {'related_to_site': [site.pk]}
+        self.assertEqual(
+            list(self.filterset(params, self.queryset).qs.order_by('vid').values_list('vid', flat=True)),
+            [2001, 2002, 2003, 3001, 3002, 3003]
+        )
+
+    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."""
+        sites = (
+            Site.objects.get(slug='site-1'),
+            Site.objects.get(slug='site-2'),
+            Site.objects.get(slug='site-4'),
+        )
+        params = {'related_to_site': [site.pk for site in sites]}
+        # VLAN 4 is scoped to the site group both site-1 and site-4 belong to, VLAN 5 to site-2's
+        self.assertEqual(
+            list(self.filterset(params, self.queryset).qs.order_by('vid').values_list('vid', flat=True)),
+            [4, 5, 7, 8, 101, 102]
+        )
+
+    def test_related_to_site_returns_dual_assignment_once(self):
+        """A VLAN matching both the direct site and a group scope is returned once."""
+        site = Site.objects.get(slug='site-1')
+        VLAN.objects.create(
+            vid=51,
+            name='Dual assignment',
+            site=site,
+            group=VLANGroup.objects.get(slug='site-group-1')
+        )
+
+        # 4 and 7 as before, plus the dual-assigned VLAN
+        params = {'related_to_site': [site.pk]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+
+    def test_related_to_site_includes_ancestor_site_group_scope(self):
+        """A group scoped to an ancestor site group relates to sites in descendant groups."""
+        parent = SiteGroup.objects.create(name='Parent Group', slug='parent-group')
+        child = SiteGroup.objects.create(name='Child Group', slug='child-group', parent=parent)
+        site = Site.objects.create(name='Site 8', slug='site-8', group=child)
+        group = VLANGroup.objects.create(name='Parent scope', slug='parent-scope', scope=parent)
+        VLAN.objects.create(vid=60, name='Parent scoped', group=group)
+
+        params = {'related_to_site': [site.pk]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
+    def test_related_to_site_ignores_matching_id_on_another_scope_type(self):
+        """A scope id equal to the site's pk under a different scope type does not match."""
+        site = Site.objects.get(slug='site-1')
+        group = VLANGroup.objects.create(
+            name='Region scope collision',
+            slug='region-scope-collision',
+            scope_type=ContentType.objects.get_by_natural_key('dcim', 'region'),
+            scope_id=site.pk
+        )
+        VLAN.objects.create(vid=61, name='Collision', group=group)
+
+        # Still only the site-scoped and site-group-scoped VLANs
+        params = {'related_to_site': [site.pk]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+    def test_related_to_site_narrows_the_supplied_queryset(self):
+        """The filter narrows the given queryset and cannot restore excluded VLANs."""
+        site = Site.objects.get(slug='site-1')
+        params = {'related_to_site': [site.pk]}
+        queryset = self.queryset.exclude(vid=4)
+
+        self.assertEqual(self.filterset(params, queryset).qs.count(), 1)
+
+    def test_related_to_site_negated(self):
+        """The negation lookup excludes everything the positive lookup returns."""
+        # related_to_site__n drops the 2 VLANs related to site-1
+        site = Site.objects.get(slug='site-1')
+        params = {'related_to_site__n': [site.pk]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), self.queryset.count() - 2)
+
+    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."""
+        sites = list(Site.objects.filter(slug__in=('site-1', 'site-2', 'site-3')))
+        site_without_group = Site.objects.get(slug='site-7')
+        # Both scope types are resolved through the ContentType cache, so warm it first.
+        ContentType.objects.get_by_natural_key('dcim', 'site')
+        ContentType.objects.get_by_natural_key('dcim', 'sitegroup')
+
+        # One query for the selected site groups, one for the VLANs
+        with self.assertNumQueries(2):
+            self.queryset.get_related_to_sites(sites[:1]).count()
+
+        with self.assertNumQueries(2):
+            self.queryset.get_related_to_sites(sites).count()
+
+        with self.assertNumQueries(2):
+            self.queryset.get_related_to_sites(sites, negate=True).count()
+
+        with self.assertNumQueries(1):
+            self.queryset.get_related_to_sites([site_without_group]).count()
+
+    def test_related_to_site_negation_partitions_the_queryset(self):
+        """The positive and negated lookups are disjoint and together cover the queryset."""
+        pks = [Site.objects.get(slug='site-1').pk, Site.objects.get(slug='site-4').pk]
+        # A VLAN with neither a site nor a group, and one in an unscoped group, belong to the negation.
+        VLAN.objects.create(vid=70, name='Unassigned')
+        VLAN.objects.create(vid=71, name='Unscoped group', group=VLANGroup.objects.get(slug='vlan-group-1'))
+
+        related_qs = self.filterset({'related_to_site': pks}, self.queryset).qs
+        unrelated_qs = self.filterset({'related_to_site__n': pks}, self.queryset).qs
+        related = set(related_qs.values_list('pk', flat=True))
+        unrelated = set(unrelated_qs.values_list('pk', flat=True))
+
+        self.assertEqual(related_qs.count(), len(related))
+        self.assertEqual(unrelated_qs.count(), len(unrelated))
+        self.assertEqual(related & unrelated, set())
+        self.assertEqual(related | unrelated, set(self.queryset.values_list('pk', flat=True)))
+        self.assertTrue(related)
+
+    def test_related_to_site_without_sites(self):
+        """No sites means no related VLANs, and the negation leaves the queryset untouched."""
+        self.assertEqual(self.queryset.get_related_to_sites([]).count(), 0)
+        self.assertEqual(self.queryset.get_related_to_sites([], negate=True).count(), self.queryset.count())
+
     def test_group(self):
         groups = VLANGroup.objects.filter(name__startswith='VLAN Group')[:2]
         params = {'group_id': [groups[0].pk, groups[1].pk]}

netbox/ipam/tests/test_forms.py

@@ -7,13 +7,14 @@ from dcim.constants import InterfaceTypeChoices
 from dcim.models import Device, DeviceRole, DeviceType, Interface, Location, Manufacturer, Region, Site, SiteGroup
 from ipam.choices import PrefixStatusChoices
 from ipam.constants import SERVICE_PORT_MAX, VLANGROUP_SCOPE_TYPES
-from ipam.filtersets import ServiceFilterSet, ServiceTemplateFilterSet
+from ipam.filtersets import ServiceFilterSet, ServiceTemplateFilterSet, VLANFilterSet
 from ipam.forms import PrefixForm, VLANGroupBulkEditForm, VLANGroupForm, VLANIDBulkCreateForm
 from ipam.forms.bulk_import import IPAddressImportForm, ServiceTemplateImportForm
 from ipam.forms.fields import PortMappingField
-from ipam.forms.filtersets import ServiceFilterForm, ServiceTemplateFilterForm
+from ipam.forms.filtersets import ServiceFilterForm, ServiceTemplateFilterForm, VLANFilterForm
 from ipam.forms.widgets import PortMappingWidget
 from ipam.models import Prefix, VLANGroup
+from utilities.forms.widgets import FilterModifierWidget
 
 
 class PrefixFormTestCase(TestCase):
@@ -234,6 +235,28 @@ class VLANFormTestCase(TestCase):
                 self.assertFalse(form.is_valid())
                 self.assertIn('pattern', form.errors)
 
+    def test_vlan_filter_form_exposes_related_to_site(self):
+        """The Location fieldset offers related to site under the same name as the filter."""
+        form = VLANFilterForm()
+        fieldset_items = [item for fieldset in VLANFilterForm.fieldsets for item in fieldset.items]
+
+        self.assertIn('related_to_site', fieldset_items)
+        self.assertIn('related_to_site', form.fields)
+        self.assertFalse(form.fields['related_to_site'].required)
+        # The form field's name must match the filter's, or the rendered query does nothing
+        self.assertIn('related_to_site', VLANFilterSet.get_filters())
+        template = Template('{% load form_helpers %}{% render_form form %}')
+        html = template.render(Context({'form': VLANFilterForm()}))
+        self.assertIn('id_related_to_site', html)
+
+    def test_vlan_filter_form_offers_related_to_site_operators(self):
+        """The declared negation filter is what puts an is/is not operator on the field."""
+        widget = VLANFilterForm().fields['related_to_site'].widget
+
+        self.assertIsInstance(widget, FilterModifierWidget)
+        self.assertEqual([lookup for lookup, _label in widget.lookups], ['exact', 'n'])
+        self.assertIn('related_to_site__n', VLANFilterSet.get_filters())
+
 
 class PortMappingFieldTestCase(TestCase):