mirror of
https://github.com/netbox-community/netbox.git
synced 2026-02-04 14:26:25 -06:00
266 lines
10 KiB
Python
266 lines
10 KiB
Python
import json
|
|
|
|
from django.test import override_settings
|
|
from django.urls import reverse
|
|
from rest_framework import status
|
|
|
|
from dcim.choices import LocationStatusChoices
|
|
from dcim.models import Site, Location
|
|
from utilities.testing import disable_warnings, APITestCase, TestCase
|
|
|
|
|
|
class GraphQLTestCase(TestCase):
|
|
|
|
@override_settings(GRAPHQL_ENABLED=False)
|
|
def test_graphql_enabled(self):
|
|
"""
|
|
The /graphql URL should return a 404 when GRAPHQL_ENABLED=False
|
|
"""
|
|
url = reverse('graphql')
|
|
response = self.client.get(url)
|
|
self.assertHttpStatus(response, 404)
|
|
|
|
@override_settings(LOGIN_REQUIRED=True)
|
|
def test_graphiql_interface(self):
|
|
"""
|
|
Test rendering of the GraphiQL interactive web interface
|
|
"""
|
|
url = reverse('graphql')
|
|
header = {
|
|
'HTTP_ACCEPT': 'text/html',
|
|
}
|
|
|
|
# Authenticated request
|
|
response = self.client.get(url, **header)
|
|
self.assertHttpStatus(response, 200)
|
|
|
|
# Non-authenticated request
|
|
self.client.logout()
|
|
response = self.client.get(url, **header)
|
|
with disable_warnings('django.request'):
|
|
self.assertHttpStatus(response, 302) # Redirect to login page
|
|
|
|
|
|
class GraphQLAPITestCase(APITestCase):
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1'),
|
|
Site(name='Site 2', slug='site-2'),
|
|
Site(name='Site 3', slug='site-3'),
|
|
Site(name='Site 4', slug='site-4'),
|
|
Site(name='Site 5', slug='site-5'),
|
|
Site(name='Site 6', slug='site-6'),
|
|
Site(name='Site 7', slug='site-7'),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
@override_settings(LOGIN_REQUIRED=True)
|
|
def test_graphql_filter_objects(self):
|
|
"""
|
|
Test the operation of filters for GraphQL API requests.
|
|
"""
|
|
self.add_permissions('dcim.view_site', 'dcim.view_location')
|
|
url = reverse('graphql')
|
|
|
|
sites = Site.objects.all()[:3]
|
|
Location.objects.create(
|
|
site=sites[0],
|
|
name='Location 1',
|
|
slug='location-1',
|
|
status=LocationStatusChoices.STATUS_PLANNED
|
|
),
|
|
Location.objects.create(
|
|
site=sites[1],
|
|
name='Location 2',
|
|
slug='location-2',
|
|
status=LocationStatusChoices.STATUS_STAGING
|
|
),
|
|
Location.objects.create(
|
|
site=sites[1],
|
|
name='Location 3',
|
|
slug='location-3',
|
|
status=LocationStatusChoices.STATUS_ACTIVE
|
|
),
|
|
|
|
# A valid request should return the filtered list
|
|
query = '{location_list(filters: {site_id: "' + str(sites[0].pk) + '"}) {id site {id}}}'
|
|
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']['location_list']), 1)
|
|
self.assertIsNotNone(data['data']['location_list'][0]['site'])
|
|
|
|
# Test OR and exact logic
|
|
query = """{
|
|
location_list( filters: {
|
|
status: {exact: STATUS_PLANNED},
|
|
OR: {status: {exact: STATUS_STAGING}}
|
|
}) {
|
|
id site {id}
|
|
}
|
|
}"""
|
|
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']['location_list']), 2)
|
|
|
|
# Test in_list logic
|
|
query = """{
|
|
location_list( filters: {
|
|
status: {in_list: [STATUS_PLANNED, STATUS_STAGING]}
|
|
}) {
|
|
id site {id}
|
|
}
|
|
}"""
|
|
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']['location_list']), 2)
|
|
|
|
# An invalid request should return an empty list
|
|
query = '{location_list(filters: {site_id: "99999"}) {id site {id}}}' # Invalid site ID
|
|
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.assertEqual(len(data['data']['location_list']), 0)
|
|
|
|
# Removing the permissions from location should result in an empty locations list
|
|
self.remove_permissions('dcim.view_location')
|
|
query = '{site(id: ' + str(sites[0].pk) + ') {id locations {id}}}'
|
|
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']['site']['locations']), 0)
|
|
|
|
def test_offset_pagination(self):
|
|
self.add_permissions('dcim.view_site')
|
|
url = reverse('graphql')
|
|
|
|
# Test `limit` only
|
|
query = """
|
|
{
|
|
site_list(pagination: {limit: 3}) {
|
|
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']['site_list']), 3)
|
|
self.assertEqual(data['data']['site_list'][0]['name'], 'Site 1')
|
|
self.assertEqual(data['data']['site_list'][1]['name'], 'Site 2')
|
|
self.assertEqual(data['data']['site_list'][2]['name'], 'Site 3')
|
|
|
|
# Test `offset` only
|
|
query = """
|
|
{
|
|
site_list(pagination: {offset: 3}) {
|
|
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']['site_list']), 4)
|
|
self.assertEqual(data['data']['site_list'][0]['name'], 'Site 4')
|
|
self.assertEqual(data['data']['site_list'][1]['name'], 'Site 5')
|
|
self.assertEqual(data['data']['site_list'][2]['name'], 'Site 6')
|
|
self.assertEqual(data['data']['site_list'][3]['name'], 'Site 7')
|
|
|
|
# Test `offset` & `limit`
|
|
query = """
|
|
{
|
|
site_list(pagination: {offset: 3, limit: 3}) {
|
|
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']['site_list']), 3)
|
|
self.assertEqual(data['data']['site_list'][0]['name'], 'Site 4')
|
|
self.assertEqual(data['data']['site_list'][1]['name'], 'Site 5')
|
|
self.assertEqual(data['data']['site_list'][2]['name'], 'Site 6')
|
|
|
|
def test_cursor_pagination(self):
|
|
self.add_permissions('dcim.view_site')
|
|
url = reverse('graphql')
|
|
|
|
# Page 1
|
|
query = """
|
|
{
|
|
site_list(pagination: {start: 0, limit: 3}) {
|
|
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']['site_list']), 3)
|
|
self.assertEqual(data['data']['site_list'][0]['name'], 'Site 1')
|
|
self.assertEqual(data['data']['site_list'][1]['name'], 'Site 2')
|
|
self.assertEqual(data['data']['site_list'][2]['name'], 'Site 3')
|
|
|
|
# Page 2
|
|
start_id = int(data['data']['site_list'][-1]['id']) + 1
|
|
query = """
|
|
{
|
|
site_list(pagination: {start: """ + str(start_id) + """, limit: 3}) {
|
|
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']['site_list']), 3)
|
|
self.assertEqual(data['data']['site_list'][0]['name'], 'Site 4')
|
|
self.assertEqual(data['data']['site_list'][1]['name'], 'Site 5')
|
|
self.assertEqual(data['data']['site_list'][2]['name'], 'Site 6')
|
|
|
|
# Page 3
|
|
start_id = int(data['data']['site_list'][-1]['id']) + 1
|
|
query = """
|
|
{
|
|
site_list(pagination: {start: """ + str(start_id) + """, limit: 3}) {
|
|
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']['site_list']), 1)
|
|
self.assertEqual(data['data']['site_list'][0]['name'], 'Site 7')
|
|
|
|
def test_pagination_conflict(self):
|
|
url = reverse('graphql')
|
|
query = """
|
|
{
|
|
site_list(pagination: {start: 1, offset: 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.assertIn('errors', data)
|
|
self.assertEqual(data['errors'][0]['message'], 'Cannot specify both `start` and `offset` in pagination.')
|