Merge pull request #23455 from netbox-community/22883-one-to-many-cable-connections

Fixes #22883: Fix editing cables with terminations on multiple devices

netbox/dcim/forms/connections.py

@@ -5,6 +5,7 @@ from django.utils.translation import gettext_lazy as _
 from circuits.models import Circuit, CircuitTermination
 from dcim.models import *
 from utilities.forms.fields import DynamicModelMultipleChoiceField
+from utilities.forms.widgets import APISelectMultiple
 
 from .model_forms import CableForm
 
@@ -39,8 +40,9 @@ def get_cable_form(a_type, b_type):
                         }
                     )
                     attrs[f'{cable_end}_terminations'] = DynamicModelMultipleChoiceField(
-                        queryset=term_cls.objects.all(),
+                        queryset=term_cls.objects.select_related('device'),
                         label=term_cls._meta.verbose_name.title(),
+                        widget=APISelectMultiple(attrs={'data-preserve-selection': True}),
                         context={
                             'disabled': '_occupied',
                             'parent': 'device',
@@ -66,6 +68,7 @@ def get_cable_form(a_type, b_type):
                     attrs[f'{cable_end}_terminations'] = DynamicModelMultipleChoiceField(
                         queryset=term_cls.objects.all(),
                         label=_('Power Feed'),
+                        widget=APISelectMultiple(attrs={'data-preserve-selection': True}),
                         context={
                             'disabled': '_occupied',
                             'parent': 'powerpanel',
@@ -89,6 +92,7 @@ def get_cable_form(a_type, b_type):
                     attrs[f'{cable_end}_terminations'] = DynamicModelMultipleChoiceField(
                         queryset=term_cls.objects.all(),
                         label=_('Side'),
+                        widget=APISelectMultiple(attrs={'data-preserve-selection': True}),
                         context={
                             'disabled': '_occupied',
                             'parent': 'circuit',

netbox/dcim/tests/test_api.py

@@ -5871,6 +5871,66 @@ class CableTestCase(APIViewTestCases.APIViewTestCase):
                 self.assertEqual(kept.cable, cable)
                 self.assertFalse(kept._path.is_complete)
 
+    def test_update_without_terminations_preserves_them(self):
+        """PATCHing a scalar field leaves both cable ends unchanged."""
+        cable = Cable.objects.get(label='Cable 1')
+        a_terminations, b_terminations = cable.a_terminations, cable.b_terminations
+
+        self.add_permissions('dcim.change_cable')
+        response = self.client.patch(
+            self._get_detail_url(cable), {'label': 'Cable 1 renamed'}, format='json', **self.header
+        )
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        cable = Cable.objects.get(pk=cable.pk)
+        self.assertEqual(cable.a_terminations, a_terminations)
+        self.assertEqual(cable.b_terminations, b_terminations)
+
+    def test_update_terminations_adds_to_a_multi_termination_end(self):
+        """PATCHing the complete termination list keeps the existing rows and adds the new termination."""
+        device1, device2 = Device.objects.order_by('name')
+        interfaces = [Interface.objects.get(device=device2, name=name) for name in ('eth7', 'eth8', 'eth9')]
+        cable = Cable(
+            a_terminations=[Interface.objects.get(device=device1, name='eth7')],
+            b_terminations=interfaces[:2],
+            label='Cable 7'
+        )
+        cable.save()
+        rows = CableTermination.objects.filter(cable=cable, cable_end=CableEndChoices.SIDE_B)
+        existing_rows = set(rows.values_list('pk', flat=True))
+
+        self.add_permissions('dcim.change_cable')
+        data = {
+            'b_terminations': [
+                {'object_type': 'dcim.interface', 'object_id': interface.pk} for interface in interfaces
+            ],
+        }
+        response = self.client.patch(self._get_detail_url(cable), data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        self.assertCountEqual(Cable.objects.get(pk=cable.pk).b_terminations, interfaces)
+        self.assertTrue(existing_rows.issubset(rows.values_list('pk', flat=True)))
+
+    def test_update_terminations_replaces_a_multi_termination_end(self):
+        """PATCHing a shorter termination list removes the omitted termination."""
+        device1, device2 = Device.objects.order_by('name')
+        interfaces = [Interface.objects.get(device=device2, name=name) for name in ('eth7', 'eth8')]
+        cable = Cable(
+            a_terminations=[Interface.objects.get(device=device1, name='eth7')],
+            b_terminations=interfaces,
+            label='Cable 7'
+        )
+        cable.save()
+
+        self.add_permissions('dcim.change_cable')
+        data = {
+            'b_terminations': [{'object_type': 'dcim.interface', 'object_id': interfaces[0].pk}],
+        }
+        response = self.client.patch(self._get_detail_url(cable), data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        self.assertEqual(Cable.objects.get(pk=cable.pk).b_terminations, [interfaces[0]])
+
     def test_graphql_cable_termination_cached_filters(self):
         """
         Validate filtering cables by cached CableTermination relations via GraphQL:

netbox/dcim/tests/test_forms.py

@@ -4,6 +4,7 @@ from django import forms
 from django.template.loader import render_to_string
 from django.test import TestCase
 
+from circuits.models import CircuitTermination
 from dcim.choices import (
     CableEndChoices,
     CableProfileChoices,
@@ -14,6 +15,7 @@ from dcim.choices import (
     LinkStatusChoices,
     PortTypeChoices,
     PowerOutletStatusChoices,
+    VirtualDeviceContextStatusChoices,
 )
 from dcim.forms import *
 from dcim.models import *
@@ -1089,6 +1091,29 @@ class InterfaceTestCase(TestCase):
         self.assertNotIn('tagged_vlans', form.cleaned_data.keys())
         self.assertNotIn('qinq_svlan', form.cleaned_data.keys())
 
+    def test_child_interface_preselects_every_parent_vdc(self):
+        """A child interface's form pre-selects every virtual device context of its parent."""
+        vdcs = (
+            VirtualDeviceContext(
+                device=self.device,
+                name='VDC 1',
+                identifier=1,
+                status=VirtualDeviceContextStatusChoices.STATUS_ACTIVE
+            ),
+            VirtualDeviceContext(
+                device=self.device,
+                name='VDC 2',
+                identifier=2,
+                status=VirtualDeviceContextStatusChoices.STATUS_ACTIVE
+            ),
+        )
+        VirtualDeviceContext.objects.bulk_create(vdcs)
+        self.interface.vdcs.set(vdcs)
+
+        form = InterfaceForm(initial={'device': self.device.pk, 'parent': self.interface.pk})
+
+        self.assertCountEqual([vdc.pk for vdc in form['vdcs'].field.queryset], [vdc.pk for vdc in vdcs])
+
 
 class CableTestCase(TestCase):
 
@@ -1123,6 +1148,14 @@ class CableTestCase(TestCase):
         )
         PowerPort.objects.bulk_create(cls.power_ports)
 
+    def _make_edit_form(self):
+        cable = Cable(
+            a_terminations=[self.interfaces_a[0]],
+            b_terminations=[self.interfaces_b[1], self.interface_c],
+        )
+        cable.save()
+        return get_cable_form(Interface, Interface)(instance=Cable.objects.get(pk=cable.pk))
+
     def test_invalid_side_designation_raises_value_error(self):
         """_clean_side rejects a side other than 'a' or 'b' with ValueError."""
         form = CableImportForm.__new__(CableImportForm)
@@ -1459,6 +1492,62 @@ class CableTestCase(TestCase):
         self.assertFalse(form.is_valid())
         self.assertIn('side termination not unique', str(form.errors.get('side_b_name')))
 
+    def test_edit_form_selects_every_device_terminating_an_end(self):
+        """The edit form selects every device terminating an end of the cable."""
+        form = self._make_edit_form()
+
+        self.assertCountEqual(
+            [device.pk for device in form['termination_b_device'].field.queryset],
+            [self.device_b.pk, self.device_c.pk]
+        )
+
+    def test_edit_form_names_the_device_of_each_termination(self):
+        """Each termination option names its device, so identically named interfaces stay distinct."""
+        html = str(self._make_edit_form()['b_terminations'])
+
+        for interface in (self.interfaces_b[1], self.interface_c):
+            self.assertInHTML(
+                f'<option value="{interface.pk}" data-parent="{interface.device}" selected>et-0/0/1</option>', html
+            )
+
+    def test_termination_fields_preserve_their_selection(self):
+        """Only the termination fields keep their selection when a parent selector changes."""
+        for termination_type, parent in (
+            (Interface, 'device'),
+            (PowerFeed, 'powerpanel'),
+            (CircuitTermination, 'circuit'),
+        ):
+            fields = get_cable_form(termination_type, termination_type)().fields
+            with self.subTest(termination_type=termination_type.__name__):
+                self.assertIn('data-preserve-selection', fields['b_terminations'].widget.attrs)
+                self.assertNotIn('data-preserve-selection', fields[f'termination_b_{parent}'].widget.attrs)
+
+    def test_edit_form_saves_terminations_outside_the_device_filter(self):
+        """Saving keeps every selected termination and its row, even when its device is not in the Device filter."""
+        cable = self._make_edit_form().instance
+        rows = CableTermination.objects.filter(cable=cable, cable_end=CableEndChoices.SIDE_B)
+        existing_rows = set(rows.values_list('pk', flat=True))
+        device_d = create_test_device('Device D', site=self.site)
+        interface_d = Interface.objects.create(
+            device=device_d, name='et-0/0/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
+        )
+        b_terminations = [self.interfaces_b[1], self.interface_c, interface_d]
+
+        form = get_cable_form(Interface, Interface)(instance=Cable.objects.get(pk=cable.pk), data={
+            'a_terminations_type': 'dcim.interface',
+            'b_terminations_type': 'dcim.interface',
+            'termination_a_device': [self.device_a.pk],
+            'termination_b_device': [device_d.pk],
+            'a_terminations': [self.interfaces_a[0].pk],
+            'b_terminations': [interface.pk for interface in b_terminations],
+            'status': LinkStatusChoices.STATUS_CONNECTED,
+        })
+        self.assertTrue(form.is_valid(), form.errors)
+        form.save()
+
+        self.assertCountEqual(Cable.objects.get(pk=cable.pk).b_terminations, b_terminations)
+        self.assertTrue(existing_rows.issubset(rows.values_list('pk', flat=True)))
+
 
 class SiteFormTestCase(TestCase):
     """

netbox/project-static/src/select/classes/dynamicTomSelect.ts

@@ -437,6 +437,12 @@ export class DynamicTomSelect extends NetBoxTomSelect {
     this.updateQueryParams(target.name);
     this.updatePathValues(target.name);
 
+    // A dependency only filters what can be added to a field that preserves its selection.
+    if (this.input.hasAttribute('data-preserve-selection')) {
+      this.load(this.lastValue);
+      return;
+    }
+
     // Clear any previous selection(s) as the parent filter has changed
     this.clear();
 

netbox/utilities/forms/fields/dynamic.py

@@ -62,6 +62,7 @@ class DynamicModelChoiceMixin:
     Attributes:
         query_params: A dictionary of additional key/value pairs to attach to the API request
         initial_params: A dictionary of child field references to use for selecting a parent field's initial value
+            (every matching object for a multiple-choice field, the first for a single-choice field)
         null_option: The string used to represent a null selection (if any)
         disabled_indicator: The name of the field which, if populated, will disable selection of the
             choice (DEPRECATED: pass `context={'disabled': '$fieldname'}` instead)
@@ -144,7 +145,12 @@ class DynamicModelChoiceMixin:
                 if value:
                     filter_kwargs[kwarg] = value
             if filter_kwargs:
-                self.initial = self.queryset.filter(**filter_kwargs).first()
+                queryset = self.queryset.filter(**filter_kwargs)
+                if isinstance(self, forms.ModelMultipleChoiceField):
+                    # A lookup across a reverse relation yields one row per matching child
+                    self.initial = list(queryset.distinct())
+                else:
+                    self.initial = queryset.first()
 
         # Modify the QuerySet of the field before we return it. Limit choices to any data already bound: Options
         # will be populated on-demand via the APISelect widget.

netbox/utilities/forms/widgets/apiselect.py

@@ -2,6 +2,7 @@ import json
 
 from django import forms
 from django.conf import settings
+from django.forms.models import ModelChoiceIteratorValue
 from django.utils.translation import gettext_lazy as _
 
 __all__ = (
@@ -51,6 +52,17 @@ class APISelect(forms.Select):
         result.static_params = {}
         return result
 
+    def create_option(self, name, value, label, selected, index, subindex=None, attrs=None):
+        """Name each model option's parent object, which options loaded from the REST API carry already."""
+        option = super().create_option(name, value, label, selected, index, subindex, attrs)
+        parent_field = self.attrs.get('ts-parent-field')
+        if parent_field and isinstance(value, ModelChoiceIteratorValue):
+            # The context names an API field, which need not be a model attribute
+            parent = getattr(value.instance, parent_field, None)
+            if parent is not None:
+                option['attrs']['data-parent'] = str(parent)
+        return option
+
     def _process_query_param(self, key, value) -> None:
         """
         Based on query param value's type and value, update instance's dynamic/static params.

netbox/utilities/tests/test_forms.py

@@ -5,14 +5,20 @@ from types import SimpleNamespace
 from django import forms
 from django.contrib.contenttypes.models import ContentType
 from django.test import TestCase, override_settings
+from django.utils.html import escape
 
-from dcim.models import Site
+from dcim.models import Rack, Site
 from netbox.choices import ImportFormatChoices
 from utilities.choices import Choice, ChoiceSet
 from utilities.forms.bulk_import import BulkImportForm
 from utilities.forms.fields import ChoiceField, MultipleChoiceField, TypedChoiceField
 from utilities.forms.fields.csv import CSVSelectWidget
-from utilities.forms.fields.dynamic import DynamicChoiceField, DynamicMultipleChoiceField
+from utilities.forms.fields.dynamic import (
+    DynamicChoiceField,
+    DynamicModelChoiceField,
+    DynamicModelMultipleChoiceField,
+    DynamicMultipleChoiceField,
+)
 from utilities.forms.fields.generic import GenericObjectChoiceField
 from utilities.forms.forms import BulkRenameForm
 from utilities.forms.mixins import GenericObjectFormMixin
@@ -686,6 +692,79 @@ class DynamicMultipleChoiceFieldTestCase(TestCase):
         self.assertEqual(form.fields['field'].choices, [])
 
 
+class DynamicModelChoiceFieldTestCase(TestCase):
+    """Validate how dynamic model choice fields seed their initial value and render their options."""
+
+    @classmethod
+    def setUpTestData(cls):
+        # The quote in Site 2's name proves that a rendered parent name is escaped
+        cls.sites = (
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site "2"', slug='site-2'),
+        )
+        Site.objects.bulk_create(cls.sites)
+
+        # Rack 2 and Rack 3 share Site 2, so the reverse lookup returns it twice
+        cls.racks = (
+            Rack(name='Rack 1', site=cls.sites[0]),
+            Rack(name='Rack 2', site=cls.sites[1]),
+            Rack(name='Rack 3', site=cls.sites[1]),
+        )
+        Rack.objects.bulk_create(cls.racks)
+
+    def _make_form(self, site_field_class, data=None, rack_context=None):
+        class TestForm(forms.Form):
+            rack = DynamicModelMultipleChoiceField(queryset=Rack.objects.all(), required=False, context=rack_context)
+            site = site_field_class(
+                queryset=Site.objects.all(),
+                required=False,
+                initial_params={'racks__in': '$rack'},
+            )
+        return TestForm(data=data, initial={'rack': [rack.pk for rack in self.racks]})
+
+    def test_multiple_choice_field_selects_every_matching_object(self):
+        """A multiple-choice parent field selects every distinct object matched by initial_params."""
+        field = self._make_form(DynamicModelMultipleChoiceField)['site'].field
+
+        self.assertCountEqual([site.pk for site in field.queryset], [site.pk for site in self.sites])
+        # The queryset filter deduplicates, so only initial reveals a repeated parent
+        self.assertCountEqual(field.initial, self.sites)
+
+    def test_multiple_choice_field_supports_changed_data(self):
+        """An unchanged submission of a multiple-choice parent field is absent from changed_data."""
+        data = {
+            'rack': [rack.pk for rack in self.racks],
+            'site': [site.pk for site in self.sites],
+        }
+        form = self._make_form(DynamicModelMultipleChoiceField, data=data)
+
+        self.assertEqual(form.changed_data, [])
+
+    def test_single_choice_field_selects_one_matching_object(self):
+        """A single-choice parent field still selects one object matched by initial_params."""
+        field = self._make_form(DynamicModelChoiceField)['site'].field
+
+        self.assertIn(field.initial, self.sites)
+        self.assertEqual(field.queryset.count(), 1)
+
+    def test_options_name_their_parent(self):
+        """Each rendered option names its parent object."""
+        html = str(self._make_form(DynamicModelMultipleChoiceField, rack_context={'parent': 'site'})['rack'])
+
+        for rack in self.racks:
+            self.assertInHTML(
+                f'<option value="{rack.pk}" data-parent="{escape(rack.site)}" selected>{rack}</option>', html
+            )
+
+    def test_options_omit_a_missing_parent(self):
+        """Options carry no parent when the context names an unset relation or no model attribute."""
+        unset_parent = self._make_form(DynamicModelMultipleChoiceField, rack_context={'parent': 'location'})
+        unknown_parent = self._make_form(DynamicModelMultipleChoiceField, rack_context={'parent': 'region'})
+
+        self.assertNotIn('data-parent', str(unset_parent['rack']))
+        self.assertNotIn('data-parent', str(unknown_parent['rack']))
+
+
 class GenericObjectChoiceFieldTestCase(TestCase):
     """Validate generic foreign key object selection via a content type plus an API-backed object selector."""