mirror of
https://github.com/netbox-community/netbox.git
synced 2026-03-13 22:07:17 +01:00
397 lines
16 KiB
Python
397 lines
16 KiB
Python
from django.test import Client, TestCase, override_settings, tag
|
|
from django.urls import reverse
|
|
from drf_spectacular.drainage import GENERATOR_STATS
|
|
from rest_framework import status
|
|
|
|
from core.models import ObjectType
|
|
from dcim.models import Region, Site
|
|
from extras.choices import CustomFieldTypeChoices
|
|
from extras.models import CustomField
|
|
from ipam.models import VLAN
|
|
from netbox.config import get_config
|
|
from utilities.api import get_view_name
|
|
from utilities.testing import APITestCase, disable_warnings
|
|
|
|
|
|
class WritableNestedSerializerTest(APITestCase):
|
|
"""
|
|
Test the operation of WritableNestedSerializer using VLANSerializer as our test subject.
|
|
"""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
|
|
self.region_a = Region.objects.create(name='Region A', slug='region-a')
|
|
self.site1 = Site.objects.create(region=self.region_a, name='Site 1', slug='site-1')
|
|
self.site2 = Site.objects.create(region=self.region_a, name='Site 2', slug='site-2')
|
|
|
|
def test_related_by_pk(self):
|
|
data = {
|
|
'vid': 100,
|
|
'name': 'Test VLAN 100',
|
|
'site': self.site1.pk,
|
|
}
|
|
url = reverse('ipam-api:vlan-list')
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data['site']['id'], self.site1.pk)
|
|
vlan = VLAN.objects.get(pk=response.data['id'])
|
|
self.assertEqual(vlan.site, self.site1)
|
|
|
|
def test_related_by_pk_no_match(self):
|
|
data = {
|
|
'vid': 100,
|
|
'name': 'Test VLAN 100',
|
|
'site': 999,
|
|
}
|
|
url = reverse('ipam-api:vlan-list')
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
with disable_warnings('django.request'):
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
self.assertEqual(VLAN.objects.count(), 0)
|
|
self.assertTrue(response.data['site'][0].startswith("Related object not found"))
|
|
|
|
def test_related_by_attributes(self):
|
|
data = {
|
|
'vid': 100,
|
|
'name': 'Test VLAN 100',
|
|
'site': {
|
|
'name': 'Site 1'
|
|
},
|
|
}
|
|
url = reverse('ipam-api:vlan-list')
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data['site']['id'], self.site1.pk)
|
|
vlan = VLAN.objects.get(pk=response.data['id'])
|
|
self.assertEqual(vlan.site, self.site1)
|
|
|
|
def test_related_by_attributes_no_match(self):
|
|
data = {
|
|
'vid': 100,
|
|
'name': 'Test VLAN 100',
|
|
'site': {
|
|
'name': 'Site X'
|
|
},
|
|
}
|
|
url = reverse('ipam-api:vlan-list')
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
with disable_warnings('django.request'):
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
self.assertEqual(VLAN.objects.count(), 0)
|
|
self.assertTrue(response.data['site'][0].startswith("Related object not found"))
|
|
|
|
def test_related_by_attributes_multiple_matches(self):
|
|
data = {
|
|
'vid': 100,
|
|
'name': 'Test VLAN 100',
|
|
'site': {
|
|
'region': {
|
|
"name": "Region A",
|
|
},
|
|
},
|
|
}
|
|
url = reverse('ipam-api:vlan-list')
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
with disable_warnings('django.request'):
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
self.assertEqual(VLAN.objects.count(), 0)
|
|
self.assertTrue(response.data['site'][0].startswith("Multiple objects match"))
|
|
|
|
def test_related_by_invalid(self):
|
|
data = {
|
|
'vid': 100,
|
|
'name': 'Test VLAN 100',
|
|
'site': 'XXX',
|
|
}
|
|
url = reverse('ipam-api:vlan-list')
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
with disable_warnings('django.request'):
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
self.assertEqual(VLAN.objects.count(), 0)
|
|
|
|
|
|
class APIPaginationTestCase(APITestCase):
|
|
user_permissions = ('dcim.view_site',)
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.url = reverse('dcim-api:site-list')
|
|
|
|
# Create a large number of Sites for testing
|
|
Site.objects.bulk_create([
|
|
Site(name=f'Site {i}', slug=f'site-{i}') for i in range(1, 101)
|
|
])
|
|
|
|
def test_default_page_size(self):
|
|
response = self.client.get(self.url, format='json', **self.header)
|
|
page_size = get_config().PAGINATE_COUNT
|
|
self.assertLess(page_size, 100, "Default page size not sufficient for data set")
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 100)
|
|
self.assertTrue(response.data['next'].endswith(f'?limit={page_size}&offset={page_size}'))
|
|
self.assertIsNone(response.data['previous'])
|
|
self.assertEqual(len(response.data['results']), page_size)
|
|
|
|
@override_settings(MAX_PAGE_SIZE=30)
|
|
def test_default_page_size_with_small_max_page_size(self):
|
|
response = self.client.get(self.url, format='json', **self.header)
|
|
page_size = get_config().MAX_PAGE_SIZE
|
|
self.assertLess(page_size, 100, "Default page size not sufficient for data set")
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 100)
|
|
self.assertTrue(response.data['next'].endswith(f'?limit={page_size}&offset={page_size}'))
|
|
self.assertIsNone(response.data['previous'])
|
|
self.assertEqual(len(response.data['results']), page_size)
|
|
|
|
def test_custom_page_size(self):
|
|
response = self.client.get(f'{self.url}?limit=10', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 100)
|
|
self.assertTrue(response.data['next'].endswith('?limit=10&offset=10'))
|
|
self.assertIsNone(response.data['previous'])
|
|
self.assertEqual(len(response.data['results']), 10)
|
|
|
|
@override_settings(MAX_PAGE_SIZE=80)
|
|
def test_max_page_size(self):
|
|
response = self.client.get(f'{self.url}?limit=0', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 100)
|
|
self.assertTrue(response.data['next'].endswith('?limit=80&offset=80'))
|
|
self.assertIsNone(response.data['previous'])
|
|
self.assertEqual(len(response.data['results']), 80)
|
|
|
|
@override_settings(MAX_PAGE_SIZE=0)
|
|
def test_max_page_size_disabled(self):
|
|
response = self.client.get(f'{self.url}?limit=0', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 100)
|
|
self.assertIsNone(response.data['next'])
|
|
self.assertIsNone(response.data['previous'])
|
|
self.assertEqual(len(response.data['results']), 100)
|
|
|
|
def test_cursor_pagination(self):
|
|
"""Basic cursor pagination returns results ordered by PK with correct next link."""
|
|
first_pk = Site.objects.order_by('pk').values_list('pk', flat=True).first()
|
|
response = self.client.get(f'{self.url}?start={first_pk}&limit=10', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertIsNone(response.data['count'])
|
|
self.assertIsNone(response.data['previous'])
|
|
self.assertEqual(len(response.data['results']), 10)
|
|
|
|
# Results should be ordered by PK
|
|
pks = [r['id'] for r in response.data['results']]
|
|
self.assertEqual(pks, sorted(pks))
|
|
|
|
# Next link should use start parameter
|
|
last_pk = pks[-1]
|
|
self.assertIn(f'start={last_pk + 1}', response.data['next'])
|
|
self.assertIn('limit=10', response.data['next'])
|
|
|
|
def test_cursor_pagination_last_page(self):
|
|
"""Cursor pagination returns null next link when fewer results than limit."""
|
|
last_pk = Site.objects.order_by('pk').values_list('pk', flat=True).last()
|
|
response = self.client.get(f'{self.url}?start={last_pk}&limit=10', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(len(response.data['results']), 1)
|
|
self.assertIsNone(response.data['next'])
|
|
self.assertIsNone(response.data['previous'])
|
|
|
|
def test_cursor_pagination_no_results(self):
|
|
"""Cursor pagination beyond all PKs returns empty results."""
|
|
max_pk = Site.objects.order_by('pk').values_list('pk', flat=True).last()
|
|
response = self.client.get(f'{self.url}?start={max_pk + 1000}&limit=10', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(len(response.data['results']), 0)
|
|
self.assertIsNone(response.data['next'])
|
|
|
|
def test_cursor_and_offset_conflict(self):
|
|
"""Specifying both start and offset returns a 400 error."""
|
|
with disable_warnings('django.request'):
|
|
response = self.client.get(f'{self.url}?start=1&offset=10', format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_cursor_and_ordering_conflict(self):
|
|
"""Specifying both start and ordering returns a 400 error."""
|
|
with disable_warnings('django.request'):
|
|
response = self.client.get(f'{self.url}?start=1&ordering=name', format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_cursor_negative_start(self):
|
|
"""Negative start value returns a 400 error."""
|
|
with disable_warnings('django.request'):
|
|
response = self.client.get(f'{self.url}?start=-1', format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_cursor_with_filters(self):
|
|
"""Cursor pagination works alongside other query filters."""
|
|
response = self.client.get(f'{self.url}?start=0&limit=10&name=Site 1', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertIsNone(response.data['count'])
|
|
results = response.data['results']
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0]['name'], 'Site 1')
|
|
|
|
def test_offset_multi_page_traversal(self):
|
|
"""Traverse all 100 objects using offset pagination and verify complete, non-overlapping coverage."""
|
|
collected_pks = []
|
|
url = f'{self.url}?limit=10'
|
|
|
|
while url:
|
|
response = self.client.get(url, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 100)
|
|
collected_pks.extend(r['id'] for r in response.data['results'])
|
|
url = response.data['next']
|
|
|
|
# Should have collected exactly 100 unique objects
|
|
self.assertEqual(len(set(collected_pks)), 100)
|
|
|
|
def test_cursor_multi_page_traversal(self):
|
|
"""Traverse all 100 objects using cursor pagination and verify complete, non-overlapping coverage."""
|
|
collected_pks = []
|
|
first_pk = Site.objects.order_by('pk').values_list('pk', flat=True).first()
|
|
url = f'{self.url}?start={first_pk}&limit=10'
|
|
|
|
while url:
|
|
response = self.client.get(url, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertIsNone(response.data['count'])
|
|
self.assertIsNone(response.data['previous'])
|
|
|
|
page_pks = [r['id'] for r in response.data['results']]
|
|
|
|
# Each page should be ordered by PK
|
|
self.assertEqual(page_pks, sorted(page_pks))
|
|
|
|
# No overlap with previously collected PKs
|
|
self.assertFalse(set(page_pks) & set(collected_pks))
|
|
|
|
collected_pks.extend(page_pks)
|
|
url = response.data['next']
|
|
|
|
# Should have collected exactly 100 unique objects
|
|
self.assertEqual(len(set(collected_pks)), 100)
|
|
|
|
# Full result set should be in PK order
|
|
self.assertEqual(collected_pks, sorted(collected_pks))
|
|
|
|
|
|
class APIOrderingTestCase(APITestCase):
|
|
user_permissions = ('dcim.view_site',)
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.url = reverse('dcim-api:site-list')
|
|
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1', facility='C', description='Z'),
|
|
Site(name='Site 2', slug='site-2', facility='C', description='Y'),
|
|
Site(name='Site 3', slug='site-3', facility='B', description='X'),
|
|
Site(name='Site 4', slug='site-4', facility='B', description='W'),
|
|
Site(name='Site 5', slug='site-5', facility='A', description='V'),
|
|
Site(name='Site 6', slug='site-6', facility='A', description='U'),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
def test_default_order(self):
|
|
response = self.client.get(self.url, format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 6)
|
|
self.assertListEqual(
|
|
[s['name'] for s in response.data['results']],
|
|
['Site 1', 'Site 2', 'Site 3', 'Site 4', 'Site 5', 'Site 6']
|
|
)
|
|
|
|
def test_order_single_field(self):
|
|
response = self.client.get(f'{self.url}?ordering=description', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 6)
|
|
self.assertListEqual(
|
|
[s['name'] for s in response.data['results']],
|
|
['Site 6', 'Site 5', 'Site 4', 'Site 3', 'Site 2', 'Site 1']
|
|
)
|
|
|
|
def test_order_reversed(self):
|
|
response = self.client.get(f'{self.url}?ordering=-name', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 6)
|
|
self.assertListEqual(
|
|
[s['name'] for s in response.data['results']],
|
|
['Site 6', 'Site 5', 'Site 4', 'Site 3', 'Site 2', 'Site 1']
|
|
)
|
|
|
|
def test_order_multiple_fields(self):
|
|
response = self.client.get(f'{self.url}?ordering=facility,name', format='json', **self.header)
|
|
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['count'], 6)
|
|
self.assertListEqual(
|
|
[s['name'] for s in response.data['results']],
|
|
['Site 5', 'Site 6', 'Site 3', 'Site 4', 'Site 1', 'Site 2']
|
|
)
|
|
|
|
|
|
class APIDocsTestCase(TestCase):
|
|
|
|
def setUp(self):
|
|
self.client = Client()
|
|
|
|
# Populate a CustomField to activate CustomFieldSerializer
|
|
object_type = ObjectType.objects.get_for_model(Site)
|
|
self.cf_text = CustomField(type=CustomFieldTypeChoices.TYPE_TEXT, name='test')
|
|
self.cf_text.save()
|
|
self.cf_text.object_types.set([object_type])
|
|
self.cf_text.save()
|
|
|
|
def test_api_docs(self):
|
|
|
|
url = reverse('api_docs')
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
url = reverse('schema')
|
|
with GENERATOR_STATS.silence(): # Suppress schema generator warnings
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
|
|
class GetViewNameTestCase(TestCase):
|
|
|
|
@tag('regression')
|
|
def test_get_view_name_with_none_queryset(self):
|
|
from rest_framework.viewsets import ReadOnlyModelViewSet
|
|
|
|
class MockViewSet(ReadOnlyModelViewSet):
|
|
queryset = None
|
|
|
|
view = MockViewSet()
|
|
view.suffix = 'List'
|
|
|
|
name = get_view_name(view)
|
|
self.assertEqual(name, 'Mock List')
|