Răsfoiți Sursa

fix(dcim): Add missing termination object filters to CableTerminationFilterSet (#22217)

Adds the cable_id FK companion filter and 9 termination object filters
(consoleport_id, consoleserverport_id, powerport_id, poweroutlet_id,
interface_id, frontport_id, rearport_id, powerfeed_id,
circuittermination_id), mirroring the CableFilterSet pattern.

Adds a corresponding CableTerminationTestCase using ChangeLoggedFilterSetTests
so future missing-filter regressions are caught automatically.

Fixes #22209
Martin Hauser 3 zile în urmă
părinte
comite
8c506c84c8
2 a modificat fișierele cu 156 adăugiri și 0 ștergeri
  1. 65 0
      netbox/dcim/filtersets.py
  2. 91 0
      netbox/dcim/tests/test_filtersets.py

+ 65 - 0
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):

+ 91 - 0
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 <component>_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