NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-25 01:48:38 -06:00
Code Issues Actions 9 Packages Projects Releases Wiki Activity

Merge branch 'develop' into develop-2.9

Browse Source
This commit is contained in:
Jeremy Stretch 2020-06-08 10:33:23 -04:00
commit 0ebd87bcb9
19 changed files with 2168 additions and 5950 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -9,6 +9,7 @@
/netbox/static /netbox/static
/venv/ /venv/
/*.sh /*.sh
local_requirements.txt
!upgrade.sh !upgrade.sh
fabfile.py fabfile.py
gunicorn.py gunicorn.py

2
docs/additional-features/custom-links.md
View File

@ -24,7 +24,7 @@ Only links which render with non-empty text are included on the page. You can em
For example, if you only want to display a link for active devices, you could set the link text to For example, if you only want to display a link for active devices, you could set the link text to
``` ```
{% if obj.status == 1 %}View NMS{% endif %} {% if obj.status == 'active' %}View NMS{% endif %}
``` ```
The link will not appear when viewing a device with any status other than "active." The link will not appear when viewing a device with any status other than "active."

14
docs/release-notes/version-2.8.md
View File

@ -1,5 +1,19 @@
# NetBox v2.8 # NetBox v2.8
## v2.8.6 (FUTURE)
### Enhancements
* [#4698](https://github.com/netbox-community/netbox/issues/4698) - Improve display of template code for object in admin UI
### Bug Fixes
* [#4702](https://github.com/netbox-community/netbox/issues/4702) - Catch IntegrityError exception when adding a non-unique secret
* [#4707](https://github.com/netbox-community/netbox/issues/4707) - Fix `prefix_count` population on VLAN API serializer
* [#4725](https://github.com/netbox-community/netbox/issues/4725) - Fix "brief" rendering of various REST API endpoints
---
## v2.8.5 (2020-05-26) ## v2.8.5 (2020-05-26)
**Note:** The minimum required version of PostgreSQL is now 9.6. **Note:** The minimum required version of PostgreSQL is now 9.6.

513
netbox/circuits/tests/test_api.py
View File

@ -1,443 +1,188 @@
from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.models import ContentType
from django.urls import reverse from django.urls import reverse
from rest_framework import status
from circuits.choices import * from circuits.choices import *
from circuits.models import Circuit, CircuitTermination, CircuitType, Provider from circuits.models import Circuit, CircuitTermination, CircuitType, Provider
from dcim.models import Site from dcim.models import Site
from extras.models import Graph from extras.models import Graph
from utilities.testing import APITestCase from utilities.testing import APITestCase, APIViewTestCases
class AppTest(APITestCase): class AppTest(APITestCase):
def test_root(self): def test_root(self):
url = reverse('circuits-api:api-root') url = reverse('circuits-api:api-root')
response = self.client.get('{}?format=api'.format(url), **self.header) response = self.client.get('{}?format=api'.format(url), **self.header)
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
class ProviderTest(APITestCase): class ProviderTest(APIViewTestCases.APIViewTestCase):
model = Provider
brief_fields = ['circuit_count', 'id', 'name', 'slug', 'url']
create_data = [
{
'name': 'Provider 4',
'slug': 'provider-4',
},
{
'name': 'Provider 5',
'slug': 'provider-5',
},
{
'name': 'Provider 6',
'slug': 'provider-6',
},
]
def setUp(self): @classmethod
def setUpTestData(cls):
super().setUp() providers = (
Provider(name='Provider 1', slug='provider-1'),
self.provider1 = Provider.objects.create(name='Test Provider 1', slug='test-provider-1') Provider(name='Provider 2', slug='provider-2'),
self.provider2 = Provider.objects.create(name='Test Provider 2', slug='test-provider-2') Provider(name='Provider 3', slug='provider-3'),
self.provider3 = Provider.objects.create(name='Test Provider 3', slug='test-provider-3') )
Provider.objects.bulk_create(providers)
def test_get_provider(self):
url = reverse('circuits-api:provider-detail', kwargs={'pk': self.provider1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.provider1.name)
def test_get_provider_graphs(self): def test_get_provider_graphs(self):
"""
Test retrieval of Graphs assigned to Providers.
"""
provider = self.model.objects.first()
ct = ContentType.objects.get(app_label='circuits', model='provider')
graphs = (
Graph(type=ct, name='Graph 1', source='http://example.com/graphs.py?provider={{ obj.slug }}&foo=1'),
Graph(type=ct, name='Graph 2', source='http://example.com/graphs.py?provider={{ obj.slug }}&foo=2'),
Graph(type=ct, name='Graph 3', source='http://example.com/graphs.py?provider={{ obj.slug }}&foo=3'),
)
Graph.objects.bulk_create(graphs)
provider_ct = ContentType.objects.get(app_label='circuits', model='provider') url = reverse('circuits-api:provider-graphs', kwargs={'pk': provider.pk})
self.graph1 = Graph.objects.create(
type=provider_ct,
name='Test Graph 1',
source='http://example.com/graphs.py?provider={{ obj.slug }}&foo=1'
)
self.graph2 = Graph.objects.create(
type=provider_ct,
name='Test Graph 2',
source='http://example.com/graphs.py?provider={{ obj.slug }}&foo=2'
)
self.graph3 = Graph.objects.create(
type=provider_ct,
name='Test Graph 3',
source='http://example.com/graphs.py?provider={{ obj.slug }}&foo=3'
)
url = reverse('circuits-api:provider-graphs', kwargs={'pk': self.provider1.pk})
response = self.client.get(url, **self.header) response = self.client.get(url, **self.header)
self.assertEqual(len(response.data), 3) self.assertEqual(len(response.data), 3)
self.assertEqual(response.data[0]['embed_url'], 'http://example.com/graphs.py?provider=test-provider-1&foo=1') self.assertEqual(response.data[0]['embed_url'], 'http://example.com/graphs.py?provider=provider-1&foo=1')
def test_list_providers(self):
url = reverse('circuits-api:provider-list') class CircuitTypeTest(APIViewTestCases.APIViewTestCase):
response = self.client.get(url, **self.header) model = CircuitType
brief_fields = ['circuit_count', 'id', 'name', 'slug', 'url']
create_data = (
{
'name': 'Circuit Type 4',
'slug': 'circuit-type-4',
},
{
'name': 'Circuit Type 5',
'slug': 'circuit-type-5',
},
{
'name': 'Circuit Type 6',
'slug': 'circuit-type-6',
},
)
self.assertEqual(response.data['count'], 3) @classmethod
def setUpTestData(cls):
def test_list_providers_brief(self): circuit_types = (
CircuitType(name='Circuit Type 1', slug='circuit-type-1'),
url = reverse('circuits-api:provider-list') CircuitType(name='Circuit Type 2', slug='circuit-type-2'),
response = self.client.get('{}?brief=1'.format(url), **self.header) CircuitType(name='Circuit Type 3', slug='circuit-type-3'),
self.assertEqual(
sorted(response.data['results'][0]),
['circuit_count', 'id', 'name', 'slug', 'url']
) )
CircuitType.objects.bulk_create(circuit_types)
def test_create_provider(self):
data = { class CircuitTest(APIViewTestCases.APIViewTestCase):
'name': 'Test Provider 4', model = Circuit
'slug': 'test-provider-4', brief_fields = ['cid', 'id', 'url']
}
url = reverse('circuits-api:provider-list') @classmethod
response = self.client.post(url, data, format='json', **self.header) def setUpTestData(cls):
self.assertHttpStatus(response, status.HTTP_201_CREATED) providers = (
self.assertEqual(Provider.objects.count(), 4) Provider(name='Provider 1', slug='provider-1'),
provider4 = Provider.objects.get(pk=response.data['id']) Provider(name='Provider 2', slug='provider-2'),
self.assertEqual(provider4.name, data['name']) )
self.assertEqual(provider4.slug, data['slug']) Provider.objects.bulk_create(providers)
def test_create_provider_bulk(self): circuit_types = (
CircuitType(name='Circuit Type 1', slug='circuit-type-1'),
CircuitType(name='Circuit Type 2', slug='circuit-type-2'),
)
CircuitType.objects.bulk_create(circuit_types)
data = [ circuits = (
Circuit(cid='Circuit 1', provider=providers[0], type=circuit_types[0]),
Circuit(cid='Circuit 2', provider=providers[0], type=circuit_types[0]),
Circuit(cid='Circuit 3', provider=providers[0], type=circuit_types[0]),
)
Circuit.objects.bulk_create(circuits)
cls.create_data = [
{ {
'name': 'Test Provider 4', 'cid': 'Circuit 4',
'slug': 'test-provider-4', 'provider': providers[1].pk,
'type': circuit_types[1].pk,
}, },
{ {
'name': 'Test Provider 5', 'cid': 'Circuit 5',
'slug': 'test-provider-5', 'provider': providers[1].pk,
'type': circuit_types[1].pk,
}, },
{ {
'name': 'Test Provider 6', 'cid': 'Circuit 6',
'slug': 'test-provider-6', 'provider': providers[1].pk,
'type': circuit_types[1].pk,
}, },
] ]
url = reverse('circuits-api:provider-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED) class CircuitTerminationTest(APIViewTestCases.APIViewTestCase):
self.assertEqual(Provider.objects.count(), 6) model = CircuitTermination
self.assertEqual(response.data[0]['name'], data[0]['name']) brief_fields = ['circuit', 'id', 'term_side', 'url']
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_provider(self): @classmethod
def setUpTestData(cls):
SIDE_A = CircuitTerminationSideChoices.SIDE_A
SIDE_Z = CircuitTerminationSideChoices.SIDE_Z
data = { sites = (
'name': 'Test Provider X', Site(name='Site 1', slug='site-1'),
'slug': 'test-provider-x', Site(name='Site 2', slug='site-2'),
}
url = reverse('circuits-api:provider-detail', kwargs={'pk': self.provider1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(Provider.objects.count(), 3)
provider1 = Provider.objects.get(pk=response.data['id'])
self.assertEqual(provider1.name, data['name'])
self.assertEqual(provider1.slug, data['slug'])
def test_delete_provider(self):
url = reverse('circuits-api:provider-detail', kwargs={'pk': self.provider1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(Provider.objects.count(), 2)
class CircuitTypeTest(APITestCase):
def setUp(self):
super().setUp()
self.circuittype1 = CircuitType.objects.create(name='Test Circuit Type 1', slug='test-circuit-type-1')
self.circuittype2 = CircuitType.objects.create(name='Test Circuit Type 2', slug='test-circuit-type-2')
self.circuittype3 = CircuitType.objects.create(name='Test Circuit Type 3', slug='test-circuit-type-3')
def test_get_circuittype(self):
url = reverse('circuits-api:circuittype-detail', kwargs={'pk': self.circuittype1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.circuittype1.name)
def test_list_circuittypes(self):
url = reverse('circuits-api:circuittype-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_circuittypes_brief(self):
url = reverse('circuits-api:circuittype-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['circuit_count', 'id', 'name', 'slug', 'url']
) )
Site.objects.bulk_create(sites)
def test_create_circuittype(self): provider = Provider.objects.create(name='Provider 1', slug='provider-1')
circuit_type = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1')
data = { circuits = (
'name': 'Test Circuit Type 4', Circuit(cid='Circuit 1', provider=provider, type=circuit_type),
'slug': 'test-circuit-type-4', Circuit(cid='Circuit 2', provider=provider, type=circuit_type),
} Circuit(cid='Circuit 3', provider=provider, type=circuit_type),
url = reverse('circuits-api:circuittype-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(CircuitType.objects.count(), 4)
circuittype4 = CircuitType.objects.get(pk=response.data['id'])
self.assertEqual(circuittype4.name, data['name'])
self.assertEqual(circuittype4.slug, data['slug'])
def test_update_circuittype(self):
data = {
'name': 'Test Circuit Type X',
'slug': 'test-circuit-type-x',
}
url = reverse('circuits-api:circuittype-detail', kwargs={'pk': self.circuittype1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(CircuitType.objects.count(), 3)
circuittype1 = CircuitType.objects.get(pk=response.data['id'])
self.assertEqual(circuittype1.name, data['name'])
self.assertEqual(circuittype1.slug, data['slug'])
def test_delete_circuittype(self):
url = reverse('circuits-api:circuittype-detail', kwargs={'pk': self.circuittype1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(CircuitType.objects.count(), 2)
class CircuitTest(APITestCase):
def setUp(self):
super().setUp()
self.provider1 = Provider.objects.create(name='Test Provider 1', slug='test-provider-1')
self.provider2 = Provider.objects.create(name='Test Provider 2', slug='test-provider-2')
self.circuittype1 = CircuitType.objects.create(name='Test Circuit Type 1', slug='test-circuit-type-1')
self.circuittype2 = CircuitType.objects.create(name='Test Circuit Type 2', slug='test-circuit-type-2')
self.circuit1 = Circuit.objects.create(cid='TEST0001', provider=self.provider1, type=self.circuittype1)
self.circuit2 = Circuit.objects.create(cid='TEST0002', provider=self.provider1, type=self.circuittype1)
self.circuit3 = Circuit.objects.create(cid='TEST0003', provider=self.provider1, type=self.circuittype1)
def test_get_circuit(self):
url = reverse('circuits-api:circuit-detail', kwargs={'pk': self.circuit1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['cid'], self.circuit1.cid)
def test_list_circuits(self):
url = reverse('circuits-api:circuit-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_circuits_brief(self):
url = reverse('circuits-api:circuit-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['cid', 'id', 'url']
) )
Circuit.objects.bulk_create(circuits)
def test_create_circuit(self): circuit_terminations = (
CircuitTermination(circuit=circuits[0], site=sites[0], port_speed=100000, term_side=SIDE_A),
CircuitTermination(circuit=circuits[0], site=sites[1], port_speed=100000, term_side=SIDE_Z),
CircuitTermination(circuit=circuits[1], site=sites[0], port_speed=100000, term_side=SIDE_A),
CircuitTermination(circuit=circuits[1], site=sites[1], port_speed=100000, term_side=SIDE_Z),
)
CircuitTermination.objects.bulk_create(circuit_terminations)
data = { cls.create_data = [
'cid': 'TEST0004',
'provider': self.provider1.pk,
'type': self.circuittype1.pk,
'status': CircuitStatusChoices.STATUS_ACTIVE,
}
url = reverse('circuits-api:circuit-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Circuit.objects.count(), 4)
circuit4 = Circuit.objects.get(pk=response.data['id'])
self.assertEqual(circuit4.cid, data['cid'])
self.assertEqual(circuit4.provider_id, data['provider'])
self.assertEqual(circuit4.type_id, data['type'])
def test_create_circuit_bulk(self):
data = [
{ {
'cid': 'TEST0004', 'circuit': circuits[2].pk,
'provider': self.provider1.pk, 'term_side': SIDE_A,
'type': self.circuittype1.pk, 'site': sites[1].pk,
'status': CircuitStatusChoices.STATUS_ACTIVE, 'port_speed': 200000,
}, },
{ {
'cid': 'TEST0005', 'circuit': circuits[2].pk,
'provider': self.provider1.pk, 'term_side': SIDE_Z,
'type': self.circuittype1.pk, 'site': sites[1].pk,
'status': CircuitStatusChoices.STATUS_ACTIVE, 'port_speed': 200000,
},
{
'cid': 'TEST0006',
'provider': self.provider1.pk,
'type': self.circuittype1.pk,
'status': CircuitStatusChoices.STATUS_ACTIVE,
}, },
] ]
url = reverse('circuits-api:circuit-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Circuit.objects.count(), 6)
self.assertEqual(response.data[0]['cid'], data[0]['cid'])
self.assertEqual(response.data[1]['cid'], data[1]['cid'])
self.assertEqual(response.data[2]['cid'], data[2]['cid'])
def test_update_circuit(self):
data = {
'cid': 'TEST000X',
'provider': self.provider2.pk,
'type': self.circuittype2.pk,
}
url = reverse('circuits-api:circuit-detail', kwargs={'pk': self.circuit1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(Circuit.objects.count(), 3)
circuit1 = Circuit.objects.get(pk=response.data['id'])
self.assertEqual(circuit1.cid, data['cid'])
self.assertEqual(circuit1.provider_id, data['provider'])
self.assertEqual(circuit1.type_id, data['type'])
def test_delete_circuit(self):
url = reverse('circuits-api:circuit-detail', kwargs={'pk': self.circuit1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(Circuit.objects.count(), 2)
class CircuitTerminationTest(APITestCase):
def setUp(self):
super().setUp()
self.site1 = Site.objects.create(name='Test Site 1', slug='test-site-1')
self.site2 = Site.objects.create(name='Test Site 2', slug='test-site-2')
provider = Provider.objects.create(name='Test Provider', slug='test-provider')
circuittype = CircuitType.objects.create(name='Test Circuit Type', slug='test-circuit-type')
self.circuit1 = Circuit.objects.create(cid='TEST0001', provider=provider, type=circuittype)
self.circuit2 = Circuit.objects.create(cid='TEST0002', provider=provider, type=circuittype)
self.circuit3 = Circuit.objects.create(cid='TEST0003', provider=provider, type=circuittype)
self.circuittermination1 = CircuitTermination.objects.create(
circuit=self.circuit1,
term_side=CircuitTerminationSideChoices.SIDE_A,
site=self.site1,
port_speed=1000000
)
self.circuittermination2 = CircuitTermination.objects.create(
circuit=self.circuit1,
term_side=CircuitTerminationSideChoices.SIDE_Z,
site=self.site2,
port_speed=1000000
)
self.circuittermination3 = CircuitTermination.objects.create(
circuit=self.circuit2,
term_side=CircuitTerminationSideChoices.SIDE_A,
site=self.site1,
port_speed=1000000
)
self.circuittermination4 = CircuitTermination.objects.create(
circuit=self.circuit2,
term_side=CircuitTerminationSideChoices.SIDE_Z,
site=self.site2,
port_speed=1000000
)
def test_get_circuittermination(self):
url = reverse('circuits-api:circuittermination-detail', kwargs={'pk': self.circuittermination1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['id'], self.circuittermination1.pk)
def test_list_circuitterminations(self):
url = reverse('circuits-api:circuittermination-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 4)
def test_create_circuittermination(self):
data = {
'circuit': self.circuit3.pk,
'term_side': CircuitTerminationSideChoices.SIDE_A,
'site': self.site1.pk,
'port_speed': 1000000,
}
url = reverse('circuits-api:circuittermination-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(CircuitTermination.objects.count(), 5)
circuittermination4 = CircuitTermination.objects.get(pk=response.data['id'])
self.assertEqual(circuittermination4.circuit_id, data['circuit'])
self.assertEqual(circuittermination4.term_side, data['term_side'])
self.assertEqual(circuittermination4.site_id, data['site'])
self.assertEqual(circuittermination4.port_speed, data['port_speed'])
def test_update_circuittermination(self):
circuittermination5 = CircuitTermination.objects.create(
circuit=self.circuit3,
term_side=CircuitTerminationSideChoices.SIDE_A,
site=self.site1,
port_speed=1000000
)
data = {
'circuit': self.circuit3.pk,
'term_side': CircuitTerminationSideChoices.SIDE_Z,
'site': self.site2.pk,
'port_speed': 1000000,
}
url = reverse('circuits-api:circuittermination-detail', kwargs={'pk': circuittermination5.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(CircuitTermination.objects.count(), 5)
circuittermination1 = CircuitTermination.objects.get(pk=response.data['id'])
self.assertEqual(circuittermination1.term_side, data['term_side'])
self.assertEqual(circuittermination1.site_id, data['site'])
self.assertEqual(circuittermination1.port_speed, data['port_speed'])
def test_delete_circuittermination(self):
url = reverse('circuits-api:circuittermination-detail', kwargs={'pk': self.circuittermination1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(CircuitTermination.objects.count(), 3)

124
netbox/dcim/api/nested_serializers.py
View File

@ -1,32 +1,35 @@
from rest_framework import serializers from rest_framework import serializers
from dcim.constants import CONNECTION_STATUS_CHOICES from dcim.constants import CONNECTION_STATUS_CHOICES
from dcim.models import ( from dcim import models
Cable, ConsolePort, ConsoleServerPort, Device, DeviceBay, DeviceType, DeviceRole, FrontPort, FrontPortTemplate,
Interface, Manufacturer, Platform, PowerFeed, PowerOutlet, PowerPanel, PowerPort, PowerPortTemplate, Rack,
RackGroup, RackRole, RearPort, RearPortTemplate, Region, Site, VirtualChassis,
)
from utilities.api import ChoiceField, WritableNestedSerializer from utilities.api import ChoiceField, WritableNestedSerializer
__all__ = [ __all__ = [
'NestedCableSerializer', 'NestedCableSerializer',
'NestedConsolePortSerializer', 'NestedConsolePortSerializer',
'NestedConsolePortTemplateSerializer',
'NestedConsoleServerPortSerializer', 'NestedConsoleServerPortSerializer',
'NestedConsoleServerPortTemplateSerializer',
'NestedDeviceBaySerializer', 'NestedDeviceBaySerializer',
'NestedDeviceBayTemplateSerializer',
'NestedDeviceRoleSerializer', 'NestedDeviceRoleSerializer',
'NestedDeviceSerializer', 'NestedDeviceSerializer',
'NestedDeviceTypeSerializer', 'NestedDeviceTypeSerializer',
'NestedFrontPortSerializer', 'NestedFrontPortSerializer',
'NestedFrontPortTemplateSerializer', 'NestedFrontPortTemplateSerializer',
'NestedInterfaceSerializer', 'NestedInterfaceSerializer',
'NestedInterfaceTemplateSerializer',
'NestedInventoryItemSerializer',
'NestedManufacturerSerializer', 'NestedManufacturerSerializer',
'NestedPlatformSerializer', 'NestedPlatformSerializer',
'NestedPowerFeedSerializer', 'NestedPowerFeedSerializer',
'NestedPowerOutletSerializer', 'NestedPowerOutletSerializer',
'NestedPowerOutletTemplateSerializer',
'NestedPowerPanelSerializer', 'NestedPowerPanelSerializer',
'NestedPowerPortSerializer', 'NestedPowerPortSerializer',
'NestedPowerPortTemplateSerializer', 'NestedPowerPortTemplateSerializer',
'NestedRackGroupSerializer', 'NestedRackGroupSerializer',
'NestedRackReservationSerializer',
'NestedRackRoleSerializer', 'NestedRackRoleSerializer',
'NestedRackSerializer', 'NestedRackSerializer',
'NestedRearPortSerializer', 'NestedRearPortSerializer',
@ -46,7 +49,7 @@ class NestedRegionSerializer(WritableNestedSerializer):
site_count = serializers.IntegerField(read_only=True) site_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = Region model = models.Region
fields = ['id', 'url', 'name', 'slug', 'site_count'] fields = ['id', 'url', 'name', 'slug', 'site_count']
@ -54,7 +57,7 @@ class NestedSiteSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:site-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:site-detail')
class Meta: class Meta:
model = Site model = models.Site
fields = ['id', 'url', 'name', 'slug'] fields = ['id', 'url', 'name', 'slug']
@ -67,7 +70,7 @@ class NestedRackGroupSerializer(WritableNestedSerializer):
rack_count = serializers.IntegerField(read_only=True) rack_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = RackGroup model = models.RackGroup
fields = ['id', 'url', 'name', 'slug', 'rack_count'] fields = ['id', 'url', 'name', 'slug', 'rack_count']
@ -76,7 +79,7 @@ class NestedRackRoleSerializer(WritableNestedSerializer):
rack_count = serializers.IntegerField(read_only=True) rack_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = RackRole model = models.RackRole
fields = ['id', 'url', 'name', 'slug', 'rack_count'] fields = ['id', 'url', 'name', 'slug', 'rack_count']
@ -85,10 +88,22 @@ class NestedRackSerializer(WritableNestedSerializer):
device_count = serializers.IntegerField(read_only=True) device_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = Rack model = models.Rack
fields = ['id', 'url', 'name', 'display_name', 'device_count'] fields = ['id', 'url', 'name', 'display_name', 'device_count']
class NestedRackReservationSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:rackreservation-detail')
user = serializers.SerializerMethodField(read_only=True)
class Meta:
model = models.RackReservation
fields = ['id', 'url', 'user', 'units']
def get_user(self, obj):
return obj.user.username
# #
# Device types # Device types
# #
@ -98,7 +113,7 @@ class NestedManufacturerSerializer(WritableNestedSerializer):
devicetype_count = serializers.IntegerField(read_only=True) devicetype_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = Manufacturer model = models.Manufacturer
fields = ['id', 'url', 'name', 'slug', 'devicetype_count'] fields = ['id', 'url', 'name', 'slug', 'devicetype_count']
@ -108,15 +123,47 @@ class NestedDeviceTypeSerializer(WritableNestedSerializer):
device_count = serializers.IntegerField(read_only=True) device_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = DeviceType model = models.DeviceType
fields = ['id', 'url', 'manufacturer', 'model', 'slug', 'display_name', 'device_count'] fields = ['id', 'url', 'manufacturer', 'model', 'slug', 'display_name', 'device_count']
class NestedConsolePortTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:consoleporttemplate-detail')
class Meta:
model = models.ConsolePortTemplate
fields = ['id', 'url', 'name']
class NestedConsoleServerPortTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:consoleserverporttemplate-detail')
class Meta:
model = models.ConsoleServerPortTemplate
fields = ['id', 'url', 'name']
class NestedPowerPortTemplateSerializer(WritableNestedSerializer): class NestedPowerPortTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:powerporttemplate-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:powerporttemplate-detail')
class Meta: class Meta:
model = PowerPortTemplate model = models.PowerPortTemplate
fields = ['id', 'url', 'name']
class NestedPowerOutletTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:poweroutlettemplate-detail')
class Meta:
model = models.PowerOutletTemplate
fields = ['id', 'url', 'name']
class NestedInterfaceTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:interfacetemplate-detail')
class Meta:
model = models.InterfaceTemplate
fields = ['id', 'url', 'name'] fields = ['id', 'url', 'name']
@ -124,7 +171,7 @@ class NestedRearPortTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:rearporttemplate-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:rearporttemplate-detail')
class Meta: class Meta:
model = RearPortTemplate model = models.RearPortTemplate
fields = ['id', 'url', 'name'] fields = ['id', 'url', 'name']
@ -132,7 +179,15 @@ class NestedFrontPortTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:frontporttemplate-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:frontporttemplate-detail')
class Meta: class Meta:
model = FrontPortTemplate model = models.FrontPortTemplate
fields = ['id', 'url', 'name']
class NestedDeviceBayTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:devicebaytemplate-detail')
class Meta:
model = models.DeviceBayTemplate
fields = ['id', 'url', 'name'] fields = ['id', 'url', 'name']
@ -146,7 +201,7 @@ class NestedDeviceRoleSerializer(WritableNestedSerializer):
virtualmachine_count = serializers.IntegerField(read_only=True) virtualmachine_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = DeviceRole model = models.DeviceRole
fields = ['id', 'url', 'name', 'slug', 'device_count', 'virtualmachine_count'] fields = ['id', 'url', 'name', 'slug', 'device_count', 'virtualmachine_count']
@ -156,7 +211,7 @@ class NestedPlatformSerializer(WritableNestedSerializer):
virtualmachine_count = serializers.IntegerField(read_only=True) virtualmachine_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = Platform model = models.Platform
fields = ['id', 'url', 'name', 'slug', 'device_count', 'virtualmachine_count'] fields = ['id', 'url', 'name', 'slug', 'device_count', 'virtualmachine_count']
@ -164,7 +219,7 @@ class NestedDeviceSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:device-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:device-detail')
class Meta: class Meta:
model = Device model = models.Device
fields = ['id', 'url', 'name', 'display_name'] fields = ['id', 'url', 'name', 'display_name']
@ -174,7 +229,7 @@ class NestedConsoleServerPortSerializer(WritableNestedSerializer):
connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True) connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True)
class Meta: class Meta:
model = ConsoleServerPort model = models.ConsoleServerPort
fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status'] fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status']
@ -184,7 +239,7 @@ class NestedConsolePortSerializer(WritableNestedSerializer):
connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True) connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True)
class Meta: class Meta:
model = ConsolePort model = models.ConsolePort
fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status'] fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status']
@ -194,7 +249,7 @@ class NestedPowerOutletSerializer(WritableNestedSerializer):
connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True) connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True)
class Meta: class Meta:
model = PowerOutlet model = models.PowerOutlet
fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status'] fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status']
@ -204,7 +259,7 @@ class NestedPowerPortSerializer(WritableNestedSerializer):
connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True) connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True)
class Meta: class Meta:
model = PowerPort model = models.PowerPort
fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status'] fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status']
@ -214,7 +269,7 @@ class NestedInterfaceSerializer(WritableNestedSerializer):
connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True) connection_status = ChoiceField(choices=CONNECTION_STATUS_CHOICES, read_only=True)
class Meta: class Meta:
model = Interface model = models.Interface
fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status'] fields = ['id', 'url', 'device', 'name', 'cable', 'connection_status']
@ -223,7 +278,7 @@ class NestedRearPortSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:rearport-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:rearport-detail')
class Meta: class Meta:
model = RearPort model = models.RearPort
fields = ['id', 'url', 'device', 'name', 'cable'] fields = ['id', 'url', 'device', 'name', 'cable']
@ -232,7 +287,7 @@ class NestedFrontPortSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:frontport-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:frontport-detail')
class Meta: class Meta:
model = FrontPort model = models.FrontPort
fields = ['id', 'url', 'device', 'name', 'cable'] fields = ['id', 'url', 'device', 'name', 'cable']
@ -241,7 +296,16 @@ class NestedDeviceBaySerializer(WritableNestedSerializer):
device = NestedDeviceSerializer(read_only=True) device = NestedDeviceSerializer(read_only=True)
class Meta: class Meta:
model = DeviceBay model = models.DeviceBay
fields = ['id', 'url', 'device', 'name']
class NestedInventoryItemSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:inventoryitem-detail')
device = NestedDeviceSerializer(read_only=True)
class Meta:
model = models.InventoryItem
fields = ['id', 'url', 'device', 'name'] fields = ['id', 'url', 'device', 'name']
@ -253,7 +317,7 @@ class NestedCableSerializer(serializers.ModelSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:cable-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:cable-detail')
class Meta: class Meta:
model = Cable model = models.Cable
fields = ['id', 'url', 'label'] fields = ['id', 'url', 'label']
@ -267,7 +331,7 @@ class NestedVirtualChassisSerializer(WritableNestedSerializer):
member_count = serializers.IntegerField(read_only=True) member_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = VirtualChassis model = models.VirtualChassis
fields = ['id', 'url', 'master', 'member_count'] fields = ['id', 'url', 'master', 'member_count']
@ -280,7 +344,7 @@ class NestedPowerPanelSerializer(WritableNestedSerializer):
powerfeed_count = serializers.IntegerField(read_only=True) powerfeed_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = PowerPanel model = models.PowerPanel
fields = ['id', 'url', 'name', 'powerfeed_count'] fields = ['id', 'url', 'name', 'powerfeed_count']
@ -288,5 +352,5 @@ class NestedPowerFeedSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:powerfeed-detail') url = serializers.HyperlinkedIdentityField(view_name='dcim-api:powerfeed-detail')
class Meta: class Meta:
model = PowerFeed model = models.PowerFeed
fields = ['id', 'url', 'name'] fields = ['id', 'url', 'name']

8
netbox/dcim/models/__init__.py
View File

@ -2140,9 +2140,9 @@ class Cable(ChangeLoggedModel):
""" """
instance = super().from_db(db, field_names, values) instance = super().from_db(db, field_names, values)
instance._orig_termination_a_type = instance.termination_a_type instance._orig_termination_a_type_id = instance.termination_a_type_id
instance._orig_termination_a_id = instance.termination_a_id instance._orig_termination_a_id = instance.termination_a_id
instance._orig_termination_b_type = instance.termination_b_type instance._orig_termination_b_type_id = instance.termination_b_type_id
instance._orig_termination_b_id = instance.termination_b_id instance._orig_termination_b_id = instance.termination_b_id
return instance return instance
@ -2179,14 +2179,14 @@ class Cable(ChangeLoggedModel):
if self.pk: if self.pk:
err_msg = 'Cable termination points may not be modified. Delete and recreate the cable instead.' err_msg = 'Cable termination points may not be modified. Delete and recreate the cable instead.'
if ( if (
self.termination_a_type != self._orig_termination_a_type or self.termination_a_type_id != self._orig_termination_a_type_id or
self.termination_a_id != self._orig_termination_a_id self.termination_a_id != self._orig_termination_a_id
): ):
raise ValidationError({ raise ValidationError({
'termination_a': err_msg 'termination_a': err_msg
}) })
if ( if (
self.termination_b_type != self._orig_termination_b_type or self.termination_b_type_id != self._orig_termination_b_type_id or
self.termination_b_id != self._orig_termination_b_id self.termination_b_id != self._orig_termination_b_id
): ):
raise ValidationError({ raise ValidationError({

4453
netbox/dcim/tests/test_api.py
View File

File diff suppressed because it is too large Load Diff

56
netbox/extras/admin.py
View File

@ -46,24 +46,19 @@ class WebhookAdmin(admin.ModelAdmin):
form = WebhookForm form = WebhookForm
fieldsets = ( fieldsets = (
(None, { (None, {
'fields': ( 'fields': ('name', 'obj_type', 'enabled')
'name', 'obj_type', 'enabled',
)
}), }),
('Events', { ('Events', {
'fields': ( 'fields': ('type_create', 'type_update', 'type_delete')
'type_create', 'type_update', 'type_delete',
)
}), }),
('HTTP Request', { ('HTTP Request', {
'fields': ( 'fields': (
'payload_url', 'http_method', 'http_content_type', 'additional_headers', 'body_template', 'secret', 'payload_url', 'http_method', 'http_content_type', 'additional_headers', 'body_template', 'secret',
) ),
'classes': ('monospace',)
}), }),
('SSL', { ('SSL', {
'fields': ( 'fields': ('ssl_verification', 'ca_file_path')
'ssl_verification', 'ca_file_path',
)
}) })
) )
@ -121,6 +116,8 @@ class CustomLinkForm(forms.ModelForm):
'url': forms.Textarea, 'url': forms.Textarea,
} }
help_texts = { help_texts = {
'weight': 'A numeric weight to influence the ordering of this link among its peers. Lower weights appear '
'first in a list.',
'text': 'Jinja2 template code for the link text. Reference the object as <code>{{ obj }}</code>. Links ' 'text': 'Jinja2 template code for the link text. Reference the object as <code>{{ obj }}</code>. Links '
'which render as empty text will not be displayed.', 'which render as empty text will not be displayed.',
'url': 'Jinja2 template code for the link URL. Reference the object as <code>{{ obj }}</code>.', 'url': 'Jinja2 template code for the link URL. Reference the object as <code>{{ obj }}</code>.',
@ -136,6 +133,15 @@ class CustomLinkForm(forms.ModelForm):
@admin.register(CustomLink) @admin.register(CustomLink)
class CustomLinkAdmin(admin.ModelAdmin): class CustomLinkAdmin(admin.ModelAdmin):
fieldsets = (
('Custom Link', {
'fields': ('content_type', 'name', 'group_name', 'weight', 'button_class', 'new_window')
}),
('Templates', {
'fields': ('text', 'url'),
'classes': ('monospace',)
})
)
list_display = [ list_display = [
'name', 'content_type', 'group_name', 'weight', 'name', 'content_type', 'group_name', 'weight',
] ]
@ -149,8 +155,29 @@ class CustomLinkAdmin(admin.ModelAdmin):
# Graphs # Graphs
# #
class GraphForm(forms.ModelForm):
class Meta:
model = Graph
exclude = ()
widgets = {
'source': forms.Textarea,
'link': forms.Textarea,
}
@admin.register(Graph) @admin.register(Graph)
class GraphAdmin(admin.ModelAdmin): class GraphAdmin(admin.ModelAdmin):
fieldsets = (
('Graph', {
'fields': ('type', 'name', 'weight')
}),
('Templates', {
'fields': ('template_language', 'source', 'link'),
'classes': ('monospace',)
})
)
form = GraphForm
list_display = [ list_display = [
'name', 'type', 'weight', 'template_language', 'source', 'name', 'type', 'weight', 'template_language', 'source',
] ]
@ -179,6 +206,15 @@ class ExportTemplateForm(forms.ModelForm):
@admin.register(ExportTemplate) @admin.register(ExportTemplate)
class ExportTemplateAdmin(admin.ModelAdmin): class ExportTemplateAdmin(admin.ModelAdmin):
fieldsets = (
('Export Template', {
'fields': ('content_type', 'name', 'description', 'mime_type', 'file_extension')
}),
('Content', {
'fields': ('template_language', 'template_code'),
'classes': ('monospace',)
})
)
list_display = [ list_display = [
'name', 'content_type', 'description', 'mime_type', 'file_extension', 'name', 'content_type', 'description', 'mime_type', 'file_extension',
] ]

44
netbox/extras/api/nested_serializers.py
View File

@ -1,15 +1,49 @@
from rest_framework import serializers from rest_framework import serializers
from extras.models import ReportResult from extras import models
from utilities.api import WritableNestedSerializer
__all__ = [ __all__ = [
'NestedConfigContextSerializer',
'NestedExportTemplateSerializer',
'NestedGraphSerializer',
'NestedReportResultSerializer', 'NestedReportResultSerializer',
'NestedTagSerializer',
] ]
# class NestedConfigContextSerializer(WritableNestedSerializer):
# Reports url = serializers.HyperlinkedIdentityField(view_name='extras-api:configcontext-detail')
#
class Meta:
model = models.ConfigContext
fields = ['id', 'url', 'name']
class NestedExportTemplateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='extras-api:exporttemplate-detail')
class Meta:
model = models.ExportTemplate
fields = ['id', 'url', 'name']
class NestedGraphSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='extras-api:graph-detail')
class Meta:
model = models.Graph
fields = ['id', 'url', 'name']
class NestedTagSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='extras-api:tag-detail')
tagged_items = serializers.IntegerField(read_only=True)
class Meta:
model = models.Tag
fields = ['id', 'url', 'name', 'slug', 'color', 'tagged_items']
class NestedReportResultSerializer(serializers.ModelSerializer): class NestedReportResultSerializer(serializers.ModelSerializer):
url = serializers.HyperlinkedIdentityField( url = serializers.HyperlinkedIdentityField(
@ -19,5 +53,5 @@ class NestedReportResultSerializer(serializers.ModelSerializer):
) )
class Meta: class Meta:
model = ReportResult model = models.ReportResult
fields = ['url', 'created', 'user', 'failed'] fields = ['url', 'created', 'user', 'failed']

604
netbox/extras/tests/test_api.py
View File

@ -5,13 +5,11 @@ from django.urls import reverse
from django.utils import timezone from django.utils import timezone
from rest_framework import status from rest_framework import status
from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Platform, Rack, RackGroup, RackRole, Region, Site from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Rack, RackGroup, RackRole, Site
from extras.api.views import ScriptViewSet from extras.api.views import ScriptViewSet
from extras.models import ConfigContext, Graph, ExportTemplate, Tag from extras.models import ConfigContext, Graph, ExportTemplate, Tag
from extras.scripts import BooleanVar, IntegerVar, Script, StringVar from extras.scripts import BooleanVar, IntegerVar, Script, StringVar
from extras.utils import FeatureQuery from utilities.testing import APITestCase, APIViewTestCases
from tenancy.models import Tenant, TenantGroup
from utilities.testing import APITestCase
class AppTest(APITestCase): class AppTest(APITestCase):
@ -24,489 +22,150 @@ class AppTest(APITestCase):
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
class GraphTest(APITestCase): class GraphTest(APIViewTestCases.APIViewTestCase):
model = Graph
def setUp(self): brief_fields = ['id', 'name', 'url']
create_data = [
super().setUp() {
site_ct = ContentType.objects.get_for_model(Site)
self.graph1 = Graph.objects.create(
type=site_ct,
name='Test Graph 1',
source='http://example.com/graphs.py?site={{ obj.name }}&foo=1'
)
self.graph2 = Graph.objects.create(
type=site_ct,
name='Test Graph 2',
source='http://example.com/graphs.py?site={{ obj.name }}&foo=2'
)
self.graph3 = Graph.objects.create(
type=site_ct,
name='Test Graph 3',
source='http://example.com/graphs.py?site={{ obj.name }}&foo=3'
)
def test_get_graph(self):
url = reverse('extras-api:graph-detail', kwargs={'pk': self.graph1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.graph1.name)
def test_list_graphs(self):
url = reverse('extras-api:graph-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_create_graph(self):
data = {
'type': 'dcim.site', 'type': 'dcim.site',
'name': 'Test Graph 4', 'name': 'Graph 4',
'source': 'http://example.com/graphs.py?site={{ obj.name }}&foo=4', 'source': 'http://example.com/graphs.py?site={{ obj.name }}&foo=4',
} },
{
url = reverse('extras-api:graph-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Graph.objects.count(), 4)
graph4 = Graph.objects.get(pk=response.data['id'])
self.assertEqual(graph4.type, ContentType.objects.get_for_model(Site))
self.assertEqual(graph4.name, data['name'])
self.assertEqual(graph4.source, data['source'])
def test_create_graph_bulk(self):
data = [
{
'type': 'dcim.site',
'name': 'Test Graph 4',
'source': 'http://example.com/graphs.py?site={{ obj.name }}&foo=4',
},
{
'type': 'dcim.site',
'name': 'Test Graph 5',
'source': 'http://example.com/graphs.py?site={{ obj.name }}&foo=5',
},
{
'type': 'dcim.site',
'name': 'Test Graph 6',
'source': 'http://example.com/graphs.py?site={{ obj.name }}&foo=6',
},
]
url = reverse('extras-api:graph-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Graph.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_graph(self):
data = {
'type': 'dcim.site', 'type': 'dcim.site',
'name': 'Test Graph X', 'name': 'Graph 5',
'source': 'http://example.com/graphs.py?site={{ obj.name }}&foo=99', 'source': 'http://example.com/graphs.py?site={{ obj.name }}&foo=5',
} },
{
'type': 'dcim.site',
'name': 'Graph 6',
'source': 'http://example.com/graphs.py?site={{ obj.name }}&foo=6',
},
]
url = reverse('extras-api:graph-detail', kwargs={'pk': self.graph1.pk}) @classmethod
response = self.client.put(url, data, format='json', **self.header) def setUpTestData(cls):
ct = ContentType.objects.get_for_model(Site)
self.assertHttpStatus(response, status.HTTP_200_OK) graphs = (
self.assertEqual(Graph.objects.count(), 3) Graph(type=ct, name='Graph 1', source='http://example.com/graphs.py?site={{ obj.name }}&foo=1'),
graph1 = Graph.objects.get(pk=response.data['id']) Graph(type=ct, name='Graph 2', source='http://example.com/graphs.py?site={{ obj.name }}&foo=2'),
self.assertEqual(graph1.type, ContentType.objects.get_for_model(Site)) Graph(type=ct, name='Graph 3', source='http://example.com/graphs.py?site={{ obj.name }}&foo=3'),
self.assertEqual(graph1.name, data['name'])
self.assertEqual(graph1.source, data['source'])
def test_delete_graph(self):
url = reverse('extras-api:graph-detail', kwargs={'pk': self.graph1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(Graph.objects.count(), 2)
class ExportTemplateTest(APITestCase):
def setUp(self):
super().setUp()
content_type = ContentType.objects.get_for_model(Device)
self.exporttemplate1 = ExportTemplate.objects.create(
content_type=content_type, name='Test Export Template 1',
template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}'
)
self.exporttemplate2 = ExportTemplate.objects.create(
content_type=content_type, name='Test Export Template 2',
template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}'
)
self.exporttemplate3 = ExportTemplate.objects.create(
content_type=content_type, name='Test Export Template 3',
template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}'
) )
Graph.objects.bulk_create(graphs)
def test_get_exporttemplate(self):
url = reverse('extras-api:exporttemplate-detail', kwargs={'pk': self.exporttemplate1.pk}) class ExportTemplateTest(APIViewTestCases.APIViewTestCase):
response = self.client.get(url, **self.header) model = ExportTemplate
brief_fields = ['id', 'name', 'url']
self.assertEqual(response.data['name'], self.exporttemplate1.name) create_data = [
{
def test_list_exporttemplates(self):
url = reverse('extras-api:exporttemplate-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_create_exporttemplate(self):
data = {
'content_type': 'dcim.device', 'content_type': 'dcim.device',
'name': 'Test Export Template 4', 'name': 'Test Export Template 4',
'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}', 'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
} },
{
url = reverse('extras-api:exporttemplate-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(ExportTemplate.objects.count(), 4)
exporttemplate4 = ExportTemplate.objects.get(pk=response.data['id'])
self.assertEqual(exporttemplate4.content_type, ContentType.objects.get_for_model(Device))
self.assertEqual(exporttemplate4.name, data['name'])
self.assertEqual(exporttemplate4.template_code, data['template_code'])
def test_create_exporttemplate_bulk(self):
data = [
{
'content_type': 'dcim.device',
'name': 'Test Export Template 4',
'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
},
{
'content_type': 'dcim.device',
'name': 'Test Export Template 5',
'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
},
{
'content_type': 'dcim.device',
'name': 'Test Export Template 6',
'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
},
]
url = reverse('extras-api:exporttemplate-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(ExportTemplate.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_exporttemplate(self):
data = {
'content_type': 'dcim.device', 'content_type': 'dcim.device',
'name': 'Test Export Template X', 'name': 'Test Export Template 5',
'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}', 'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
} },
{
'content_type': 'dcim.device',
'name': 'Test Export Template 6',
'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
},
]
url = reverse('extras-api:exporttemplate-detail', kwargs={'pk': self.exporttemplate1.pk}) @classmethod
response = self.client.put(url, data, format='json', **self.header) def setUpTestData(cls):
ct = ContentType.objects.get_for_model(Device)
self.assertHttpStatus(response, status.HTTP_200_OK) export_templates = (
self.assertEqual(ExportTemplate.objects.count(), 3) ExportTemplate(
exporttemplate1 = ExportTemplate.objects.get(pk=response.data['id']) content_type=ct,
self.assertEqual(exporttemplate1.name, data['name']) name='Export Template 1',
self.assertEqual(exporttemplate1.template_code, data['template_code']) template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}'
),
def test_delete_exporttemplate(self): ExportTemplate(
content_type=ct,
url = reverse('extras-api:exporttemplate-detail', kwargs={'pk': self.exporttemplate1.pk}) name='Export Template 2',
response = self.client.delete(url, **self.header) template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}'
),
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT) ExportTemplate(
self.assertEqual(ExportTemplate.objects.count(), 2) content_type=ct,
name='Export Template 3',
template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}'
class TagTest(APITestCase): ),
def setUp(self):
super().setUp()
self.tag1 = Tag.objects.create(name='Test Tag 1', slug='test-tag-1')
self.tag2 = Tag.objects.create(name='Test Tag 2', slug='test-tag-2')
self.tag3 = Tag.objects.create(name='Test Tag 3', slug='test-tag-3')
def test_get_tag(self):
url = reverse('extras-api:tag-detail', kwargs={'pk': self.tag1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.tag1.name)
def test_list_tags(self):
url = reverse('extras-api:tag-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_create_tag(self):
data = {
'name': 'Test Tag 4',
'slug': 'test-tag-4',
}
url = reverse('extras-api:tag-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Tag.objects.count(), 4)
tag4 = Tag.objects.get(pk=response.data['id'])
self.assertEqual(tag4.name, data['name'])
self.assertEqual(tag4.slug, data['slug'])
def test_create_tag_bulk(self):
data = [
{
'name': 'Test Tag 4',
'slug': 'test-tag-4',
},
{
'name': 'Test Tag 5',
'slug': 'test-tag-5',
},
{
'name': 'Test Tag 6',
'slug': 'test-tag-6',
},
]
url = reverse('extras-api:tag-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Tag.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_tag(self):
data = {
'name': 'Test Tag X',
'slug': 'test-tag-x',
}
url = reverse('extras-api:tag-detail', kwargs={'pk': self.tag1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(Tag.objects.count(), 3)
tag1 = Tag.objects.get(pk=response.data['id'])
self.assertEqual(tag1.name, data['name'])
self.assertEqual(tag1.slug, data['slug'])
def test_delete_tag(self):
url = reverse('extras-api:tag-detail', kwargs={'pk': self.tag1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(Tag.objects.count(), 2)
class ConfigContextTest(APITestCase):
def setUp(self):
super().setUp()
self.configcontext1 = ConfigContext.objects.create(
name='Test Config Context 1',
weight=100,
data={'foo': 123}
) )
self.configcontext2 = ConfigContext.objects.create( ExportTemplate.objects.bulk_create(export_templates)
name='Test Config Context 2',
weight=200,
data={'bar': 456} class TagTest(APIViewTestCases.APIViewTestCase):
model = Tag
brief_fields = ['color', 'id', 'name', 'slug', 'tagged_items', 'url']
create_data = [
{
'name': 'Tag 4',
'slug': 'tag-4',
},
{
'name': 'Tag 5',
'slug': 'tag-5',
},
{
'name': 'Tag 6',
'slug': 'tag-6',
},
]
@classmethod
def setUpTestData(cls):
tags = (
Tag(name='Tag 1', slug='tag-1'),
Tag(name='Tag 2', slug='tag-2'),
Tag(name='Tag 3', slug='tag-3'),
) )
self.configcontext3 = ConfigContext.objects.create( Tag.objects.bulk_create(tags)
name='Test Config Context 3',
weight=300,
data={'baz': 789} class ConfigContextTest(APIViewTestCases.APIViewTestCase):
model = ConfigContext
brief_fields = ['id', 'name', 'url']
create_data = [
{
'name': 'Config Context 4',
'data': {'more_foo': True},
},
{
'name': 'Config Context 5',
'data': {'more_bar': False},
},
{
'name': 'Config Context 6',
'data': {'more_baz': None},
},
]
@classmethod
def setUpTestData(cls):
config_contexts = (
ConfigContext(name='Config Context 1', weight=100, data={'foo': 123}),
ConfigContext(name='Config Context 2', weight=200, data={'bar': 456}),
ConfigContext(name='Config Context 3', weight=300, data={'baz': 789}),
) )
ConfigContext.objects.bulk_create(config_contexts)
def test_get_configcontext(self):
url = reverse('extras-api:configcontext-detail', kwargs={'pk': self.configcontext1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.configcontext1.name)
self.assertEqual(response.data['data'], self.configcontext1.data)
def test_list_configcontexts(self):
url = reverse('extras-api:configcontext-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_create_configcontext(self):
region1 = Region.objects.create(name='Test Region 1', slug='test-region-1')
region2 = Region.objects.create(name='Test Region 2', slug='test-region-2')
site1 = Site.objects.create(name='Test Site 1', slug='test-site-1')
site2 = Site.objects.create(name='Test Site 2', slug='test-site-2')
role1 = DeviceRole.objects.create(name='Test Role 1', slug='test-role-1')
role2 = DeviceRole.objects.create(name='Test Role 2', slug='test-role-2')
platform1 = Platform.objects.create(name='Test Platform 1', slug='test-platform-1')
platform2 = Platform.objects.create(name='Test Platform 2', slug='test-platform-2')
tenantgroup1 = TenantGroup(name='Test Tenant Group 1', slug='test-tenant-group-1')
tenantgroup1.save()
tenantgroup2 = TenantGroup(name='Test Tenant Group 2', slug='test-tenant-group-2')
tenantgroup2.save()
tenant1 = Tenant.objects.create(name='Test Tenant 1', slug='test-tenant-1')
tenant2 = Tenant.objects.create(name='Test Tenant 2', slug='test-tenant-2')
tag1 = Tag.objects.create(name='Test Tag 1', slug='test-tag-1')
tag2 = Tag.objects.create(name='Test Tag 2', slug='test-tag-2')
data = {
'name': 'Test Config Context 4',
'weight': 1000,
'regions': [region1.pk, region2.pk],
'sites': [site1.pk, site2.pk],
'roles': [role1.pk, role2.pk],
'platforms': [platform1.pk, platform2.pk],
'tenant_groups': [tenantgroup1.pk, tenantgroup2.pk],
'tenants': [tenant1.pk, tenant2.pk],
'tags': [tag1.slug, tag2.slug],
'data': {'foo': 'XXX'}
}
url = reverse('extras-api:configcontext-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(ConfigContext.objects.count(), 4)
configcontext4 = ConfigContext.objects.get(pk=response.data['id'])
self.assertEqual(configcontext4.name, data['name'])
self.assertEqual(region1.pk, data['regions'][0])
self.assertEqual(region2.pk, data['regions'][1])
self.assertEqual(site1.pk, data['sites'][0])
self.assertEqual(site2.pk, data['sites'][1])
self.assertEqual(role1.pk, data['roles'][0])
self.assertEqual(role2.pk, data['roles'][1])
self.assertEqual(platform1.pk, data['platforms'][0])
self.assertEqual(platform2.pk, data['platforms'][1])
self.assertEqual(tenantgroup1.pk, data['tenant_groups'][0])
self.assertEqual(tenantgroup2.pk, data['tenant_groups'][1])
self.assertEqual(tenant1.pk, data['tenants'][0])
self.assertEqual(tenant2.pk, data['tenants'][1])
self.assertEqual(tag1.slug, data['tags'][0])
self.assertEqual(tag2.slug, data['tags'][1])
self.assertEqual(configcontext4.data, data['data'])
def test_create_configcontext_bulk(self):
data = [
{
'name': 'Test Config Context 4',
'data': {'more_foo': True},
},
{
'name': 'Test Config Context 5',
'data': {'more_bar': False},
},
{
'name': 'Test Config Context 6',
'data': {'more_baz': None},
},
]
url = reverse('extras-api:configcontext-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(ConfigContext.objects.count(), 6)
for i in range(0, 3):
self.assertEqual(response.data[i]['name'], data[i]['name'])
self.assertEqual(response.data[i]['data'], data[i]['data'])
def test_update_configcontext(self):
region1 = Region.objects.create(name='Test Region 1', slug='test-region-1')
region2 = Region.objects.create(name='Test Region 2', slug='test-region-2')
data = {
'name': 'Test Config Context X',
'weight': 999,
'regions': [region1.pk, region2.pk],
'data': {'foo': 'XXX'}
}
url = reverse('extras-api:configcontext-detail', kwargs={'pk': self.configcontext1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(ConfigContext.objects.count(), 3)
configcontext1 = ConfigContext.objects.get(pk=response.data['id'])
self.assertEqual(configcontext1.name, data['name'])
self.assertEqual(configcontext1.weight, data['weight'])
self.assertEqual(sorted([r.pk for r in configcontext1.regions.all()]), sorted(data['regions']))
self.assertEqual(configcontext1.data, data['data'])
def test_delete_configcontext(self):
url = reverse('extras-api:configcontext-detail', kwargs={'pk': self.configcontext1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(ConfigContext.objects.count(), 2)
def test_render_configcontext_for_object(self): def test_render_configcontext_for_object(self):
"""
# Create a Device for which we'll render a config context Test rendering config context data for a device.
manufacturer = Manufacturer.objects.create( """
name='Test Manufacturer', manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
slug='test-manufacturer' devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1')
) devicerole = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
device_type = DeviceType.objects.create( site = Site.objects.create(name='Site-1', slug='site-1')
manufacturer=manufacturer, device = Device.objects.create(name='Device 1', device_type=devicetype, device_role=devicerole, site=site)
model='Test Device Type'
)
device_role = DeviceRole.objects.create(
name='Test Role',
slug='test-role'
)
site = Site.objects.create(
name='Test Site',
slug='test-site'
)
device = Device.objects.create(
name='Test Device',
device_type=device_type,
device_role=device_role,
site=site
)
# Test default config contexts (created at test setup) # Test default config contexts (created at test setup)
rendered_context = device.get_config_context() rendered_context = device.get_config_context()
@ -516,7 +175,7 @@ class ConfigContextTest(APITestCase):
# Add another context specific to the site # Add another context specific to the site
configcontext4 = ConfigContext( configcontext4 = ConfigContext(
name='Test Config Context 4', name='Config Context 4',
data={'site_data': 'ABC'} data={'site_data': 'ABC'}
) )
configcontext4.save() configcontext4.save()
@ -526,7 +185,7 @@ class ConfigContextTest(APITestCase):
# Override one of the default contexts # Override one of the default contexts
configcontext5 = ConfigContext( configcontext5 = ConfigContext(
name='Test Config Context 5', name='Config Context 5',
weight=2000, weight=2000,
data={'foo': 999} data={'foo': 999}
) )
@ -536,12 +195,9 @@ class ConfigContextTest(APITestCase):
self.assertEqual(rendered_context['foo'], 999) self.assertEqual(rendered_context['foo'], 999)
# Add a context which does NOT match our device and ensure it does not apply # Add a context which does NOT match our device and ensure it does not apply
site2 = Site.objects.create( site2 = Site.objects.create(name='Site 2', slug='site-2')
name='Test Site 2',
slug='test-site-2'
)
configcontext6 = ConfigContext( configcontext6 = ConfigContext(
name='Test Config Context 6', name='Config Context 6',
weight=2000, weight=2000,
data={'bar': 999} data={'bar': 999}
) )

32
netbox/ipam/api/nested_serializers.py
View File

@ -1,6 +1,6 @@
from rest_framework import serializers from rest_framework import serializers
from ipam.models import Aggregate, IPAddress, Prefix, RIR, Role, VLAN, VLANGroup, VRF from ipam import models
from utilities.api import WritableNestedSerializer from utilities.api import WritableNestedSerializer
__all__ = [ __all__ = [
@ -9,6 +9,7 @@ __all__ = [
'NestedPrefixSerializer', 'NestedPrefixSerializer',
'NestedRIRSerializer', 'NestedRIRSerializer',
'NestedRoleSerializer', 'NestedRoleSerializer',
'NestedServiceSerializer',
'NestedVLANGroupSerializer', 'NestedVLANGroupSerializer',
'NestedVLANSerializer', 'NestedVLANSerializer',
'NestedVRFSerializer', 'NestedVRFSerializer',
@ -24,7 +25,7 @@ class NestedVRFSerializer(WritableNestedSerializer):
prefix_count = serializers.IntegerField(read_only=True) prefix_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = VRF model = models.VRF
fields = ['id', 'url', 'name', 'rd', 'prefix_count'] fields = ['id', 'url', 'name', 'rd', 'prefix_count']
@ -37,7 +38,7 @@ class NestedRIRSerializer(WritableNestedSerializer):
aggregate_count = serializers.IntegerField(read_only=True) aggregate_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = RIR model = models.RIR
fields = ['id', 'url', 'name', 'slug', 'aggregate_count'] fields = ['id', 'url', 'name', 'slug', 'aggregate_count']
@ -45,7 +46,7 @@ class NestedAggregateSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:aggregate-detail') url = serializers.HyperlinkedIdentityField(view_name='ipam-api:aggregate-detail')
class Meta: class Meta:
model = Aggregate model = models.Aggregate
fields = ['id', 'url', 'family', 'prefix'] fields = ['id', 'url', 'family', 'prefix']
@ -59,7 +60,7 @@ class NestedRoleSerializer(WritableNestedSerializer):
vlan_count = serializers.IntegerField(read_only=True) vlan_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = Role model = models.Role
fields = ['id', 'url', 'name', 'slug', 'prefix_count', 'vlan_count'] fields = ['id', 'url', 'name', 'slug', 'prefix_count', 'vlan_count']
@ -68,7 +69,7 @@ class NestedVLANGroupSerializer(WritableNestedSerializer):
vlan_count = serializers.IntegerField(read_only=True) vlan_count = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = VLANGroup model = models.VLANGroup
fields = ['id', 'url', 'name', 'slug', 'vlan_count'] fields = ['id', 'url', 'name', 'slug', 'vlan_count']
@ -76,7 +77,7 @@ class NestedVLANSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:vlan-detail') url = serializers.HyperlinkedIdentityField(view_name='ipam-api:vlan-detail')
class Meta: class Meta:
model = VLAN model = models.VLAN
fields = ['id', 'url', 'vid', 'name', 'display_name'] fields = ['id', 'url', 'vid', 'name', 'display_name']
@ -88,7 +89,7 @@ class NestedPrefixSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:prefix-detail') url = serializers.HyperlinkedIdentityField(view_name='ipam-api:prefix-detail')
class Meta: class Meta:
model = Prefix model = models.Prefix
fields = ['id', 'url', 'family', 'prefix'] fields = ['id', 'url', 'family', 'prefix']
@ -96,10 +97,21 @@ class NestedPrefixSerializer(WritableNestedSerializer):
# IP addresses # IP addresses
# #
class NestedIPAddressSerializer(WritableNestedSerializer): class NestedIPAddressSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:ipaddress-detail') url = serializers.HyperlinkedIdentityField(view_name='ipam-api:ipaddress-detail')
class Meta: class Meta:
model = IPAddress model = models.IPAddress
fields = ['id', 'url', 'family', 'address'] fields = ['id', 'url', 'family', 'address']
#
# Services
#
class NestedServiceSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:service-detail')
class Meta:
model = models.Service
fields = ['id', 'url', 'name', 'protocol', 'port']

2
netbox/ipam/api/views.py
View File

@ -276,7 +276,7 @@ class VLANViewSet(CustomFieldModelViewSet):
queryset = VLAN.objects.prefetch_related( queryset = VLAN.objects.prefetch_related(
'site', 'group', 'tenant', 'role', 'tags' 'site', 'group', 'tenant', 'role', 'tags'
).annotate( ).annotate(
prefix_count=get_subquery(Prefix, 'role') prefix_count=get_subquery(Prefix, 'vlan')
) )
serializer_class = serializers.VLANSerializer serializer_class = serializers.VLANSerializer
filterset_class = filters.VLANFilterSet filterset_class = filters.VLANFilterSet

1116
netbox/ipam/tests/test_api.py
View File

File diff suppressed because it is too large Load Diff

10
netbox/secrets/forms.py
View File

@ -115,6 +115,16 @@ class SecretForm(BootstrapMixin, CustomFieldModelForm):
'plaintext2': "The two given plaintext values do not match. Please check your input." 'plaintext2': "The two given plaintext values do not match. Please check your input."
}) })
# Validate uniqueness
if Secret.objects.filter(
device=self.cleaned_data['device'],
role=self.cleaned_data['role'],
name=self.cleaned_data['name']
).exists():
raise forms.ValidationError(
"Each secret assigned to a device must have a unique combination of role and name"
)
class SecretCSVForm(CustomFieldModelCSVForm): class SecretCSVForm(CustomFieldModelCSVForm):
device = CSVModelChoiceField( device = CSVModelChoiceField(

123
netbox/secrets/tests/test_api.py
View File

@ -5,7 +5,7 @@ from rest_framework import status
from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site
from secrets.models import Secret, SecretRole, SessionKey, UserKey from secrets.models import Secret, SecretRole, SessionKey, UserKey
from utilities.testing import APITestCase from utilities.testing import APITestCase, APIViewTestCases
from .constants import PRIVATE_KEY, PUBLIC_KEY from .constants import PRIVATE_KEY, PUBLIC_KEY
@ -19,107 +19,36 @@ class AppTest(APITestCase):
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
class SecretRoleTest(APITestCase): class SecretRoleTest(APIViewTestCases.APIViewTestCase):
model = SecretRole
brief_fields = ['id', 'name', 'secret_count', 'slug', 'url']
create_data = [
{
'name': 'Secret Role 4',
'slug': 'secret-role-4',
},
{
'name': 'Secret Role 5',
'slug': 'secret-role-5',
},
{
'name': 'Secret Role 6',
'slug': 'secret-role-6',
},
]
def setUp(self): @classmethod
def setUpTestData(cls):
super().setUp() secret_roles = (
SecretRole(name='Secret Role 1', slug='secret-role-1'),
self.secretrole1 = SecretRole.objects.create(name='Test Secret Role 1', slug='test-secret-role-1') SecretRole(name='Secret Role 2', slug='secret-role-2'),
self.secretrole2 = SecretRole.objects.create(name='Test Secret Role 2', slug='test-secret-role-2') SecretRole(name='Secret Role 3', slug='secret-role-3'),
self.secretrole3 = SecretRole.objects.create(name='Test Secret Role 3', slug='test-secret-role-3')
def test_get_secretrole(self):
url = reverse('secrets-api:secretrole-detail', kwargs={'pk': self.secretrole1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.secretrole1.name)
def test_list_secretroles(self):
url = reverse('secrets-api:secretrole-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_secretroles_brief(self):
url = reverse('secrets-api:secretrole-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['id', 'name', 'secret_count', 'slug', 'url']
) )
SecretRole.objects.bulk_create(secret_roles)
def test_create_secretrole(self):
data = {
'name': 'Test Secret Role 4',
'slug': 'test-secret-role-4',
}
url = reverse('secrets-api:secretrole-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(SecretRole.objects.count(), 4)
secretrole4 = SecretRole.objects.get(pk=response.data['id'])
self.assertEqual(secretrole4.name, data['name'])
self.assertEqual(secretrole4.slug, data['slug'])
def test_create_secretrole_bulk(self):
data = [
{
'name': 'Test Secret Role 4',
'slug': 'test-secret-role-4',
},
{
'name': 'Test Secret Role 5',
'slug': 'test-secret-role-5',
},
{
'name': 'Test Secret Role 6',
'slug': 'test-secret-role-6',
},
]
url = reverse('secrets-api:secretrole-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(SecretRole.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_secretrole(self):
data = {
'name': 'Test SecretRole X',
'slug': 'test-secretrole-x',
}
url = reverse('secrets-api:secretrole-detail', kwargs={'pk': self.secretrole1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(SecretRole.objects.count(), 3)
secretrole1 = SecretRole.objects.get(pk=response.data['id'])
self.assertEqual(secretrole1.name, data['name'])
self.assertEqual(secretrole1.slug, data['slug'])
def test_delete_secretrole(self):
url = reverse('secrets-api:secretrole-detail', kwargs={'pk': self.secretrole1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(SecretRole.objects.count(), 2)
# TODO: Standardize SecretTest
class SecretTest(APITestCase): class SecretTest(APITestCase):
def setUp(self): def setUp(self):

242
netbox/tenancy/tests/test_api.py
View File

@ -1,8 +1,7 @@
from django.urls import reverse from django.urls import reverse
from rest_framework import status
from tenancy.models import Tenant, TenantGroup from tenancy.models import Tenant, TenantGroup
from utilities.testing import APITestCase from utilities.testing import APITestCase, APIViewTestCases
class AppTest(APITestCase): class AppTest(APITestCase):
@ -15,235 +14,74 @@ class AppTest(APITestCase):
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
class TenantGroupTest(APITestCase): class TenantGroupTest(APIViewTestCases.APIViewTestCase):
model = TenantGroup
brief_fields = ['id', 'name', 'slug', 'tenant_count', 'url']
def setUp(self): @classmethod
def setUpTestData(cls):
super().setUp() parent_tenant_groups = (
TenantGroup.objects.create(name='Parent Tenant Group 1', slug='parent-tenant-group-1'),
self.parent_tenant_groups = ( TenantGroup.objects.create(name='Parent Tenant Group 2', slug='parent-tenant-group-2'),
TenantGroup(name='Parent Tenant Group 1', slug='parent-tenant-group-1'),
TenantGroup(name='Parent Tenant Group 2', slug='parent-tenant-group-2'),
)
for tenantgroup in self.parent_tenant_groups:
tenantgroup.save()
self.tenant_groups = (
TenantGroup(name='Tenant Group 1', slug='tenant-group-1', parent=self.parent_tenant_groups[0]),
TenantGroup(name='Tenant Group 2', slug='tenant-group-2', parent=self.parent_tenant_groups[0]),
TenantGroup(name='Tenant Group 3', slug='tenant-group-3', parent=self.parent_tenant_groups[0]),
)
for tenantgroup in self.tenant_groups:
tenantgroup.save()
def test_get_tenantgroup(self):
url = reverse('tenancy-api:tenantgroup-detail', kwargs={'pk': self.tenant_groups[0].pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.tenant_groups[0].name)
def test_list_tenantgroups(self):
url = reverse('tenancy-api:tenantgroup-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 5)
def test_list_tenantgroups_brief(self):
url = reverse('tenancy-api:tenantgroup-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['id', 'name', 'slug', 'tenant_count', 'url']
) )
def test_create_tenantgroup(self): TenantGroup.objects.create(name='Tenant Group 1', slug='tenant-group-1', parent=parent_tenant_groups[0])
TenantGroup.objects.create(name='Tenant Group 2', slug='tenant-group-2', parent=parent_tenant_groups[0])
TenantGroup.objects.create(name='Tenant Group 3', slug='tenant-group-3', parent=parent_tenant_groups[0])
data = { cls.create_data = [
'name': 'Tenant Group 4',
'slug': 'tenant-group-4',
'parent': self.parent_tenant_groups[0].pk,
}
url = reverse('tenancy-api:tenantgroup-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(TenantGroup.objects.count(), 6)
tenantgroup4 = TenantGroup.objects.get(pk=response.data['id'])
self.assertEqual(tenantgroup4.name, data['name'])
self.assertEqual(tenantgroup4.slug, data['slug'])
self.assertEqual(tenantgroup4.parent_id, data['parent'])
def test_create_tenantgroup_bulk(self):
data = [
{ {
'name': 'Tenant Group 4', 'name': 'Tenant Group 4',
'slug': 'tenant-group-4', 'slug': 'tenant-group-4',
'parent': self.parent_tenant_groups[0].pk, 'parent': parent_tenant_groups[1].pk,
}, },
{ {
'name': 'Tenant Group 5', 'name': 'Tenant Group 5',
'slug': 'tenant-group-5', 'slug': 'tenant-group-5',
'parent': self.parent_tenant_groups[0].pk, 'parent': parent_tenant_groups[1].pk,
}, },
{ {
'name': 'Tenant Group 6', 'name': 'Tenant Group 6',
'slug': 'tenant-group-6', 'slug': 'tenant-group-6',
'parent': self.parent_tenant_groups[0].pk, 'parent': parent_tenant_groups[1].pk,
}, },
] ]
url = reverse('tenancy-api:tenantgroup-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED) class TenantTest(APIViewTestCases.APIViewTestCase):
self.assertEqual(TenantGroup.objects.count(), 8) model = Tenant
self.assertEqual(response.data[0]['name'], data[0]['name']) brief_fields = ['id', 'name', 'slug', 'url']
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_tenantgroup(self): @classmethod
def setUpTestData(cls):
data = { tenant_groups = (
'name': 'Tenant Group X', TenantGroup.objects.create(name='Tenant Group 1', slug='tenant-group-1'),
'slug': 'tenant-group-x', TenantGroup.objects.create(name='Tenant Group 2', slug='tenant-group-2'),
'parent': self.parent_tenant_groups[1].pk,
}
url = reverse('tenancy-api:tenantgroup-detail', kwargs={'pk': self.tenant_groups[0].pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(TenantGroup.objects.count(), 5)
tenantgroup1 = TenantGroup.objects.get(pk=response.data['id'])
self.assertEqual(tenantgroup1.name, data['name'])
self.assertEqual(tenantgroup1.slug, data['slug'])
self.assertEqual(tenantgroup1.parent_id, data['parent'])
def test_delete_tenantgroup(self):
url = reverse('tenancy-api:tenantgroup-detail', kwargs={'pk': self.tenant_groups[0].pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(TenantGroup.objects.count(), 4)
class TenantTest(APITestCase):
def setUp(self):
super().setUp()
self.tenant_groups = (
TenantGroup(name='Tenant Group 1', slug='tenant-group-1'),
TenantGroup(name='Tenant Group 2', slug='tenant-group-2'),
)
for tenantgroup in self.tenant_groups:
tenantgroup.save()
self.tenants = (
Tenant(name='Test Tenant 1', slug='test-tenant-1', group=self.tenant_groups[0]),
Tenant(name='Test Tenant 2', slug='test-tenant-2', group=self.tenant_groups[0]),
Tenant(name='Test Tenant 3', slug='test-tenant-3', group=self.tenant_groups[0]),
)
Tenant.objects.bulk_create(self.tenants)
def test_get_tenant(self):
url = reverse('tenancy-api:tenant-detail', kwargs={'pk': self.tenants[0].pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.tenants[0].name)
def test_list_tenants(self):
url = reverse('tenancy-api:tenant-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_tenants_brief(self):
url = reverse('tenancy-api:tenant-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['id', 'name', 'slug', 'url']
) )
def test_create_tenant(self): tenants = (
Tenant(name='Tenant 1', slug='tenant-1', group=tenant_groups[0]),
Tenant(name='Tenant 2', slug='tenant-2', group=tenant_groups[0]),
Tenant(name='Tenant 3', slug='tenant-3', group=tenant_groups[0]),
)
Tenant.objects.bulk_create(tenants)
data = { cls.create_data = [
'name': 'Test Tenant 4',
'slug': 'test-tenant-4',
'group': self.tenant_groups[0].pk,
}
url = reverse('tenancy-api:tenant-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Tenant.objects.count(), 4)
tenant4 = Tenant.objects.get(pk=response.data['id'])
self.assertEqual(tenant4.name, data['name'])
self.assertEqual(tenant4.slug, data['slug'])
self.assertEqual(tenant4.group_id, data['group'])
def test_create_tenant_bulk(self):
data = [
{ {
'name': 'Test Tenant 4', 'name': 'Tenant 4',
'slug': 'test-tenant-4', 'slug': 'tenant-4',
'group': tenant_groups[1].pk,
}, },
{ {
'name': 'Test Tenant 5', 'name': 'Tenant 5',
'slug': 'test-tenant-5', 'slug': 'tenant-5',
'group': tenant_groups[1].pk,
}, },
{ {
'name': 'Test Tenant 6', 'name': 'Tenant 6',
'slug': 'test-tenant-6', 'slug': 'tenant-6',
'group': tenant_groups[1].pk,
}, },
] ]
url = reverse('tenancy-api:tenant-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Tenant.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_tenant(self):
data = {
'name': 'Test Tenant X',
'slug': 'test-tenant-x',
'group': self.tenant_groups[1].pk,
}
url = reverse('tenancy-api:tenant-detail', kwargs={'pk': self.tenants[0].pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(Tenant.objects.count(), 3)
tenant1 = Tenant.objects.get(pk=response.data['id'])
self.assertEqual(tenant1.name, data['name'])
self.assertEqual(tenant1.slug, data['slug'])
self.assertEqual(tenant1.group_id, data['group'])
def test_delete_tenant(self):
url = reverse('tenancy-api:tenant-detail', kwargs={'pk': self.tenants[0].pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(Tenant.objects.count(), 2)

216
netbox/utilities/testing/testcases.py
View File

@ -1,9 +1,11 @@
from django.contrib.contenttypes.models import ContentType
from django.contrib.auth.models import User from django.contrib.auth.models import User
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist from django.core.exceptions import ObjectDoesNotExist
from django.forms.models import model_to_dict from django.forms.models import model_to_dict
from django.test import Client, TestCase as _TestCase, override_settings from django.test import Client, TestCase as _TestCase, override_settings
from django.urls import reverse, NoReverseMatch from django.urls import reverse, NoReverseMatch
from netaddr import IPNetwork
from rest_framework import status
from rest_framework.test import APIClient from rest_framework.test import APIClient
from users.models import ObjectPermission, Token from users.models import ObjectPermission, Token
@ -52,6 +54,49 @@ class TestCase(_TestCase):
expected_status, response.status_code, getattr(response, 'data', 'No data') expected_status, response.status_code, getattr(response, 'data', 'No data')
)) ))
def assertInstanceEqual(self, instance, data, api=False):
"""
Compare a model instance to a dictionary, checking that its attribute values match those specified
in the dictionary.
:instance: Python object instance
:data: Dictionary of test data used to define the instance
:api: Set to True is the data is a JSON representation of the instance
"""
model_dict = model_to_dict(instance, fields=data.keys())
for key, value in list(model_dict.items()):
# TODO: Differentiate between tags assigned to the instance and a M2M field for tags (ex: ConfigContext)
if key == 'tags':
model_dict[key] = ','.join(sorted([tag.name for tag in value]))
# Convert ManyToManyField to list of instance PKs
elif model_dict[key] and type(value) in (list, tuple) and hasattr(value[0], 'pk'):
model_dict[key] = [obj.pk for obj in value]
if api:
# Replace ContentType numeric IDs with <app_label>.<model>
if type(getattr(instance, key)) is ContentType:
ct = ContentType.objects.get(pk=value)
model_dict[key] = f'{ct.app_label}.{ct.model}'
# Convert IPNetwork instances to strings
if type(value) is IPNetwork:
model_dict[key] = str(value)
# Omit any dictionary keys which are not instance attributes
relevant_data = {
k: v for k, v in data.items() if hasattr(instance, k)
}
self.assertDictEqual(model_dict, relevant_data)
#
# UI Tests
#
class ModelViewTestCase(TestCase): class ModelViewTestCase(TestCase):
""" """
@ -99,42 +144,6 @@ class ModelViewTestCase(TestCase):
else: else:
raise Exception("Invalid action for URL resolution: {}".format(action)) raise Exception("Invalid action for URL resolution: {}".format(action))
def assertInstanceEqual(self, instance, data):
"""
Compare a model instance to a dictionary, checking that its attribute values match those specified
in the dictionary.
"""
model_dict = model_to_dict(instance, fields=data.keys())
for key in list(model_dict.keys()):
# TODO: Differentiate between tags assigned to the instance and a M2M field for tags (ex: ConfigContext)
if key == 'tags':
model_dict[key] = ','.join(sorted([tag.name for tag in model_dict['tags']]))
# Convert ManyToManyField to list of instance PKs
elif model_dict[key] and type(model_dict[key]) in (list, tuple) and hasattr(model_dict[key][0], 'pk'):
model_dict[key] = [obj.pk for obj in model_dict[key]]
# Omit any dictionary keys which are not instance attributes
relevant_data = {
k: v for k, v in data.items() if hasattr(instance, k)
}
self.assertDictEqual(model_dict, relevant_data)
class APITestCase(TestCase):
client_class = APIClient
def setUp(self):
"""
Create a superuser and token for API calls.
"""
self.user = User.objects.create(username='testuser', is_superuser=True)
self.token = Token.objects.create(user=self.user)
self.header = {'HTTP_AUTHORIZATION': 'Token {}'.format(self.token.key)}
class ViewTestCases: class ViewTestCases:
""" """
@ -193,6 +202,13 @@ class ViewTestCases:
# Try GET to non-permitted object # Try GET to non-permitted object
self.assertHttpStatus(self.client.get(instance2.get_absolute_url()), 404) self.assertHttpStatus(self.client.get(instance2.get_absolute_url()), 404)
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_get_object_anonymous(self):
# Make the request as an unauthenticated user
self.client.logout()
response = self.client.get(self.model.objects.first().get_absolute_url())
self.assertHttpStatus(response, 200)
class CreateObjectViewTestCase(ModelViewTestCase): class CreateObjectViewTestCase(ModelViewTestCase):
""" """
Create a single new instance. Create a single new instance.
@ -783,3 +799,129 @@ class ViewTestCases:
TestCase suitable for testing device component models (ConsolePorts, Interfaces, etc.) TestCase suitable for testing device component models (ConsolePorts, Interfaces, etc.)
""" """
maxDiff = None maxDiff = None
#
# REST API Tests
#
class APITestCase(TestCase):
client_class = APIClient
model = None
def setUp(self):
"""
Create a superuser and token for API calls.
"""
self.user = User.objects.create(username='testuser', is_superuser=True)
self.token = Token.objects.create(user=self.user)
self.header = {'HTTP_AUTHORIZATION': 'Token {}'.format(self.token.key)}
def _get_detail_url(self, instance):
viewname = f'{instance._meta.app_label}-api:{instance._meta.model_name}-detail'
return reverse(viewname, kwargs={'pk': instance.pk})
def _get_list_url(self):
viewname = f'{self.model._meta.app_label}-api:{self.model._meta.model_name}-list'
return reverse(viewname)
class APIViewTestCases:
class GetObjectViewTestCase(APITestCase):
def test_get_object(self):
"""
GET a single object identified by its numeric ID.
"""
instance = self.model.objects.first()
url = self._get_detail_url(instance)
response = self.client.get(url, **self.header)
self.assertEqual(response.data['id'], instance.pk)
class ListObjectsViewTestCase(APITestCase):
brief_fields = []
def test_list_objects(self):
"""
GET a list of objects.
"""
url = self._get_list_url()
response = self.client.get(url, **self.header)
self.assertEqual(len(response.data['results']), self.model.objects.count())
def test_list_objects_brief(self):
"""
GET a list of objects using the "brief" parameter.
"""
url = f'{self._get_list_url()}?brief=1'
response = self.client.get(url, **self.header)
self.assertEqual(len(response.data['results']), self.model.objects.count())
self.assertEqual(sorted(response.data['results'][0]), self.brief_fields)
class CreateObjectViewTestCase(APITestCase):
create_data = []
def test_create_object(self):
"""
POST a single object.
"""
initial_count = self.model.objects.count()
url = self._get_list_url()
response = self.client.post(url, self.create_data[0], format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(self.model.objects.count(), initial_count + 1)
self.assertInstanceEqual(self.model.objects.get(pk=response.data['id']), self.create_data[0], api=True)
def test_bulk_create_object(self):
"""
POST a set of objects in a single request.
"""
initial_count = self.model.objects.count()
url = self._get_list_url()
response = self.client.post(url, self.create_data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(self.model.objects.count(), initial_count + len(self.create_data))
class UpdateObjectViewTestCase(APITestCase):
update_data = {}
def test_update_object(self):
"""
PATCH a single object identified by its numeric ID.
"""
instance = self.model.objects.first()
url = self._get_detail_url(instance)
update_data = self.update_data or getattr(self, 'create_data')[0]
response = self.client.patch(url, update_data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
instance.refresh_from_db()
self.assertInstanceEqual(instance, self.update_data, api=True)
class DeleteObjectViewTestCase(APITestCase):
def test_delete_object(self):
"""
DELETE a single object identified by its numeric ID.
"""
instance = self.model.objects.first()
url = self._get_detail_url(instance)
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertFalse(self.model.objects.filter(pk=instance.pk).exists())
class APIViewTestCase(
GetObjectViewTestCase,
ListObjectsViewTestCase,
CreateObjectViewTestCase,
UpdateObjectViewTestCase,
DeleteObjectViewTestCase
):
pass

557
netbox/virtualization/tests/test_api.py
View File

@ -1,11 +1,10 @@
from django.urls import reverse from django.urls import reverse
from netaddr import IPNetwork
from rest_framework import status from rest_framework import status
from dcim.choices import InterfaceModeChoices from dcim.choices import InterfaceModeChoices
from dcim.models import Interface from dcim.models import Interface
from ipam.models import IPAddress, VLAN from ipam.models import VLAN
from utilities.testing import APITestCase, disable_warnings from utilities.testing import APITestCase, APIViewTestCases
from virtualization.choices import * from virtualization.choices import *
from virtualization.models import Cluster, ClusterGroup, ClusterType, VirtualMachine from virtualization.models import Cluster, ClusterGroup, ClusterType, VirtualMachine
@ -20,487 +19,181 @@ class AppTest(APITestCase):
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
class ClusterTypeTest(APITestCase): class ClusterTypeTest(APIViewTestCases.APIViewTestCase):
model = ClusterType
brief_fields = ['cluster_count', 'id', 'name', 'slug', 'url']
create_data = [
{
'name': 'Cluster Type 4',
'slug': 'cluster-type-4',
},
{
'name': 'Cluster Type 5',
'slug': 'cluster-type-5',
},
{
'name': 'Cluster Type 6',
'slug': 'cluster-type-6',
},
]
def setUp(self): @classmethod
def setUpTestData(cls):
super().setUp() cluster_types = (
ClusterType(name='Cluster Type 1', slug='cluster-type-1'),
self.clustertype1 = ClusterType.objects.create(name='Test Cluster Type 1', slug='test-cluster-type-1') ClusterType(name='Cluster Type 2', slug='cluster-type-2'),
self.clustertype2 = ClusterType.objects.create(name='Test Cluster Type 2', slug='test-cluster-type-2') ClusterType(name='Cluster Type 3', slug='cluster-type-3'),
self.clustertype3 = ClusterType.objects.create(name='Test Cluster Type 3', slug='test-cluster-type-3')
def test_get_clustertype(self):
url = reverse('virtualization-api:clustertype-detail', kwargs={'pk': self.clustertype1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.clustertype1.name)
def test_list_clustertypes(self):
url = reverse('virtualization-api:clustertype-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_clustertypes_brief(self):
url = reverse('virtualization-api:clustertype-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['cluster_count', 'id', 'name', 'slug', 'url']
) )
ClusterType.objects.bulk_create(cluster_types)
def test_create_clustertype(self):
data = { class ClusterGroupTest(APIViewTestCases.APIViewTestCase):
'name': 'Test Cluster Type 4', model = ClusterGroup
'slug': 'test-cluster-type-4', brief_fields = ['cluster_count', 'id', 'name', 'slug', 'url']
} create_data = [
{
'name': 'Cluster Group 4',
'slug': 'cluster-type-4',
},
{
'name': 'Cluster Group 5',
'slug': 'cluster-type-5',
},
{
'name': 'Cluster Group 6',
'slug': 'cluster-type-6',
},
]
url = reverse('virtualization-api:clustertype-list') @classmethod
response = self.client.post(url, data, format='json', **self.header) def setUpTestData(cls):
self.assertHttpStatus(response, status.HTTP_201_CREATED) cluster_Groups = (
self.assertEqual(ClusterType.objects.count(), 4) ClusterGroup(name='Cluster Group 1', slug='cluster-type-1'),
clustertype4 = ClusterType.objects.get(pk=response.data['id']) ClusterGroup(name='Cluster Group 2', slug='cluster-type-2'),
self.assertEqual(clustertype4.name, data['name']) ClusterGroup(name='Cluster Group 3', slug='cluster-type-3'),
self.assertEqual(clustertype4.slug, data['slug']) )
ClusterGroup.objects.bulk_create(cluster_Groups)
def test_create_clustertype_bulk(self):
data = [ class ClusterTest(APIViewTestCases.APIViewTestCase):
model = Cluster
brief_fields = ['id', 'name', 'url', 'virtualmachine_count']
@classmethod
def setUpTestData(cls):
cluster_types = (
ClusterType(name='Cluster Type 1', slug='cluster-type-1'),
ClusterType(name='Cluster Type 2', slug='cluster-type-2'),
)
ClusterType.objects.bulk_create(cluster_types)
cluster_groups = (
ClusterGroup(name='Cluster Group 1', slug='cluster-group-1'),
ClusterGroup(name='Cluster Group 2', slug='cluster-group-2'),
)
ClusterGroup.objects.bulk_create(cluster_groups)
clusters = (
Cluster(name='Cluster 1', type=cluster_types[0], group=cluster_groups[0]),
Cluster(name='Cluster 2', type=cluster_types[0], group=cluster_groups[0]),
Cluster(name='Cluster 3', type=cluster_types[0], group=cluster_groups[0]),
)
Cluster.objects.bulk_create(clusters)
cls.create_data = [
{ {
'name': 'Test Cluster Type 4', 'name': 'Cluster 4',
'slug': 'test-cluster-type-4', 'type': cluster_types[1].pk,
'group': cluster_groups[1].pk,
}, },
{ {
'name': 'Test Cluster Type 5', 'name': 'Cluster 5',
'slug': 'test-cluster-type-5', 'type': cluster_types[1].pk,
'group': cluster_groups[1].pk,
}, },
{ {
'name': 'Test Cluster Type 6', 'name': 'Cluster 6',
'slug': 'test-cluster-type-6', 'type': cluster_types[1].pk,
'group': cluster_groups[1].pk,
}, },
] ]
url = reverse('virtualization-api:clustertype-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED) class VirtualMachineTest(APIViewTestCases.APIViewTestCase):
self.assertEqual(ClusterType.objects.count(), 6) model = VirtualMachine
self.assertEqual(response.data[0]['name'], data[0]['name']) brief_fields = ['id', 'name', 'url']
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_clustertype(self): @classmethod
def setUpTestData(cls):
clustertype = ClusterType.objects.create(name='Cluster Type 1', slug='cluster-type-1')
clustergroup = ClusterGroup.objects.create(name='Cluster Group 1', slug='cluster-group-1')
data = { clusters = (
'name': 'Test Cluster Type X', Cluster(name='Cluster 1', type=clustertype, group=clustergroup),
'slug': 'test-cluster-type-x', Cluster(name='Cluster 2', type=clustertype, group=clustergroup),
}
url = reverse('virtualization-api:clustertype-detail', kwargs={'pk': self.clustertype1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(ClusterType.objects.count(), 3)
clustertype1 = ClusterType.objects.get(pk=response.data['id'])
self.assertEqual(clustertype1.name, data['name'])
self.assertEqual(clustertype1.slug, data['slug'])
def test_delete_clustertype(self):
url = reverse('virtualization-api:clustertype-detail', kwargs={'pk': self.clustertype1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(ClusterType.objects.count(), 2)
class ClusterGroupTest(APITestCase):
def setUp(self):
super().setUp()
self.clustergroup1 = ClusterGroup.objects.create(name='Test Cluster Group 1', slug='test-cluster-group-1')
self.clustergroup2 = ClusterGroup.objects.create(name='Test Cluster Group 2', slug='test-cluster-group-2')
self.clustergroup3 = ClusterGroup.objects.create(name='Test Cluster Group 3', slug='test-cluster-group-3')
def test_get_clustergroup(self):
url = reverse('virtualization-api:clustergroup-detail', kwargs={'pk': self.clustergroup1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.clustergroup1.name)
def test_list_clustergroups(self):
url = reverse('virtualization-api:clustergroup-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_clustergroups_brief(self):
url = reverse('virtualization-api:clustergroup-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['cluster_count', 'id', 'name', 'slug', 'url']
) )
Cluster.objects.bulk_create(clusters)
def test_create_clustergroup(self): virtual_machines = (
VirtualMachine(name='Virtual Machine 1', cluster=clusters[0], local_context_data={'A': 1}),
VirtualMachine(name='Virtual Machine 2', cluster=clusters[0], local_context_data={'B': 2}),
VirtualMachine(name='Virtual Machine 3', cluster=clusters[0], local_context_data={'C': 3}),
)
VirtualMachine.objects.bulk_create(virtual_machines)
data = { cls.create_data = [
'name': 'Test Cluster Group 4',
'slug': 'test-cluster-group-4',
}
url = reverse('virtualization-api:clustergroup-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(ClusterGroup.objects.count(), 4)
clustergroup4 = ClusterGroup.objects.get(pk=response.data['id'])
self.assertEqual(clustergroup4.name, data['name'])
self.assertEqual(clustergroup4.slug, data['slug'])
def test_create_clustergroup_bulk(self):
data = [
{ {
'name': 'Test Cluster Group 4', 'name': 'Virtual Machine 4',
'slug': 'test-cluster-group-4', 'cluster': clusters[1].pk,
}, },
{ {
'name': 'Test Cluster Group 5', 'name': 'Virtual Machine 5',
'slug': 'test-cluster-group-5', 'cluster': clusters[1].pk,
}, },
{ {
'name': 'Test Cluster Group 6', 'name': 'Virtual Machine 6',
'slug': 'test-cluster-group-6', 'cluster': clusters[1].pk,
}, },
] ]
url = reverse('virtualization-api:clustergroup-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(ClusterGroup.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_clustergroup(self):
data = {
'name': 'Test Cluster Group X',
'slug': 'test-cluster-group-x',
}
url = reverse('virtualization-api:clustergroup-detail', kwargs={'pk': self.clustergroup1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(ClusterGroup.objects.count(), 3)
clustergroup1 = ClusterGroup.objects.get(pk=response.data['id'])
self.assertEqual(clustergroup1.name, data['name'])
self.assertEqual(clustergroup1.slug, data['slug'])
def test_delete_clustergroup(self):
url = reverse('virtualization-api:clustergroup-detail', kwargs={'pk': self.clustergroup1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(ClusterGroup.objects.count(), 2)
class ClusterTest(APITestCase):
def setUp(self):
super().setUp()
cluster_type = ClusterType.objects.create(name='Test Cluster Type 1', slug='test-cluster-type-1')
cluster_group = ClusterGroup.objects.create(name='Test Cluster Group 1', slug='test-cluster-group-1')
self.cluster1 = Cluster.objects.create(name='Test Cluster 1', type=cluster_type, group=cluster_group)
self.cluster2 = Cluster.objects.create(name='Test Cluster 2', type=cluster_type, group=cluster_group)
self.cluster3 = Cluster.objects.create(name='Test Cluster 3', type=cluster_type, group=cluster_group)
def test_get_cluster(self):
url = reverse('virtualization-api:cluster-detail', kwargs={'pk': self.cluster1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.cluster1.name)
def test_list_clusters(self):
url = reverse('virtualization-api:cluster-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_clusters_brief(self):
url = reverse('virtualization-api:cluster-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['id', 'name', 'url', 'virtualmachine_count']
)
def test_create_cluster(self):
data = {
'name': 'Test Cluster 4',
'type': ClusterType.objects.first().pk,
'group': ClusterGroup.objects.first().pk,
}
url = reverse('virtualization-api:cluster-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Cluster.objects.count(), 4)
cluster4 = Cluster.objects.get(pk=response.data['id'])
self.assertEqual(cluster4.name, data['name'])
self.assertEqual(cluster4.type.pk, data['type'])
self.assertEqual(cluster4.group.pk, data['group'])
def test_create_cluster_bulk(self):
data = [
{
'name': 'Test Cluster 4',
'type': ClusterType.objects.first().pk,
'group': ClusterGroup.objects.first().pk,
},
{
'name': 'Test Cluster 5',
'type': ClusterType.objects.first().pk,
'group': ClusterGroup.objects.first().pk,
},
{
'name': 'Test Cluster 6',
'type': ClusterType.objects.first().pk,
'group': ClusterGroup.objects.first().pk,
},
]
url = reverse('virtualization-api:cluster-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Cluster.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_cluster(self):
cluster_type2 = ClusterType.objects.create(name='Test Cluster Type 2', slug='test-cluster-type-2')
cluster_group2 = ClusterGroup.objects.create(name='Test Cluster Group 2', slug='test-cluster-group-2')
data = {
'name': 'Test Cluster X',
'type': cluster_type2.pk,
'group': cluster_group2.pk,
}
url = reverse('virtualization-api:cluster-detail', kwargs={'pk': self.cluster1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(Cluster.objects.count(), 3)
cluster1 = Cluster.objects.get(pk=response.data['id'])
self.assertEqual(cluster1.name, data['name'])
self.assertEqual(cluster1.type.pk, data['type'])
self.assertEqual(cluster1.group.pk, data['group'])
def test_delete_cluster(self):
url = reverse('virtualization-api:cluster-detail', kwargs={'pk': self.cluster1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(Cluster.objects.count(), 2)
class VirtualMachineTest(APITestCase):
def setUp(self):
super().setUp()
cluster_type = ClusterType.objects.create(name='Test Cluster Type 1', slug='test-cluster-type-1')
cluster_group = ClusterGroup.objects.create(name='Test Cluster Group 1', slug='test-cluster-group-1')
self.cluster1 = Cluster.objects.create(name='Test Cluster 1', type=cluster_type, group=cluster_group)
self.virtualmachine1 = VirtualMachine.objects.create(name='Test Virtual Machine 1', cluster=self.cluster1)
self.virtualmachine2 = VirtualMachine.objects.create(name='Test Virtual Machine 2', cluster=self.cluster1)
self.virtualmachine3 = VirtualMachine.objects.create(name='Test Virtual Machine 3', cluster=self.cluster1)
self.virtualmachine_with_context_data = VirtualMachine.objects.create(
name='VM with context data',
cluster=self.cluster1,
local_context_data={
'A': 1,
'B': 2
}
)
def test_get_virtualmachine(self):
url = reverse('virtualization-api:virtualmachine-detail', kwargs={'pk': self.virtualmachine1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.virtualmachine1.name)
def test_list_virtualmachines(self):
url = reverse('virtualization-api:virtualmachine-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 4)
def test_list_virtualmachines_brief(self):
url = reverse('virtualization-api:virtualmachine-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['id', 'name', 'url']
)
def test_create_virtualmachine(self):
data = {
'name': 'Test Virtual Machine 4',
'cluster': self.cluster1.pk,
}
url = reverse('virtualization-api:virtualmachine-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(VirtualMachine.objects.count(), 5)
virtualmachine4 = VirtualMachine.objects.get(pk=response.data['id'])
self.assertEqual(virtualmachine4.name, data['name'])
self.assertEqual(virtualmachine4.cluster.pk, data['cluster'])
def test_create_virtualmachine_without_cluster(self):
data = {
'name': 'Test Virtual Machine 4',
}
url = reverse('virtualization-api:virtualmachine-list')
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(VirtualMachine.objects.count(), 4)
def test_create_virtualmachine_bulk(self):
data = [
{
'name': 'Test Virtual Machine 4',
'cluster': self.cluster1.pk,
},
{
'name': 'Test Virtual Machine 5',
'cluster': self.cluster1.pk,
},
{
'name': 'Test Virtual Machine 6',
'cluster': self.cluster1.pk,
},
]
url = reverse('virtualization-api:virtualmachine-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(VirtualMachine.objects.count(), 7)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_virtualmachine(self):
interface = Interface.objects.create(name='Test Interface 1', virtual_machine=self.virtualmachine1)
ip4_address = IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), interface=interface)
ip6_address = IPAddress.objects.create(address=IPNetwork('2001:db8::1/64'), interface=interface)
cluster2 = Cluster.objects.create(
name='Test Cluster 2',
type=ClusterType.objects.first(),
group=ClusterGroup.objects.first()
)
data = {
'name': 'Test Virtual Machine X',
'cluster': cluster2.pk,
'primary_ip4': ip4_address.pk,
'primary_ip6': ip6_address.pk,
}
url = reverse('virtualization-api:virtualmachine-detail', kwargs={'pk': self.virtualmachine1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(VirtualMachine.objects.count(), 4)
virtualmachine1 = VirtualMachine.objects.get(pk=response.data['id'])
self.assertEqual(virtualmachine1.name, data['name'])
self.assertEqual(virtualmachine1.cluster.pk, data['cluster'])
self.assertEqual(virtualmachine1.primary_ip4.pk, data['primary_ip4'])
self.assertEqual(virtualmachine1.primary_ip6.pk, data['primary_ip6'])
def test_delete_virtualmachine(self):
url = reverse('virtualization-api:virtualmachine-detail', kwargs={'pk': self.virtualmachine1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(VirtualMachine.objects.count(), 3)
def test_config_context_included_by_default_in_list_view(self): def test_config_context_included_by_default_in_list_view(self):
"""
Check that config context data is included by default in the virtual machines list.
"""
virtualmachine = VirtualMachine.objects.first()
url = reverse('virtualization-api:virtualmachine-list') url = reverse('virtualization-api:virtualmachine-list')
url = '{}?id={}'.format(url, self.virtualmachine_with_context_data.pk) url = '{}?id={}'.format(url, virtualmachine.pk)
response = self.client.get(url, **self.header) response = self.client.get(url, **self.header)
self.assertEqual(response.data['results'][0].get('config_context', {}).get('A'), 1) self.assertEqual(response.data['results'][0].get('config_context', {}).get('A'), 1)
def test_config_context_excluded(self): def test_config_context_excluded(self):
"""
Check that config context data can be excluded by passing ?exclude=config_context.
"""
url = reverse('virtualization-api:virtualmachine-list') + '?exclude=config_context' url = reverse('virtualization-api:virtualmachine-list') + '?exclude=config_context'
response = self.client.get(url, **self.header) response = self.client.get(url, **self.header)
self.assertFalse('config_context' in response.data['results'][0]) self.assertFalse('config_context' in response.data['results'][0])
def test_unique_name_per_cluster_constraint(self): def test_unique_name_per_cluster_constraint(self):
"""
Check that creating a virtual machine with a duplicate name fails.
"""
data = { data = {
'name': 'Test Virtual Machine 1', 'name': 'Virtual Machine 1',
'cluster': self.cluster1.pk, 'cluster': Cluster.objects.first().pk,
} }
url = reverse('virtualization-api:virtualmachine-list') url = reverse('virtualization-api:virtualmachine-list')
response = self.client.post(url, data, format='json', **self.header) response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
# TODO: Standardize InterfaceTest (pending #4721)
class InterfaceTest(APITestCase): class InterfaceTest(APITestCase):
def setUp(self): def setUp(self):

1
netbox/virtualization/tests/test_views.py
View File

@ -187,6 +187,7 @@ class VirtualMachineTestCase(ViewTestCases.PrimaryObjectViewTestCase):
# TODO: Update base class to DeviceComponentViewTestCase # TODO: Update base class to DeviceComponentViewTestCase
class InterfaceTestCase( class InterfaceTestCase(
ViewTestCases.GetObjectViewTestCase,
ViewTestCases.EditObjectViewTestCase, ViewTestCases.EditObjectViewTestCase,
ViewTestCases.DeleteObjectViewTestCase, ViewTestCases.DeleteObjectViewTestCase,
ViewTestCases.BulkCreateObjectsViewTestCase, ViewTestCases.BulkCreateObjectsViewTestCase,