|
|
@@ -42,10 +42,11 @@ from utilities.testing import (
|
|
|
create_test_device,
|
|
|
create_test_image,
|
|
|
create_test_nat_ip_pair,
|
|
|
+ create_test_virtualmachine,
|
|
|
disable_logging,
|
|
|
disable_warnings,
|
|
|
)
|
|
|
-from virtualization.models import Cluster, ClusterType
|
|
|
+from virtualization.models import Cluster, ClusterType, VMInterface
|
|
|
from vpn.choices import L2VPNTypeChoices
|
|
|
from vpn.models import L2VPN, L2VPNTermination
|
|
|
from wireless.choices import WirelessChannelChoices
|
|
|
@@ -2806,6 +2807,56 @@ class DeviceTestCase(Mixins.FieldPrefetchMixin, APIViewTestCases.APIViewTestCase
|
|
|
self.assertNotIn('errors', data)
|
|
|
self.assertEqual({row['name']: row['config_context'] for row in data['data']['device_list']}, expected)
|
|
|
|
|
|
+ @tag('regression') # Issue #23448
|
|
|
+ def test_config_context_graphql_selection(self):
|
|
|
+ """
|
|
|
+ GraphQL annotates config context for devices with no cache however config_context is selected, including
|
|
|
+ through fragments and nested relations, without a query per device.
|
|
|
+ """
|
|
|
+ self.add_permissions('dcim.view_device', 'dcim.view_interface')
|
|
|
+ ConfigContext.objects.create(name='Config Context 1', weight=100, data={'foo': 123})
|
|
|
+ for device in Device.objects.all():
|
|
|
+ Interface.objects.create(device=device, name='eth0', type=InterfaceTypeChoices.TYPE_1GE_FIXED)
|
|
|
+ Device.objects.update(_config_context_data=None)
|
|
|
+ expected = {device.name: device.render_config_context() for device in Device.objects.all()}
|
|
|
+
|
|
|
+ fragment = 'fragment F on DeviceType { name config_context }'
|
|
|
+ queries = {
|
|
|
+ 'field': '{ device_list { name config_context } }',
|
|
|
+ 'inline fragment': '{ device_list { ... on DeviceType { name config_context } } }',
|
|
|
+ 'named fragment': f'{fragment} query {{ device_list {{ ...F }} }}',
|
|
|
+ 'nested field': '{ interface_list { device { name config_context } } }',
|
|
|
+ 'nested inline fragment': '{ interface_list { device { ... on DeviceType { name config_context } } } }',
|
|
|
+ 'nested named fragment': f'{fragment} query {{ interface_list {{ device {{ ...F }} }} }}',
|
|
|
+ }
|
|
|
+ for shape, query in queries.items():
|
|
|
+ with self.subTest(shape=shape):
|
|
|
+ with CaptureQueriesContext(connection) as captured:
|
|
|
+ response = self.client.post(
|
|
|
+ reverse('graphql'), data={'query': query}, format='json', **self.header
|
|
|
+ )
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = response.json()
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ if shape.startswith('nested'):
|
|
|
+ rows = [row['device'] for row in data['data']['interface_list']]
|
|
|
+ else:
|
|
|
+ rows = data['data']['device_list']
|
|
|
+ self.assertEqual({row['name']: row['config_context'] for row in rows}, expected)
|
|
|
+ # Rendered by the annotation in a single query, not one fallback query per device
|
|
|
+ sql = [captured_query['sql'] for captured_query in captured.captured_queries]
|
|
|
+ self.assertEqual(len([s for s in sql if '"extras_configcontext"' in s]), 1)
|
|
|
+
|
|
|
+ # Not annotated when config_context is not selected
|
|
|
+ with CaptureQueriesContext(connection) as captured:
|
|
|
+ response = self.client.post(
|
|
|
+ reverse('graphql'), data={'query': '{ device_list { name } }'}, format='json', **self.header
|
|
|
+ )
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ self.assertNotIn('errors', response.json())
|
|
|
+ for captured_query in captured.captured_queries:
|
|
|
+ self.assertNotIn('"config_context_data"', captured_query['sql'])
|
|
|
+
|
|
|
def test_config_context_included_by_default_in_list_view(self):
|
|
|
"""
|
|
|
Check that config context data is included by default in the devices list.
|
|
|
@@ -4088,8 +4139,6 @@ class InterfaceTestCase(Mixins.FieldPrefetchMixin, Mixins.ComponentTraceMixin, A
|
|
|
GenericPrefetch('_path__path_objects', [Interface.objects.select_related('device', 'cable')]),
|
|
|
'virtual_circuit_termination',
|
|
|
'l2vpn_terminations',
|
|
|
- 'ip_addresses',
|
|
|
- 'fhrp_group_assignments',
|
|
|
)
|
|
|
brief_fields = ['_occupied', 'cable', 'description', 'device', 'display', 'id', 'name', 'url']
|
|
|
bulk_update_data = {
|
|
|
@@ -4732,37 +4781,44 @@ class InterfaceTestCase(Mixins.FieldPrefetchMixin, Mixins.ComponentTraceMixin, A
|
|
|
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
|
|
|
def test_field_prefetches_follow_requested_fields(self):
|
|
|
- """List and detail responses carry the declared prefetches of exactly the fields they include."""
|
|
|
+ """List and detail responses prefetch the declarations of included fields, counts annotated instead."""
|
|
|
self.add_permissions('dcim.view_interface')
|
|
|
declared = InterfaceViewSet.field_prefetches
|
|
|
+ declared_before = dict(declared)
|
|
|
peers = list(declared[('link_peers', 'link_peers_type')])
|
|
|
endpoints = list(declared[('connected_endpoints', 'connected_endpoints_type')])
|
|
|
path = list(declared[('connected_endpoints_reachable',)])
|
|
|
l2vpn = list(declared[('l2vpn_termination',)])
|
|
|
- ips, fhrp = list(declared[('count_ipaddresses',)]), list(declared[('count_fhrp_groups',)])
|
|
|
+ counts = {'_count_ipaddresses', '_count_fhrp_groups'}
|
|
|
# fields takes precedence over omit, and both over brief
|
|
|
cases = {
|
|
|
- '': peers + endpoints + path + l2vpn + ips + fhrp,
|
|
|
- 'brief=1': [],
|
|
|
- 'fields=id,link_peers_type': peers,
|
|
|
- 'fields=id,connected_endpoints_type': endpoints,
|
|
|
- 'fields=id,connected_endpoints_reachable': path,
|
|
|
- 'fields=id,l2vpn_termination': l2vpn,
|
|
|
- 'fields=id,count_ipaddresses': ips,
|
|
|
- 'fields=id,count_fhrp_groups': fhrp,
|
|
|
- 'omit=link_peers,link_peers_type': endpoints + path + l2vpn + ips + fhrp,
|
|
|
- 'fields=id,link_peers&omit=link_peers': peers,
|
|
|
- 'brief=1&omit=l2vpn_termination': peers + endpoints + path + ips + fhrp,
|
|
|
- 'brief=1&fields=id': [],
|
|
|
+ '': (peers + endpoints + path + l2vpn, counts),
|
|
|
+ 'brief=1': ([], set()),
|
|
|
+ 'fields=id,link_peers_type': (peers, set()),
|
|
|
+ 'fields=id,connected_endpoints_type': (endpoints, set()),
|
|
|
+ 'fields=id,connected_endpoints_reachable': (path, set()),
|
|
|
+ 'fields=id,l2vpn_termination': (l2vpn, set()),
|
|
|
+ 'fields=id,count_ipaddresses': ([], {'_count_ipaddresses'}),
|
|
|
+ 'fields=id,count_fhrp_groups': ([], {'_count_fhrp_groups'}),
|
|
|
+ 'fields=count_ipaddresses': ([], {'_count_ipaddresses'}),
|
|
|
+ 'omit=link_peers,link_peers_type': (endpoints + path + l2vpn, counts),
|
|
|
+ 'omit=count_ipaddresses': (peers + endpoints + path + l2vpn, {'_count_fhrp_groups'}),
|
|
|
+ 'fields=id,link_peers&omit=link_peers': (peers, set()),
|
|
|
+ 'brief=1&omit=l2vpn_termination': (peers + endpoints + path, counts),
|
|
|
+ 'brief=1&fields=id': ([], set()),
|
|
|
}
|
|
|
for url in (self._get_list_url(), self._get_detail_url(Interface.objects.get(name='Interface 1'))):
|
|
|
- for params, expected in cases.items():
|
|
|
+ for params, (expected, annotations) in cases.items():
|
|
|
with self.subTest(url=url, params=params):
|
|
|
with self.capture_querysets(InterfaceViewSet) as querysets:
|
|
|
response = self.client.get(f'{url}?{params}', **self.header)
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
self.assert_lookups(querysets, expected)
|
|
|
+ for queryset in querysets:
|
|
|
+ self.assertEqual(set(queryset.query.annotations), annotations)
|
|
|
self.assertEqual(InterfaceViewSet.queryset._prefetch_related_lookups, ())
|
|
|
+ # Reads narrow their own field set, never the class declarations
|
|
|
+ self.assertEqual(InterfaceViewSet.field_prefetches, declared_before)
|
|
|
|
|
|
def test_field_prefetches_keep_class_lookups(self):
|
|
|
"""Lookups on the class queryset stay ahead of the declared ones, and a subclass inherits the declarations."""
|
|
|
@@ -4779,6 +4835,18 @@ class InterfaceTestCase(Mixins.FieldPrefetchMixin, Mixins.ComponentTraceMixin, A
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
self.assert_lookups(querysets, [*queryset._prefetch_related_lookups, *peers])
|
|
|
|
|
|
+ # A class lookup on a counted relation stays when its count is annotated
|
|
|
+ queryset = InterfaceViewSet.queryset.prefetch_related('ip_addresses')
|
|
|
+ with (
|
|
|
+ patch.object(InterfaceViewSet, 'queryset', queryset),
|
|
|
+ self.capture_querysets(InterfaceViewSet) as querysets,
|
|
|
+ ):
|
|
|
+ response = self.client.get(f'{url}?fields=id,count_ipaddresses', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ self.assert_lookups(querysets, list(queryset._prefetch_related_lookups))
|
|
|
+ for captured in querysets:
|
|
|
+ self.assertEqual(set(captured.query.annotations), {'_count_ipaddresses'})
|
|
|
+
|
|
|
class PluginInterfaceViewSet(InterfaceViewSet):
|
|
|
pass
|
|
|
|
|
|
@@ -4789,8 +4857,25 @@ class InterfaceTestCase(Mixins.FieldPrefetchMixin, Mixins.ComponentTraceMixin, A
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
self.assert_lookups(querysets, peers)
|
|
|
|
|
|
+ def test_field_prefetches_keep_shared_declarations(self):
|
|
|
+ """A declaration a count shares with another requested field stays when the count is annotated."""
|
|
|
+ self.add_permissions('dcim.view_interface')
|
|
|
+ shared = ('ip_addresses',)
|
|
|
+ field_prefetches = {**InterfaceViewSet.field_prefetches, ('count_ipaddresses', 'description'): shared}
|
|
|
+ with (
|
|
|
+ patch.object(InterfaceViewSet, 'field_prefetches', field_prefetches),
|
|
|
+ self.capture_querysets(InterfaceViewSet) as querysets,
|
|
|
+ ):
|
|
|
+ response = self.client.get(
|
|
|
+ f'{self._get_list_url()}?fields=id,description,count_ipaddresses', **self.header
|
|
|
+ )
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ self.assert_lookups(querysets, list(shared))
|
|
|
+ for queryset in querysets:
|
|
|
+ self.assertEqual(set(queryset.query.annotations), {'_count_ipaddresses'})
|
|
|
+
|
|
|
def test_field_prefetches_for_other_reads(self):
|
|
|
- """Exports and writes carry every declaration, the trace action its path nodes."""
|
|
|
+ """Exports and writes carry every declaration and annotate no count, the trace action its path nodes."""
|
|
|
self.add_permissions('dcim.view_interface', 'dcim.change_interface', 'extras.view_exporttemplate')
|
|
|
declared = [lookup for lookups in InterfaceViewSet.field_prefetches.values() for lookup in lookups]
|
|
|
interface = Interface.objects.get(name='Interface 1')
|
|
|
@@ -4801,13 +4886,16 @@ class InterfaceTestCase(Mixins.FieldPrefetchMixin, Mixins.ComponentTraceMixin, A
|
|
|
|
|
|
# Exports and writes load every declaration, whatever fields the response selects
|
|
|
with self.capture_querysets(InterfaceViewSet) as querysets:
|
|
|
- response = self.client.get(f'{self._get_list_url()}?fields=id&export=Interfaces', **self.header)
|
|
|
- self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
- response = self.client.patch(
|
|
|
- f'{self._get_detail_url(interface)}?fields=id', {'description': 'New'}, format='json', **self.header
|
|
|
- )
|
|
|
- self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ for params in ('fields=id&export=Interfaces', 'export=Interfaces'):
|
|
|
+ response = self.client.get(f'{self._get_list_url()}?{params}', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ for params in ('fields=id', 'fields=id,count_ipaddresses'):
|
|
|
+ response = self.client.patch(
|
|
|
+ f'{self._get_detail_url(interface)}?{params}', {'description': 'New'}, format='json', **self.header
|
|
|
+ )
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
self.assert_lookups(querysets, declared)
|
|
|
+ self.assertFalse([queryset.query.annotations for queryset in querysets if queryset.query.annotations])
|
|
|
|
|
|
# The trace action loads the path nodes as the connected endpoint fields do, whatever fields are selected
|
|
|
peer = Interface.objects.create(device=interface.device, name='Peer', type=InterfaceTypeChoices.TYPE_1GE_FIXED)
|
|
|
@@ -4836,6 +4924,126 @@ class InterfaceTestCase(Mixins.FieldPrefetchMixin, Mixins.ComponentTraceMixin, A
|
|
|
self.assertLessEqual(count, eager_count)
|
|
|
self.assertLess(count, unprefetched_count)
|
|
|
|
|
|
+ def test_count_ipaddresses_and_fhrp_groups(self):
|
|
|
+ """Counts come from annotations of the interface's object type, never from loaded related objects."""
|
|
|
+ self.add_permissions('dcim.view_interface')
|
|
|
+ interface = Interface.objects.get(name='Interface 1')
|
|
|
+ group = FHRPGroup.objects.create(
|
|
|
+ pk=interface.pk, protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=1
|
|
|
+ )
|
|
|
+ other_group = FHRPGroup.objects.create(
|
|
|
+ pk=interface.pk + 1, protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=2
|
|
|
+ )
|
|
|
+ vm_interface = VMInterface.objects.create(
|
|
|
+ pk=interface.pk, virtual_machine=create_test_virtualmachine('Virtual Machine 1'), name='eth0'
|
|
|
+ )
|
|
|
+ IPAddress.objects.bulk_create((
|
|
|
+ IPAddress(address='192.0.2.1/24', assigned_object=interface),
|
|
|
+ IPAddress(address='2001:db8::1/64', assigned_object=interface),
|
|
|
+ IPAddress(address='192.0.2.2/24', assigned_object=vm_interface),
|
|
|
+ IPAddress(address='192.0.2.254/24', assigned_object=group),
|
|
|
+ ))
|
|
|
+ FHRPGroupAssignment.objects.bulk_create((
|
|
|
+ FHRPGroupAssignment(group=group, interface=interface, priority=100),
|
|
|
+ # Two on the VM interface, so a count of the wrong object type cannot match
|
|
|
+ FHRPGroupAssignment(group=group, interface=vm_interface, priority=100),
|
|
|
+ FHRPGroupAssignment(group=other_group, interface=vm_interface, priority=100),
|
|
|
+ ))
|
|
|
+
|
|
|
+ with CaptureQueriesContext(connection) as queries:
|
|
|
+ response = self.client.get(self._get_list_url(), **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ counts = {
|
|
|
+ row['name']: (row['count_ipaddresses'], row['count_fhrp_groups']) for row in response.data['results']
|
|
|
+ }
|
|
|
+ self.assertEqual(counts, {'Interface 1': (2, 1), 'Interface 2': (0, 0), 'Interface 3': (0, 0)})
|
|
|
+ # Only the interface statement reads the related tables, through its two count subqueries
|
|
|
+ statements = [
|
|
|
+ query['sql'] for query in queries.captured_queries
|
|
|
+ if '"ipam_ipaddress"' in query['sql'] or '"ipam_fhrpgroupassignment"' in query['sql']
|
|
|
+ ]
|
|
|
+ self.assertEqual(len(statements), 1)
|
|
|
+ self.assertTrue(statements[0].startswith('SELECT "dcim_interface"'))
|
|
|
+ # Unannotated instances count through the relations
|
|
|
+ self.assertEqual((interface.count_ipaddresses, interface.count_fhrp_groups), (2, 1))
|
|
|
+
|
|
|
+ def test_count_event_payloads_use_write_prefetches(self):
|
|
|
+ """Write and delete event payloads take both counts from the write prefetches, whatever fields are selected."""
|
|
|
+ self.add_permissions('dcim.view_interface', 'dcim.change_interface', 'dcim.delete_interface')
|
|
|
+ interfaces = Interface.objects.filter(name__in=('Interface 1', 'Interface 2')).order_by('name')
|
|
|
+ interface1, interface2 = interfaces
|
|
|
+ group = FHRPGroup.objects.create(protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=1)
|
|
|
+ IPAddress.objects.bulk_create((
|
|
|
+ IPAddress(address='192.0.2.1/24', assigned_object=interface1),
|
|
|
+ IPAddress(address='192.0.2.2/24', assigned_object=interface1),
|
|
|
+ IPAddress(address='192.0.2.3/24', assigned_object=interface2),
|
|
|
+ ))
|
|
|
+ FHRPGroupAssignment.objects.create(group=group, interface=interface1, priority=100)
|
|
|
+
|
|
|
+ def payload_counts(events):
|
|
|
+ return {
|
|
|
+ event['data']['name']: (event['data']['count_ipaddresses'], event['data']['count_fhrp_groups'])
|
|
|
+ for event in events if isinstance(event['object'], Interface)
|
|
|
+ }
|
|
|
+
|
|
|
+ def fallback_counts(queries):
|
|
|
+ # The count properties' per-interface fallback
|
|
|
+ return [
|
|
|
+ query['sql'] for query in queries.captured_queries
|
|
|
+ if query['sql'].startswith('SELECT COUNT(*)') and (
|
|
|
+ '"ipam_ipaddress"."assigned_object_id"' in query['sql']
|
|
|
+ or '"ipam_fhrpgroupassignment"."interface_id"' in query['sql']
|
|
|
+ )
|
|
|
+ ]
|
|
|
+
|
|
|
+ writes = (
|
|
|
+ ('single', self._get_detail_url(interface1), {'description': 'New'}, {'Interface 1': (2, 1)}),
|
|
|
+ (
|
|
|
+ 'bulk', self._get_list_url(),
|
|
|
+ [{'id': interface.pk, 'description': 'Bulk'} for interface in interfaces],
|
|
|
+ {'Interface 1': (2, 1), 'Interface 2': (1, 0)},
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ for label, url, data, expected in writes:
|
|
|
+ with self.subTest(write=label):
|
|
|
+ with (
|
|
|
+ patch('netbox.context_managers.flush_events') as flush_events,
|
|
|
+ CaptureQueriesContext(connection) as queries,
|
|
|
+ ):
|
|
|
+ response = self.client.patch(f'{url}?fields=id,description', data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ (events,) = flush_events.call_args.args
|
|
|
+ # Update payloads serialize on first access
|
|
|
+ counts = payload_counts(events)
|
|
|
+ self.assertEqual(counts, expected)
|
|
|
+ self.assertEqual(fallback_counts(queries), [])
|
|
|
+
|
|
|
+ # Delete payloads serialize before the rows go, within the request
|
|
|
+ with (
|
|
|
+ patch('netbox.context_managers.flush_events') as flush_events,
|
|
|
+ CaptureQueriesContext(connection) as queries,
|
|
|
+ ):
|
|
|
+ response = self.client.delete(f'{self._get_detail_url(interface1)}?fields=id', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
|
|
|
+ (events,) = flush_events.call_args.args
|
|
|
+ self.assertEqual(payload_counts(events), {'Interface 1': (2, 1)})
|
|
|
+ self.assertEqual(fallback_counts(queries), [])
|
|
|
+
|
|
|
+ def test_pagination_count_excludes_count_subqueries(self):
|
|
|
+ """The pagination count of a DISTINCT-filtered list leaves out the count subqueries."""
|
|
|
+ self.add_permissions('dcim.view_interface')
|
|
|
+ with CaptureQueriesContext(connection) as queries:
|
|
|
+ response = self.client.get(self._get_list_url(), {'name': 'Interface 1'}, **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ self.assertEqual(response.data['count'], 1)
|
|
|
+ (count_sql,) = [
|
|
|
+ query['sql'] for query in queries.captured_queries if query['sql'].startswith('SELECT COUNT(*)')
|
|
|
+ ]
|
|
|
+ # The name filter must make the list DISTINCT for this test to mean anything
|
|
|
+ self.assertIn('DISTINCT', count_sql)
|
|
|
+ self.assertNotIn('"ipam_ipaddress"', count_sql)
|
|
|
+ self.assertNotIn('"ipam_fhrpgroupassignment"', count_sql)
|
|
|
+
|
|
|
|
|
|
class FrontPortTestCase(APIViewTestCases.APIViewTestCase):
|
|
|
model = FrontPort
|
|
|
@@ -5663,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:
|