diff --git a/netbox/dcim/filtersets.py b/netbox/dcim/filtersets.py index 05a9751dc..b0bbb9b50 100644 --- a/netbox/dcim/filtersets.py +++ b/netbox/dcim/filtersets.py @@ -2806,12 +2806,77 @@ class CableFilterSet(TenancyFilterSet, PrimaryModelFilterSet): @register_filterset class CableTerminationFilterSet(ChangeLoggedModelFilterSet): + cable_id = django_filters.ModelMultipleChoiceFilter( + queryset=Cable.objects.all(), + distinct=False, + label=_('Cable (ID)'), + ) termination_type = MultiValueContentTypeFilter() + # Termination object filters + consoleport_id = MultiValueNumberFilter( + method='filter_by_consoleport' + ) + consoleserverport_id = MultiValueNumberFilter( + method='filter_by_consoleserverport' + ) + powerport_id = MultiValueNumberFilter( + method='filter_by_powerport' + ) + poweroutlet_id = MultiValueNumberFilter( + method='filter_by_poweroutlet' + ) + interface_id = MultiValueNumberFilter( + method='filter_by_interface' + ) + frontport_id = MultiValueNumberFilter( + method='filter_by_frontport' + ) + rearport_id = MultiValueNumberFilter( + method='filter_by_rearport' + ) + powerfeed_id = MultiValueNumberFilter( + method='filter_by_powerfeed' + ) + circuittermination_id = MultiValueNumberFilter( + method='filter_by_circuittermination' + ) + class Meta: model = CableTermination fields = ('id', 'cable', 'cable_end', 'termination_type', 'termination_id') + def filter_by_termination_object(self, queryset, model, value): + content_type = ContentType.objects.get_for_model(model) + return queryset.filter(termination_type=content_type, termination_id__in=value) + + def filter_by_consoleport(self, queryset, name, value): + return self.filter_by_termination_object(queryset, ConsolePort, value) + + def filter_by_consoleserverport(self, queryset, name, value): + return self.filter_by_termination_object(queryset, ConsoleServerPort, value) + + def filter_by_powerport(self, queryset, name, value): + return self.filter_by_termination_object(queryset, PowerPort, value) + + def filter_by_poweroutlet(self, queryset, name, value): + return self.filter_by_termination_object(queryset, PowerOutlet, value) + + def filter_by_interface(self, queryset, name, value): + return self.filter_by_termination_object(queryset, Interface, value) + + def filter_by_frontport(self, queryset, name, value): + return self.filter_by_termination_object(queryset, FrontPort, value) + + def filter_by_rearport(self, queryset, name, value): + return self.filter_by_termination_object(queryset, RearPort, value) + + def filter_by_powerfeed(self, queryset, name, value): + return self.filter_by_termination_object(queryset, PowerFeed, value) + + def filter_by_circuittermination(self, queryset, name, value): + return self.filter_by_termination_object(queryset, CircuitTermination, value) + @register_filterset class PowerPanelFilterSet(PrimaryModelFilterSet, ContactModelFilterSet): diff --git a/netbox/dcim/tests/test_filtersets.py b/netbox/dcim/tests/test_filtersets.py index e003487dd..f5c089344 100644 --- a/netbox/dcim/tests/test_filtersets.py +++ b/netbox/dcim/tests/test_filtersets.py @@ -1,4 +1,5 @@ from django.conf import settings +from django.contrib.contenttypes.models import ContentType from django.test import TestCase from circuits.models import Circuit, CircuitTermination, CircuitType, Provider @@ -7050,6 +7051,96 @@ class CableTestCase(TestCase, ChangeLoggedFilterSetTests): self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) +class CableTerminationTestCase(TestCase, ChangeLoggedFilterSetTests): + queryset = CableTermination.objects.all() + filterset = CableTerminationFilterSet + ignore_fields = ('connector', 'positions') + filter_name_map = { + 'consoleport': 'consoleport_id', + 'consoleserverport': 'consoleserverport_id', + 'powerport': 'powerport_id', + 'poweroutlet': 'poweroutlet_id', + 'interface': 'interface_id', + 'frontport': 'frontport_id', + 'rearport': 'rearport_id', + 'powerfeed': 'powerfeed_id', + 'circuittermination': 'circuittermination_id', + } + + @classmethod + def setUpTestData(cls): + site = Site.objects.create(name='Site 1', slug='site-1') + device = create_test_device('Device 1', site=site) + + cls.interfaces = [ + Interface(device=device, name=f'eth{i}', type=InterfaceTypeChoices.TYPE_1GE_FIXED) + for i in range(4) + ] + Interface.objects.bulk_create(cls.interfaces) + + cls.consoleport = ConsolePort.objects.create(device=device, name='Console Port 1') + cls.consoleserverport = ConsoleServerPort.objects.create(device=device, name='Console Server Port 1') + cls.powerport = PowerPort.objects.create(device=device, name='Power Port 1') + cls.poweroutlet = PowerOutlet.objects.create(device=device, name='Power Outlet 1') + cls.rearport = RearPort.objects.create(device=device, name='Rear Port 1', type=PortTypeChoices.TYPE_8P8C) + cls.frontport = FrontPort.objects.create(device=device, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C) + PortMapping.objects.create(device=device, front_port=cls.frontport, rear_port=cls.rearport) + + power_panel = PowerPanel.objects.create(name='Power Panel 1', site=site) + cls.powerfeed = PowerFeed.objects.create(name='Power Feed 1', power_panel=power_panel) + + provider = Provider.objects.create(name='Provider 1', slug='provider-1') + circuit_type = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1') + circuit = Circuit.objects.create(cid='Circuit 1', provider=provider, type=circuit_type) + cls.circuittermination = CircuitTermination.objects.create( + circuit=circuit, term_side='A', termination=site, + ) + + # Two bipartite interface cables (4 CableTerminations) plus one single-end cable per + # non-Interface component (8 CableTerminations). + cables = [ + Cable(a_terminations=[cls.interfaces[0]], b_terminations=[cls.interfaces[1]], label='Cable 1'), + Cable(a_terminations=[cls.interfaces[2]], b_terminations=[cls.interfaces[3]], label='Cable 2'), + ] + for component in ( + cls.consoleport, cls.consoleserverport, cls.powerport, cls.poweroutlet, + cls.rearport, cls.frontport, cls.powerfeed, cls.circuittermination, + ): + cables.append(Cable(a_terminations=[component], label=f'Cable for {component._meta.model_name}')) + for cable in cables: + cable.save() + + def test_cable(self): + """Filter CableTerminations by cable ID.""" + cables = Cable.objects.all()[:2] + params = {'cable_id': [cables[0].pk]} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + params = {'cable_id': [cables[0].pk, cables[1].pk]} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4) + + def test_termination_object_filters(self): + """Each _id filter resolves to only matching terminations.""" + cases = ( + ('consoleport_id', ConsolePort, self.consoleport), + ('consoleserverport_id', ConsoleServerPort, self.consoleserverport), + ('powerport_id', PowerPort, self.powerport), + ('poweroutlet_id', PowerOutlet, self.poweroutlet), + ('interface_id', Interface, self.interfaces[0]), + ('frontport_id', FrontPort, self.frontport), + ('rearport_id', RearPort, self.rearport), + ('powerfeed_id', PowerFeed, self.powerfeed), + ('circuittermination_id', CircuitTermination, self.circuittermination), + ) + for filter_name, model, obj in cases: + with self.subTest(filter_name=filter_name): + ct = ContentType.objects.get_for_model(model) + params = {filter_name: [obj.pk]} + results = self.filterset(params, self.queryset).qs + self.assertEqual(results.count(), 1) + self.assertEqual(results.first().termination_type, ct) + self.assertEqual(results.first().termination_id, obj.pk) + + class PowerPanelTestCase(TestCase, ChangeLoggedFilterSetTests): queryset = PowerPanel.objects.all() filterset = PowerPanelFilterSet