|
|
@@ -1442,7 +1442,7 @@ class VLANTranslationRuleTestCase(APIViewTestCases.APIViewTestCase):
|
|
|
|
|
|
class ServiceTemplateTestCase(APIViewTestCases.APIViewTestCase):
|
|
|
model = ServiceTemplate
|
|
|
- brief_fields = ['description', 'display', 'id', 'name', 'ports', 'protocol', 'url']
|
|
|
+ brief_fields = ['description', 'display', 'id', 'name', 'port_mappings', 'url']
|
|
|
bulk_update_data = {
|
|
|
'description': 'New description',
|
|
|
}
|
|
|
@@ -1450,35 +1450,283 @@ class ServiceTemplateTestCase(APIViewTestCases.APIViewTestCase):
|
|
|
|
|
|
@classmethod
|
|
|
def setUpTestData(cls):
|
|
|
- service_templates = (
|
|
|
- ServiceTemplate(name='Service Template 1', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[1, 2]),
|
|
|
- ServiceTemplate(name='Service Template 2', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[3, 4]),
|
|
|
- ServiceTemplate(name='Service Template 3', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[5, 6]),
|
|
|
- )
|
|
|
- ServiceTemplate.objects.bulk_create(service_templates)
|
|
|
+ ServiceTemplate.objects.bulk_create([
|
|
|
+ ServiceTemplate(name='Service Template 1', port_mappings=['tcp/1', 'tcp/2']),
|
|
|
+ ServiceTemplate(name='Service Template 2', port_mappings=['tcp/3', 'tcp/4']),
|
|
|
+ ServiceTemplate(name='Service Template 3', port_mappings=['tcp/5', 'tcp/6']),
|
|
|
+ ])
|
|
|
|
|
|
cls.create_data = [
|
|
|
{
|
|
|
'name': 'Service Template 4',
|
|
|
- 'protocol': ServiceProtocolChoices.PROTOCOL_TCP,
|
|
|
- 'ports': [7, 8],
|
|
|
+ 'port_mappings': ['tcp/7', 'tcp/8'],
|
|
|
},
|
|
|
{
|
|
|
'name': 'Service Template 5',
|
|
|
- 'protocol': ServiceProtocolChoices.PROTOCOL_TCP,
|
|
|
- 'ports': [9, 10],
|
|
|
+ 'port_mappings': ['tcp/53', 'udp/53'],
|
|
|
},
|
|
|
{
|
|
|
'name': 'Service Template 6',
|
|
|
- 'protocol': ServiceProtocolChoices.PROTOCOL_TCP,
|
|
|
- 'ports': [11, 12],
|
|
|
+ 'port_mappings': ['tcp/11', 'tcp/12'],
|
|
|
},
|
|
|
]
|
|
|
|
|
|
+ def test_graphql_port_mappings(self):
|
|
|
+ """port_mappings is exposed over GraphQL as a flat list of protocol/port strings."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ template = ServiceTemplate.objects.create(name='GQL Mappings', port_mappings=['tcp/80', 'udp/53'])
|
|
|
+ url = reverse('graphql')
|
|
|
+ query = f'{{ service_template(id: {template.pk}) {{ port_mappings }} }}'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual(data['data']['service_template']['port_mappings'], ['tcp/80', 'udp/53'])
|
|
|
+
|
|
|
+ def test_graphql_protocol_and_port_filter(self):
|
|
|
+ """Combined protocol+port filtering works for ServiceTemplate over GraphQL."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ url = reverse('graphql')
|
|
|
+ query = '{ service_template_list(filters: {protocol: [TCP], port: [1]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ # Only Service Template 1 exposes tcp/1.
|
|
|
+ self.assertEqual([t['name'] for t in data['data']['service_template_list']], ['Service Template 1'])
|
|
|
+
|
|
|
+ def test_graphql_port_only_filter(self):
|
|
|
+ """A port-only GraphQL filter (no protocol) works for ServiceTemplate."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ url = reverse('graphql')
|
|
|
+ query = '{ service_template_list(filters: {port: [3]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ # Only Service Template 2 exposes port 3 (tcp/3).
|
|
|
+ self.assertEqual([t['name'] for t in data['data']['service_template_list']], ['Service Template 2'])
|
|
|
+
|
|
|
+ def test_graphql_port_mappings_filter(self):
|
|
|
+ """The whole-mapping GraphQL filter matches an exact protocol/port pair for ServiceTemplate."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ url = reverse('graphql')
|
|
|
+ query = '{ service_template_list(filters: {port_mappings: ["tcp/3"]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual([t['name'] for t in data['data']['service_template_list']], ['Service Template 2'])
|
|
|
+
|
|
|
+ # udp/3 does not exist, though tcp/3 does
|
|
|
+ query = '{ service_template_list(filters: {port_mappings: ["udp/3"]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual(data['data']['service_template_list'], [])
|
|
|
+
|
|
|
+ def test_graphql_port_range_lookups(self):
|
|
|
+ """The port range lookups are available on ServiceTemplate too, and stay correlated."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ url = reverse('graphql')
|
|
|
+
|
|
|
+ # Templates 1-3 expose tcp/1-2, tcp/3-4 and tcp/5-6 respectively
|
|
|
+ query = '{ service_template_list(filters: {port__gte: [3], port__lte: [4]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual([t['name'] for t in data['data']['service_template_list']], ['Service Template 2'])
|
|
|
+
|
|
|
+ # A protocol which no template exposes narrows the same range to nothing
|
|
|
+ query = '{ service_template_list(filters: {protocol: [UDP], port__gte: [3], port__lte: [4]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual(data['data']['service_template_list'], [])
|
|
|
+
|
|
|
+ def test_create_duplicate_mapping_rejected(self):
|
|
|
+ """A duplicate protocol/port entry is rejected with a clean 400 (not a 500)."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'Duplicate', 'port_mappings': ['tcp/80', 'tcp/80']}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+
|
|
|
+ def test_create_port_out_of_range_rejected(self):
|
|
|
+ """Ports outside SERVICE_PORT_MIN..SERVICE_PORT_MAX are rejected with a 400."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'OutOfRange', 'port_mappings': ['tcp/70000']}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+
|
|
|
+ def test_create_without_port_mappings_rejected(self):
|
|
|
+ """A service (template) must define at least one port mapping (400, not a portless object)."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'Portless', 'port_mappings': []}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+
|
|
|
+ def test_create_normalizes_port_mappings(self):
|
|
|
+ """Input is normalized (e.g. leading zeros stripped) into the model's canonical form."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'Normalized', 'port_mappings': ['tcp/443', 'tcp/080']}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
|
+ template = ServiceTemplate.objects.get(name='Normalized')
|
|
|
+ self.assertEqual(template.port_mappings, ['tcp/443', 'tcp/80'])
|
|
|
+
|
|
|
+ def test_port_mappings_read(self):
|
|
|
+ """port_mappings reads back as the stored flat list of protocol/port strings."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ template = ServiceTemplate.objects.create(name='Mappings', port_mappings=['tcp/443', 'tcp/80', 'udp/53'])
|
|
|
+ response = self.client.get(self._get_detail_url(template), **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ self.assertEqual(response.data['port_mappings'], ['tcp/443', 'tcp/80', 'udp/53'])
|
|
|
+
|
|
|
+ def test_legacy_read_single_protocol(self):
|
|
|
+ """A single-protocol service reports the deprecated protocol/ports fields for compatibility."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ template = ServiceTemplate.objects.create(name='Legacy Single', port_mappings=['tcp/80', 'tcp/443'])
|
|
|
+ response = self.client.get(self._get_detail_url(template), **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ # The legacy protocol field keeps the standard choice-field {value, label} read shape.
|
|
|
+ self.assertEqual(response.data['protocol'], {'value': 'tcp', 'label': 'TCP'})
|
|
|
+ self.assertEqual(response.data['ports'], [80, 443])
|
|
|
+ self.assertEqual(response.data['port_mappings'], ['tcp/80', 'tcp/443'])
|
|
|
+
|
|
|
+ def test_legacy_read_multiple_protocols_null(self):
|
|
|
+ """A multi-protocol service can't be expressed in the old format, so protocol/ports are null."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ template = ServiceTemplate.objects.create(name='Legacy Multi', port_mappings=['tcp/53', 'udp/53'])
|
|
|
+ response = self.client.get(self._get_detail_url(template), **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ self.assertIsNone(response.data['protocol'])
|
|
|
+ self.assertIsNone(response.data['ports'])
|
|
|
+ self.assertEqual(response.data['port_mappings'], ['tcp/53', 'udp/53'])
|
|
|
+
|
|
|
+ def test_legacy_read_empty_distinct_from_multiple(self):
|
|
|
+ """An empty service is distinguishable from a multi-protocol one: ports=[] vs ports=null."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ # A mapping-less template is normally prevented by validation, but can exist via migrated data;
|
|
|
+ # objects.create() bypasses full_clean() so we can exercise the read path here.
|
|
|
+ template = ServiceTemplate.objects.create(name='Legacy Empty', port_mappings=[])
|
|
|
+ response = self.client.get(self._get_detail_url(template), **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ self.assertIsNone(response.data['protocol'])
|
|
|
+ self.assertEqual(response.data['ports'], [])
|
|
|
+ self.assertEqual(response.data['port_mappings'], [])
|
|
|
+
|
|
|
+ def test_create_via_legacy_format(self):
|
|
|
+ """The deprecated protocol/ports format is accepted on write and translated to port_mappings."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'Legacy Create', 'protocol': 'tcp', 'ports': [80, 443]}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
|
+ template = ServiceTemplate.objects.get(name='Legacy Create')
|
|
|
+ self.assertEqual(template.port_mappings, ['tcp/80', 'tcp/443'])
|
|
|
+
|
|
|
+ def test_legacy_empty_ports_reports_at_least_one(self):
|
|
|
+ """
|
|
|
+ A legacy write with an explicitly-empty ports list (allowed by the old API) is rejected with the
|
|
|
+ at-least-one-mapping message keyed to ports, not the misleading "both are required" error.
|
|
|
+ """
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'Legacy Empty', 'protocol': 'tcp', 'ports': []}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+ self.assertIn('ports', response.data)
|
|
|
+
|
|
|
+ def test_create_port_mappings_case_insensitive(self):
|
|
|
+ """port_mappings accepts protocols in any case (e.g. 'TCP/80') and stores the canonical value."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'Case Insensitive', 'port_mappings': ['TCP/80', 'UDP/53']}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
|
+ template = ServiceTemplate.objects.get(name='Case Insensitive')
|
|
|
+ self.assertEqual(template.port_mappings, ['tcp/80', 'udp/53'])
|
|
|
+
|
|
|
+ def test_both_formats_rejected(self):
|
|
|
+ """Supplying both port_mappings and the legacy protocol/ports is ambiguous and must 400."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ # port_mappings is a well-formed flat list so it passes field-level parsing and actually reaches
|
|
|
+ # the validate() mutual-exclusion guard (rather than 400ing on a malformed value first).
|
|
|
+ data = {
|
|
|
+ 'name': 'Both Formats',
|
|
|
+ 'port_mappings': ['udp/53'],
|
|
|
+ 'protocol': 'tcp',
|
|
|
+ 'ports': [80],
|
|
|
+ }
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+ self.assertFalse(ServiceTemplate.objects.filter(name='Both Formats').exists())
|
|
|
+
|
|
|
+ def test_create_legacy_port_out_of_range_rejected(self):
|
|
|
+ """A legacy ports value outside the permitted range is rejected with a 400 (not a 500)."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'Legacy OOR', 'protocol': 'tcp', 'ports': [70000]}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+
|
|
|
+ def test_create_legacy_protocol_without_ports_rejected(self):
|
|
|
+ """One half of the legacy pair is ambiguous and must 400, not silently drop the input."""
|
|
|
+ self.add_permissions('ipam.add_servicetemplate')
|
|
|
+ data = {'name': 'Legacy Half', 'protocol': 'tcp'}
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+
|
|
|
+ def test_update_legacy_ports_only(self):
|
|
|
+ """A partial update supplying only legacy 'ports' keeps the existing single protocol."""
|
|
|
+ self.add_permissions('ipam.change_servicetemplate')
|
|
|
+ template = ServiceTemplate.objects.create(name='Legacy Patch', port_mappings=['tcp/80'])
|
|
|
+ response = self.client.patch(
|
|
|
+ self._get_detail_url(template), {'ports': [8080]}, format='json', **self.header
|
|
|
+ )
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ template.refresh_from_db()
|
|
|
+ self.assertEqual(template.port_mappings, ['tcp/8080'])
|
|
|
+
|
|
|
+ def test_update_legacy_protocol_only(self):
|
|
|
+ """A partial update supplying only legacy 'protocol' keeps the existing ports."""
|
|
|
+ self.add_permissions('ipam.change_servicetemplate')
|
|
|
+ template = ServiceTemplate.objects.create(name='Legacy Patch', port_mappings=['tcp/80', 'tcp/443'])
|
|
|
+ response = self.client.patch(
|
|
|
+ self._get_detail_url(template), {'protocol': 'udp'}, format='json', **self.header
|
|
|
+ )
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ template.refresh_from_db()
|
|
|
+ self.assertEqual(template.port_mappings, ['udp/80', 'udp/443'])
|
|
|
+
|
|
|
+ def test_update_legacy_single_field_multiprotocol_rejected(self):
|
|
|
+ """A single legacy field can't patch a multi-protocol service (no single-protocol form)."""
|
|
|
+ self.add_permissions('ipam.change_servicetemplate')
|
|
|
+ template = ServiceTemplate.objects.create(name='Legacy Patch', port_mappings=['tcp/80', 'udp/53'])
|
|
|
+ response = self.client.patch(
|
|
|
+ self._get_detail_url(template), {'ports': [8080]}, format='json', **self.header
|
|
|
+ )
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+ template.refresh_from_db()
|
|
|
+ self.assertEqual(template.port_mappings, ['tcp/80', 'udp/53'])
|
|
|
+
|
|
|
+ def test_read_malformed_port_mapping_degrades(self):
|
|
|
+ """A malformed stored mapping (validation bypassed) must degrade on API read, not raise a 500."""
|
|
|
+ self.add_permissions('ipam.view_servicetemplate')
|
|
|
+ # objects.create bypasses full_clean, simulating a raw-DB/plugin write of a non-numeric port
|
|
|
+ template = ServiceTemplate.objects.create(name='Malformed', port_mappings=['tcp/80', 'tcp/abc'])
|
|
|
+ response = self.client.get(self._get_detail_url(template), **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ # port_mappings echoes the stored values verbatim (no reformatting). The legacy view can't
|
|
|
+ # faithfully represent a mapping that fails integer coercion, so rather than silently returning
|
|
|
+ # a subset it reports ports=null — the same "not representable" signal used for multi-protocol.
|
|
|
+ self.assertEqual(response.data['port_mappings'], ['tcp/80', 'tcp/abc'])
|
|
|
+ self.assertIsNone(response.data['ports'])
|
|
|
+ self.assertIsNone(response.data['protocol'])
|
|
|
+
|
|
|
|
|
|
class ServiceTestCase(APIViewTestCases.APIViewTestCase):
|
|
|
model = Service
|
|
|
- brief_fields = ['description', 'display', 'id', 'name', 'ports', 'protocol', 'url']
|
|
|
+ brief_fields = ['description', 'display', 'id', 'name', 'port_mappings', 'url']
|
|
|
bulk_update_data = {
|
|
|
'description': 'New description',
|
|
|
}
|
|
|
@@ -1497,33 +1745,254 @@ class ServiceTestCase(APIViewTestCases.APIViewTestCase):
|
|
|
)
|
|
|
Device.objects.bulk_create(devices)
|
|
|
|
|
|
- services = (
|
|
|
- Service(parent=devices[0], name='Service 1', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[1]),
|
|
|
- Service(parent=devices[0], name='Service 2', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[2]),
|
|
|
- Service(parent=devices[0], name='Service 3', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[3]),
|
|
|
- )
|
|
|
- Service.objects.bulk_create(services)
|
|
|
+ Service.objects.bulk_create([
|
|
|
+ Service(parent=devices[0], name='Service 1', port_mappings=['tcp/1']),
|
|
|
+ Service(parent=devices[0], name='Service 2', port_mappings=['tcp/2']),
|
|
|
+ Service(parent=devices[0], name='Service 3', port_mappings=['tcp/3']),
|
|
|
+ ])
|
|
|
|
|
|
cls.create_data = [
|
|
|
{
|
|
|
'parent_object_id': devices[1].pk,
|
|
|
'parent_object_type': 'dcim.device',
|
|
|
'name': 'Service 4',
|
|
|
- 'protocol': ServiceProtocolChoices.PROTOCOL_TCP,
|
|
|
- 'ports': [4],
|
|
|
+ 'port_mappings': ['tcp/4'],
|
|
|
},
|
|
|
{
|
|
|
'parent_object_id': devices[1].pk,
|
|
|
'parent_object_type': 'dcim.device',
|
|
|
- 'name': 'Service 5',
|
|
|
- 'protocol': ServiceProtocolChoices.PROTOCOL_TCP,
|
|
|
- 'ports': [5],
|
|
|
+ 'name': 'dns',
|
|
|
+ 'port_mappings': ['tcp/53', 'udp/53'],
|
|
|
},
|
|
|
{
|
|
|
'parent_object_id': devices[1].pk,
|
|
|
'parent_object_type': 'dcim.device',
|
|
|
'name': 'Service 6',
|
|
|
- 'protocol': ServiceProtocolChoices.PROTOCOL_TCP,
|
|
|
- 'ports': [6],
|
|
|
+ 'port_mappings': ['tcp/6'],
|
|
|
},
|
|
|
]
|
|
|
+
|
|
|
+ def test_graphql_protocol_and_port_filter(self):
|
|
|
+ """Combined protocol + port filtering works over GraphQL (port mappings live in an array)."""
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ url = reverse('graphql')
|
|
|
+ query = '{ service_list(filters: {protocol: [TCP], port: [1]}) { id name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual(len(data['data']['service_list']), 1)
|
|
|
+ self.assertEqual(data['data']['service_list'][0]['name'], 'Service 1')
|
|
|
+
|
|
|
+ def test_graphql_protocol_and_port_filter_multiprotocol(self):
|
|
|
+ """
|
|
|
+ A combined protocol+port filter must match a single mapping, not protocol and port matched
|
|
|
+ independently across different mappings on the same object (GraphQL parity with the FilterSet).
|
|
|
+ """
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ device = Device.objects.first()
|
|
|
+ Service.objects.create(parent=device, name='dns-multi', port_mappings=['tcp/8080', 'udp/53'])
|
|
|
+ url = reverse('graphql')
|
|
|
+
|
|
|
+ # tcp/8080 exists on the service -> matches
|
|
|
+ query = '{ service_list(filters: {protocol: [TCP], port: [8080]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual([s['name'] for s in data['data']['service_list']], ['dns-multi'])
|
|
|
+
|
|
|
+ # udp/8080 does not exist, even though the service has udp (on 53) and 8080 (on tcp)
|
|
|
+ query = '{ service_list(filters: {protocol: [UDP], port: [8080]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual(data['data']['service_list'], [])
|
|
|
+
|
|
|
+ def test_graphql_port_mappings(self):
|
|
|
+ """port_mappings is exposed over GraphQL as a flat list of protocol/port strings."""
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ device = Device.objects.first()
|
|
|
+ service = Service.objects.create(parent=device, name='GQL Mappings', port_mappings=['tcp/80', 'udp/53'])
|
|
|
+ url = reverse('graphql')
|
|
|
+ query = f'{{ service(id: {service.pk}) {{ port_mappings }} }}'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual(data['data']['service']['port_mappings'], ['tcp/80', 'udp/53'])
|
|
|
+
|
|
|
+ def test_graphql_port_only_filter(self):
|
|
|
+ """A port-only GraphQL filter (no protocol) matches the port across any protocol."""
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ device = Device.objects.first()
|
|
|
+ Service.objects.create(parent=device, name='udp-on-1', port_mappings=['udp/1'])
|
|
|
+ url = reverse('graphql')
|
|
|
+ query = '{ service_list(filters: {port: [1]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ # Service 1 (tcp/1) and the new udp-on-1 both expose port 1, on different protocols.
|
|
|
+ self.assertEqual({s['name'] for s in data['data']['service_list']}, {'Service 1', 'udp-on-1'})
|
|
|
+
|
|
|
+ def test_graphql_port_mappings_filter(self):
|
|
|
+ """The whole-mapping GraphQL filter matches an exact protocol/port pair, OR'd across values."""
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ device = Device.objects.first()
|
|
|
+ Service.objects.create(parent=device, name='udp-on-1', port_mappings=['udp/1'])
|
|
|
+ url = reverse('graphql')
|
|
|
+
|
|
|
+ # tcp/1 must not match the udp-only service, even though both expose port 1
|
|
|
+ query = '{ service_list(filters: {port_mappings: ["tcp/1"]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual([s['name'] for s in data['data']['service_list']], ['Service 1'])
|
|
|
+
|
|
|
+ # Multiple values are OR'd, and input is normalized ('UDP/001' -> 'udp/1')
|
|
|
+ query = '{ service_list(filters: {port_mappings: ["tcp/1", "UDP/001"]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual({s['name'] for s in data['data']['service_list']}, {'Service 1', 'udp-on-1'})
|
|
|
+
|
|
|
+ def test_graphql_protocol_only_filter(self):
|
|
|
+ """A protocol-only GraphQL filter matches services exposing that protocol on any port."""
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ device = Device.objects.first()
|
|
|
+ Service.objects.create(parent=device, name='udp-svc', port_mappings=['udp/9'])
|
|
|
+ url = reverse('graphql')
|
|
|
+ query = '{ service_list(filters: {protocol: [UDP]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ # Only the udp service matches; the seeded Service 1-3 are all tcp.
|
|
|
+ self.assertEqual([s['name'] for s in data['data']['service_list']], ['udp-svc'])
|
|
|
+
|
|
|
+ def test_graphql_port_range_lookups(self):
|
|
|
+ """The port__gt/gte/lt/lte GraphQL lookups mirror their identically-named REST counterparts."""
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ url = reverse('graphql')
|
|
|
+
|
|
|
+ # Seeded services expose tcp/1, tcp/2 and tcp/3 respectively
|
|
|
+ for filters, expected in (
|
|
|
+ ('{port__gt: [2]}', {'Service 3'}),
|
|
|
+ ('{port__gte: [2]}', {'Service 2', 'Service 3'}),
|
|
|
+ ('{port__lt: [2]}', {'Service 1'}),
|
|
|
+ ('{port__lte: [2]}', {'Service 1', 'Service 2'}),
|
|
|
+ ):
|
|
|
+ query = f'{{ service_list(filters: {filters}) {{ name }} }}'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual({s['name'] for s in data['data']['service_list']}, expected, msg=filters)
|
|
|
+
|
|
|
+ def test_graphql_port_range_bounds_correlated(self):
|
|
|
+ """
|
|
|
+ Combined range bounds must be satisfied by a *single* mapping, so a service straddling the range
|
|
|
+ without any port inside it does not match (GraphQL parity with the FilterSet).
|
|
|
+ """
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ device = Device.objects.first()
|
|
|
+ Service.objects.create(parent=device, name='straddles', port_mappings=['tcp/500', 'tcp/5000'])
|
|
|
+ Service.objects.create(parent=device, name='inside', port_mappings=['tcp/1500'])
|
|
|
+ url = reverse('graphql')
|
|
|
+
|
|
|
+ query = '{ service_list(filters: {port__gte: [1000], port__lte: [2000]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual([s['name'] for s in data['data']['service_list']], ['inside'])
|
|
|
+
|
|
|
+ def test_graphql_protocol_and_port_range_correlated(self):
|
|
|
+ """A protocol combined with a range lookup must also be satisfied by a single mapping."""
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ device = Device.objects.first()
|
|
|
+ Service.objects.create(parent=device, name='mixed', port_mappings=['tcp/80', 'udp/9999'])
|
|
|
+ Service.objects.create(parent=device, name='tcp-high', port_mappings=['tcp/9999'])
|
|
|
+ url = reverse('graphql')
|
|
|
+
|
|
|
+ # 'mixed' has a tcp mapping and a mapping above 1000, but no tcp mapping above 1000
|
|
|
+ query = '{ service_list(filters: {protocol: [TCP], port__gt: [1000]}) { name } }'
|
|
|
+ response = self.client.post(url, data={'query': query}, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ data = json.loads(response.content)
|
|
|
+ self.assertNotIn('errors', data)
|
|
|
+ self.assertEqual([s['name'] for s in data['data']['service_list']], ['tcp-high'])
|
|
|
+
|
|
|
+ def test_port_mapping_prefix_branch(self):
|
|
|
+ """
|
|
|
+ The nested-relation (prefix) branch of the shared port filter resolves matches through a
|
|
|
+ relation. No GraphQL type currently exposes a nested Service filter, so exercise the helper
|
|
|
+ directly via the IPAddress -> services reverse relation.
|
|
|
+ """
|
|
|
+ from ipam.graphql.filters import _port_mapping_prefix_q
|
|
|
+
|
|
|
+ device = Device.objects.first()
|
|
|
+ service = Service.objects.create(parent=device, name='svc-with-ip', port_mappings=['tcp/1'])
|
|
|
+ ip = IPAddress.objects.create(address='192.0.2.1/32')
|
|
|
+ service.ipaddresses.add(ip)
|
|
|
+
|
|
|
+ match = _port_mapping_prefix_q(Service, ['tcp'], [('exact', [1])], 'services__')
|
|
|
+ self.assertIn(ip, IPAddress.objects.filter(match))
|
|
|
+ miss = _port_mapping_prefix_q(Service, ['tcp'], [('exact', [999])], 'services__')
|
|
|
+ self.assertNotIn(ip, IPAddress.objects.filter(miss))
|
|
|
+
|
|
|
+ def test_update_full_body_roundtrip(self):
|
|
|
+ """
|
|
|
+ A full-object round-trip (GET then PUT of the same body, including the legacy protocol/ports the
|
|
|
+ read emitted alongside port_mappings) succeeds; only a genuine conflict is rejected.
|
|
|
+ """
|
|
|
+ self.add_permissions('ipam.view_service', 'ipam.change_service')
|
|
|
+ service = Service.objects.get(name='Service 1') # tcp/1
|
|
|
+ read = self.client.get(self._get_detail_url(service), **self.header).data
|
|
|
+ put_data = {
|
|
|
+ 'parent_object_type': 'dcim.device',
|
|
|
+ 'parent_object_id': service.parent_object_id,
|
|
|
+ 'name': service.name,
|
|
|
+ 'port_mappings': read['port_mappings'],
|
|
|
+ # The legacy protocol field reads as {value, label}; on write NetBox choice fields take the
|
|
|
+ # raw value, so a well-behaved round-trip resubmits read['protocol']['value'].
|
|
|
+ 'protocol': read['protocol']['value'],
|
|
|
+ 'ports': read['ports'],
|
|
|
+ }
|
|
|
+ response = self.client.put(self._get_detail_url(service), put_data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+
|
|
|
+ # A legacy field that disagrees with port_mappings is still rejected as a conflict.
|
|
|
+ put_data['protocol'] = 'udp'
|
|
|
+ response = self.client.put(self._get_detail_url(service), put_data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
+
|
|
|
+ def test_legacy_read_single_protocol(self):
|
|
|
+ """A single-protocol service reports the deprecated protocol/ports fields for compatibility."""
|
|
|
+ self.add_permissions('ipam.view_service')
|
|
|
+ service = Service.objects.get(name='Service 1') # port_mappings=['tcp/1']
|
|
|
+ response = self.client.get(self._get_detail_url(service), **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
+ # The legacy protocol field keeps the standard choice-field {value, label} read shape.
|
|
|
+ self.assertEqual(response.data['protocol'], {'value': 'tcp', 'label': 'TCP'})
|
|
|
+ self.assertEqual(response.data['ports'], [1])
|
|
|
+
|
|
|
+ def test_create_via_legacy_format(self):
|
|
|
+ """The deprecated protocol/ports format is accepted on write and translated to port_mappings."""
|
|
|
+ self.add_permissions('ipam.add_service')
|
|
|
+ device = Device.objects.first()
|
|
|
+ data = {
|
|
|
+ 'parent_object_type': 'dcim.device',
|
|
|
+ 'parent_object_id': device.pk,
|
|
|
+ 'name': 'Legacy Service',
|
|
|
+ 'protocol': 'udp',
|
|
|
+ 'ports': [53, 67],
|
|
|
+ }
|
|
|
+ response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
|
|
+ self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
|
+ service = Service.objects.get(name='Legacy Service')
|
|
|
+ self.assertEqual(service.port_mappings, ['udp/53', 'udp/67'])
|