mirror of
https://github.com/netbox-community/netbox.git
synced 2026-03-24 10:22:07 +01:00
fd6e0e9784
Introduce a new configuration parameter, `CHANGELOG_RETAIN_CREATE_LAST_UPDATE`, to retain each object's create record and most recent update record when pruning expired changelog entries (per `CHANGELOG_RETENTION`). Update documentation, templates, and forms to reflect this change. Fixes #21409
800 lines
30 KiB
Python
800 lines
30 KiB
Python
import logging
|
|
import uuid
|
|
from datetime import timedelta
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.test import TestCase, override_settings
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from rest_framework import status
|
|
|
|
from core.choices import ObjectChangeActionChoices
|
|
from core.jobs import SystemHousekeepingJob
|
|
from core.models import ObjectChange, ObjectType
|
|
from dcim.choices import InterfaceTypeChoices, ModuleStatusChoices, SiteStatusChoices
|
|
from dcim.models import (
|
|
Cable,
|
|
CableTermination,
|
|
Device,
|
|
DeviceRole,
|
|
DeviceType,
|
|
Interface,
|
|
Manufacturer,
|
|
Module,
|
|
ModuleBay,
|
|
ModuleType,
|
|
Site,
|
|
)
|
|
from extras.choices import *
|
|
from extras.models import CustomField, CustomFieldChoiceSet, Tag
|
|
from utilities.testing import APITestCase
|
|
from utilities.testing.utils import create_tags, create_test_device, post_data
|
|
from utilities.testing.views import ModelViewTestCase
|
|
|
|
|
|
class ChangeLogViewTest(ModelViewTestCase):
|
|
model = Site
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
choice_set = CustomFieldChoiceSet.objects.create(
|
|
name='Choice Set 1',
|
|
extra_choices=(('foo', 'Foo'), ('bar', 'Bar'))
|
|
)
|
|
|
|
# Create a custom field on the Site model
|
|
site_type = ObjectType.objects.get_for_model(Site)
|
|
cf = CustomField(
|
|
type=CustomFieldTypeChoices.TYPE_TEXT,
|
|
name='cf1',
|
|
required=False
|
|
)
|
|
cf.save()
|
|
cf.object_types.set([site_type])
|
|
|
|
# Create a select custom field on the Site model
|
|
cf_select = CustomField(
|
|
type=CustomFieldTypeChoices.TYPE_SELECT,
|
|
name='cf2',
|
|
required=False,
|
|
choice_set=choice_set
|
|
)
|
|
cf_select.save()
|
|
cf_select.object_types.set([site_type])
|
|
|
|
def test_create_object(self):
|
|
tags = create_tags('Tag 1', 'Tag 2')
|
|
form_data = {
|
|
'name': 'Site 1',
|
|
'slug': 'site-1',
|
|
'status': SiteStatusChoices.STATUS_ACTIVE,
|
|
'cf_cf1': 'ABC',
|
|
'cf_cf2': 'bar',
|
|
'tags': [tag.pk for tag in tags],
|
|
}
|
|
|
|
request = {
|
|
'path': self._get_url('add'),
|
|
'data': post_data(form_data),
|
|
}
|
|
self.add_permissions('dcim.add_site', 'extras.view_tag')
|
|
response = self.client.post(**request)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
# Verify the creation of a new ObjectChange record
|
|
site = Site.objects.get(name='Site 1')
|
|
oc = ObjectChange.objects.get(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=site.pk
|
|
)
|
|
self.assertEqual(oc.changed_object, site)
|
|
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE)
|
|
self.assertEqual(oc.prechange_data, None)
|
|
self.assertEqual(oc.postchange_data['custom_fields']['cf1'], form_data['cf_cf1'])
|
|
self.assertEqual(oc.postchange_data['custom_fields']['cf2'], form_data['cf_cf2'])
|
|
self.assertEqual(oc.postchange_data['tags'], ['Tag 1', 'Tag 2'])
|
|
|
|
def test_update_object(self):
|
|
site = Site(name='Site 1', slug='site-1')
|
|
site.save()
|
|
tags = create_tags('Tag 1', 'Tag 2', 'Tag 3')
|
|
site.tags.set(['Tag 1', 'Tag 2'])
|
|
|
|
form_data = {
|
|
'name': 'Site X',
|
|
'slug': 'site-x',
|
|
'status': SiteStatusChoices.STATUS_PLANNED,
|
|
'cf_cf1': 'DEF',
|
|
'cf_cf2': 'foo',
|
|
'tags': [tags[2].pk],
|
|
}
|
|
|
|
request = {
|
|
'path': self._get_url('edit', instance=site),
|
|
'data': post_data(form_data),
|
|
}
|
|
self.add_permissions('dcim.change_site', 'extras.view_tag')
|
|
response = self.client.post(**request)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
# Verify the creation of a new ObjectChange record
|
|
site.refresh_from_db()
|
|
oc = ObjectChange.objects.filter(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=site.pk
|
|
).first()
|
|
self.assertEqual(oc.changed_object, site)
|
|
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_UPDATE)
|
|
self.assertEqual(oc.prechange_data['name'], 'Site 1')
|
|
self.assertEqual(oc.prechange_data['tags'], ['Tag 1', 'Tag 2'])
|
|
self.assertEqual(oc.postchange_data['custom_fields']['cf1'], form_data['cf_cf1'])
|
|
self.assertEqual(oc.postchange_data['custom_fields']['cf2'], form_data['cf_cf2'])
|
|
self.assertEqual(oc.postchange_data['tags'], ['Tag 3'])
|
|
|
|
def test_delete_object(self):
|
|
site = Site(
|
|
name='Site 1',
|
|
slug='site-1',
|
|
custom_field_data={
|
|
'cf1': 'ABC',
|
|
'cf2': 'Bar'
|
|
}
|
|
)
|
|
site.save()
|
|
create_tags('Tag 1', 'Tag 2')
|
|
site.tags.set(['Tag 1', 'Tag 2'])
|
|
|
|
request = {
|
|
'path': self._get_url('delete', instance=site),
|
|
'data': post_data({'confirm': True}),
|
|
}
|
|
self.add_permissions('dcim.delete_site')
|
|
response = self.client.post(**request)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
oc = ObjectChange.objects.first()
|
|
self.assertEqual(oc.changed_object, None)
|
|
self.assertEqual(oc.object_repr, site.name)
|
|
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_DELETE)
|
|
self.assertEqual(oc.prechange_data['custom_fields']['cf1'], 'ABC')
|
|
self.assertEqual(oc.prechange_data['custom_fields']['cf2'], 'Bar')
|
|
self.assertEqual(oc.prechange_data['tags'], ['Tag 1', 'Tag 2'])
|
|
self.assertEqual(oc.postchange_data, None)
|
|
|
|
def test_bulk_update_objects(self):
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE),
|
|
Site(name='Site 2', slug='site-2', status=SiteStatusChoices.STATUS_ACTIVE),
|
|
Site(name='Site 3', slug='site-3', status=SiteStatusChoices.STATUS_ACTIVE),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
form_data = {
|
|
'pk': [site.pk for site in sites],
|
|
'_apply': True,
|
|
'status': SiteStatusChoices.STATUS_PLANNED,
|
|
'description': 'New description',
|
|
}
|
|
|
|
request = {
|
|
'path': self._get_url('bulk_edit'),
|
|
'data': post_data(form_data),
|
|
}
|
|
self.add_permissions('dcim.view_site', 'dcim.change_site')
|
|
response = self.client.post(**request)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
objectchange = ObjectChange.objects.get(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=sites[0].pk
|
|
)
|
|
self.assertEqual(objectchange.changed_object, sites[0])
|
|
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_UPDATE)
|
|
self.assertEqual(objectchange.prechange_data['status'], SiteStatusChoices.STATUS_ACTIVE)
|
|
self.assertEqual(objectchange.prechange_data['description'], '')
|
|
self.assertEqual(objectchange.postchange_data['status'], form_data['status'])
|
|
self.assertEqual(objectchange.postchange_data['description'], form_data['description'])
|
|
|
|
def test_bulk_delete_objects(self):
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE),
|
|
Site(name='Site 2', slug='site-2', status=SiteStatusChoices.STATUS_ACTIVE),
|
|
Site(name='Site 3', slug='site-3', status=SiteStatusChoices.STATUS_ACTIVE),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
form_data = {
|
|
'pk': [site.pk for site in sites],
|
|
'confirm': True,
|
|
'_confirm': True,
|
|
}
|
|
|
|
request = {
|
|
'path': self._get_url('bulk_delete'),
|
|
'data': post_data(form_data),
|
|
}
|
|
self.add_permissions('dcim.delete_site')
|
|
response = self.client.post(**request)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
objectchange = ObjectChange.objects.get(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=sites[0].pk
|
|
)
|
|
self.assertEqual(objectchange.changed_object_type, ContentType.objects.get_for_model(Site))
|
|
self.assertEqual(objectchange.changed_object_id, sites[0].pk)
|
|
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_DELETE)
|
|
self.assertEqual(objectchange.prechange_data['name'], sites[0].name)
|
|
self.assertEqual(objectchange.prechange_data['slug'], sites[0].slug)
|
|
self.assertEqual(objectchange.postchange_data, None)
|
|
|
|
@override_settings(CHANGELOG_SKIP_EMPTY_CHANGES=False)
|
|
def test_update_object_change(self):
|
|
# Create a Site
|
|
site = Site.objects.create(
|
|
name='Site 1',
|
|
slug='site-1',
|
|
status=SiteStatusChoices.STATUS_PLANNED,
|
|
custom_field_data={
|
|
'cf1': None,
|
|
'cf2': None
|
|
}
|
|
)
|
|
|
|
# Update it with the same field values
|
|
form_data = {
|
|
'name': site.name,
|
|
'slug': site.slug,
|
|
'status': SiteStatusChoices.STATUS_PLANNED,
|
|
}
|
|
request = {
|
|
'path': self._get_url('edit', instance=site),
|
|
'data': post_data(form_data),
|
|
}
|
|
self.add_permissions('dcim.change_site', 'extras.view_tag')
|
|
response = self.client.post(**request)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
# Check that an ObjectChange record has been created
|
|
self.assertEqual(ObjectChange.objects.count(), 1)
|
|
|
|
@override_settings(CHANGELOG_SKIP_EMPTY_CHANGES=True)
|
|
def test_update_object_nochange(self):
|
|
# Create a Site
|
|
site = Site.objects.create(
|
|
name='Site 1',
|
|
slug='site-1',
|
|
status=SiteStatusChoices.STATUS_PLANNED,
|
|
custom_field_data={
|
|
'cf1': None,
|
|
'cf2': None
|
|
}
|
|
)
|
|
|
|
# Update it with the same field values
|
|
form_data = {
|
|
'name': site.name,
|
|
'slug': site.slug,
|
|
'status': SiteStatusChoices.STATUS_PLANNED,
|
|
}
|
|
request = {
|
|
'path': self._get_url('edit', instance=site),
|
|
'data': post_data(form_data),
|
|
}
|
|
self.add_permissions('dcim.change_site', 'extras.view_tag')
|
|
response = self.client.post(**request)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
# Check that no ObjectChange records have been created
|
|
self.assertEqual(ObjectChange.objects.count(), 0)
|
|
|
|
def test_ordering_genericrelation(self):
|
|
# Create required objects first
|
|
manufacturer = Manufacturer.objects.create(name='Manufacturer 1')
|
|
device_type = DeviceType.objects.create(
|
|
manufacturer=manufacturer,
|
|
model='Model 1',
|
|
slug='model-1'
|
|
)
|
|
device_role = DeviceRole.objects.create(
|
|
name='Role 1',
|
|
slug='role-1'
|
|
)
|
|
site = Site.objects.create(
|
|
name='Site 1',
|
|
slug='site-1'
|
|
)
|
|
|
|
# Create two devices
|
|
device1 = Device.objects.create(
|
|
name='Device 1',
|
|
device_type=device_type,
|
|
role=device_role,
|
|
site=site
|
|
)
|
|
device2 = Device.objects.create(
|
|
name='Device 2',
|
|
device_type=device_type,
|
|
role=device_role,
|
|
site=site
|
|
)
|
|
|
|
# Create interfaces on both devices
|
|
interface1 = Interface.objects.create(
|
|
device=device1,
|
|
name='eth0',
|
|
type='1000base-t'
|
|
)
|
|
interface2 = Interface.objects.create(
|
|
device=device2,
|
|
name='eth0',
|
|
type='1000base-t'
|
|
)
|
|
|
|
# Create a cable between the interfaces
|
|
_ = Cable.objects.create(
|
|
a_terminations=[interface1],
|
|
b_terminations=[interface2],
|
|
status='connected'
|
|
)
|
|
|
|
# Delete device1
|
|
request = {
|
|
'path': reverse('dcim:device_delete', kwargs={'pk': device1.pk}),
|
|
'data': post_data({'confirm': True}),
|
|
}
|
|
self.add_permissions(
|
|
'dcim.delete_device',
|
|
'dcim.delete_interface',
|
|
'dcim.delete_cable',
|
|
'dcim.delete_cabletermination'
|
|
)
|
|
response = self.client.post(**request)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
# Get the ObjectChange records for delete actions ordered by time
|
|
changes = ObjectChange.objects.filter(
|
|
action=ObjectChangeActionChoices.ACTION_DELETE
|
|
).order_by('time')[:3]
|
|
|
|
# Verify the order of deletion
|
|
self.assertEqual(len(changes), 3)
|
|
self.assertEqual(changes[0].changed_object_type, ContentType.objects.get_for_model(CableTermination))
|
|
self.assertEqual(changes[1].changed_object_type, ContentType.objects.get_for_model(Interface))
|
|
self.assertEqual(changes[2].changed_object_type, ContentType.objects.get_for_model(Device))
|
|
|
|
def test_duplicate_deletions(self):
|
|
"""
|
|
Check that a cascading deletion event does not generate multiple "deleted" ObjectChange records for
|
|
the same object.
|
|
"""
|
|
role1 = DeviceRole(name='Role 1', slug='role-1')
|
|
role1.save()
|
|
role2 = DeviceRole(name='Role 2', slug='role-2', parent=role1)
|
|
role2.save()
|
|
pk_list = [role1.pk, role2.pk]
|
|
|
|
# Delete both objects simultaneously
|
|
form_data = {
|
|
'pk': pk_list,
|
|
'confirm': True,
|
|
'_confirm': True,
|
|
}
|
|
request = {
|
|
'path': reverse('dcim:devicerole_bulk_delete'),
|
|
'data': post_data(form_data),
|
|
}
|
|
self.add_permissions('dcim.delete_devicerole')
|
|
self.assertHttpStatus(self.client.post(**request), 302)
|
|
|
|
# This should result in exactly one change record per object
|
|
objectchanges = ObjectChange.objects.filter(
|
|
changed_object_type=ContentType.objects.get_for_model(DeviceRole),
|
|
changed_object_id__in=pk_list,
|
|
action=ObjectChangeActionChoices.ACTION_DELETE
|
|
)
|
|
self.assertEqual(objectchanges.count(), 2)
|
|
|
|
|
|
class ChangeLogAPITest(APITestCase):
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
# Create a custom field on the Site model
|
|
site_type = ObjectType.objects.get_for_model(Site)
|
|
cf = CustomField(
|
|
type=CustomFieldTypeChoices.TYPE_TEXT,
|
|
name='cf1',
|
|
required=False
|
|
)
|
|
cf.save()
|
|
cf.object_types.set([site_type])
|
|
|
|
# Create a select custom field on the Site model
|
|
choice_set = CustomFieldChoiceSet.objects.create(
|
|
name='Choice Set 1',
|
|
extra_choices=(('foo', 'Foo'), ('bar', 'Bar'))
|
|
)
|
|
cf_select = CustomField(
|
|
type=CustomFieldTypeChoices.TYPE_SELECT,
|
|
name='cf2',
|
|
required=False,
|
|
choice_set=choice_set
|
|
)
|
|
cf_select.save()
|
|
cf_select.object_types.set([site_type])
|
|
|
|
# Create some tags
|
|
tags = (
|
|
Tag(name='Tag 1', slug='tag-1'),
|
|
Tag(name='Tag 2', slug='tag-2'),
|
|
Tag(name='Tag 3', slug='tag-3'),
|
|
)
|
|
Tag.objects.bulk_create(tags)
|
|
|
|
def test_create_object(self):
|
|
data = {
|
|
'name': 'Site 1',
|
|
'slug': 'site-1',
|
|
'custom_fields': {
|
|
'cf1': 'ABC',
|
|
'cf2': 'bar',
|
|
},
|
|
'tags': [
|
|
{'name': 'Tag 1'},
|
|
{'name': 'Tag 2'},
|
|
]
|
|
}
|
|
self.assertEqual(ObjectChange.objects.count(), 0)
|
|
url = reverse('dcim-api:site-list')
|
|
self.add_permissions('dcim.add_site')
|
|
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
|
|
site = Site.objects.get(pk=response.data['id'])
|
|
oc = ObjectChange.objects.get(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=site.pk
|
|
)
|
|
self.assertEqual(oc.changed_object, site)
|
|
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE)
|
|
self.assertEqual(oc.prechange_data, None)
|
|
self.assertEqual(oc.postchange_data['custom_fields'], data['custom_fields'])
|
|
self.assertEqual(oc.postchange_data['tags'], ['Tag 1', 'Tag 2'])
|
|
|
|
def test_update_object(self):
|
|
site = Site(name='Site 1', slug='site-1')
|
|
site.save()
|
|
|
|
data = {
|
|
'name': 'Site X',
|
|
'slug': 'site-x',
|
|
'custom_fields': {
|
|
'cf1': 'DEF',
|
|
'cf2': 'foo',
|
|
},
|
|
'tags': [
|
|
{'name': 'Tag 3'}
|
|
]
|
|
}
|
|
self.assertEqual(ObjectChange.objects.count(), 0)
|
|
self.add_permissions('dcim.change_site')
|
|
url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
|
|
|
|
response = self.client.put(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
|
|
site = Site.objects.get(pk=response.data['id'])
|
|
oc = ObjectChange.objects.get(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=site.pk
|
|
)
|
|
self.assertEqual(oc.changed_object, site)
|
|
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_UPDATE)
|
|
self.assertEqual(oc.postchange_data['custom_fields'], data['custom_fields'])
|
|
self.assertEqual(oc.postchange_data['tags'], ['Tag 3'])
|
|
|
|
def test_delete_object(self):
|
|
site = Site(
|
|
name='Site 1',
|
|
slug='site-1',
|
|
custom_field_data={
|
|
'cf1': 'ABC',
|
|
'cf2': 'Bar'
|
|
}
|
|
)
|
|
site.save()
|
|
site.tags.set(Tag.objects.all()[:2])
|
|
self.assertEqual(ObjectChange.objects.count(), 0)
|
|
self.add_permissions('dcim.delete_site')
|
|
url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
|
|
|
|
response = self.client.delete(url, **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
|
|
self.assertEqual(Site.objects.count(), 0)
|
|
|
|
oc = ObjectChange.objects.first()
|
|
self.assertEqual(oc.changed_object, None)
|
|
self.assertEqual(oc.object_repr, site.name)
|
|
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_DELETE)
|
|
self.assertEqual(oc.prechange_data['custom_fields']['cf1'], 'ABC')
|
|
self.assertEqual(oc.prechange_data['custom_fields']['cf2'], 'Bar')
|
|
self.assertEqual(oc.prechange_data['tags'], ['Tag 1', 'Tag 2'])
|
|
self.assertEqual(oc.postchange_data, None)
|
|
|
|
def test_bulk_create_objects(self):
|
|
data = (
|
|
{
|
|
'name': 'Site 1',
|
|
'slug': 'site-1',
|
|
},
|
|
{
|
|
'name': 'Site 2',
|
|
'slug': 'site-2',
|
|
},
|
|
{
|
|
'name': 'Site 3',
|
|
'slug': 'site-3',
|
|
},
|
|
)
|
|
self.assertEqual(ObjectChange.objects.count(), 0)
|
|
url = reverse('dcim-api:site-list')
|
|
self.add_permissions('dcim.add_site')
|
|
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
self.assertEqual(ObjectChange.objects.count(), 3)
|
|
|
|
site1 = Site.objects.get(pk=response.data[0]['id'])
|
|
objectchange = ObjectChange.objects.get(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=site1.pk
|
|
)
|
|
self.assertEqual(objectchange.changed_object, site1)
|
|
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_CREATE)
|
|
self.assertEqual(objectchange.prechange_data, None)
|
|
self.assertEqual(objectchange.postchange_data['name'], data[0]['name'])
|
|
self.assertEqual(objectchange.postchange_data['slug'], data[0]['slug'])
|
|
|
|
def test_bulk_edit_objects(self):
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1'),
|
|
Site(name='Site 2', slug='site-2'),
|
|
Site(name='Site 3', slug='site-3'),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
data = (
|
|
{
|
|
'id': sites[0].pk,
|
|
'name': 'Site A',
|
|
'slug': 'site-A',
|
|
},
|
|
{
|
|
'id': sites[1].pk,
|
|
'name': 'Site B',
|
|
'slug': 'site-b',
|
|
},
|
|
{
|
|
'id': sites[2].pk,
|
|
'name': 'Site C',
|
|
'slug': 'site-c',
|
|
},
|
|
)
|
|
self.assertEqual(ObjectChange.objects.count(), 0)
|
|
url = reverse('dcim-api:site-list')
|
|
self.add_permissions('dcim.change_site')
|
|
|
|
response = self.client.patch(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_200_OK)
|
|
self.assertEqual(ObjectChange.objects.count(), 3)
|
|
|
|
objectchange = ObjectChange.objects.get(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=sites[0].pk
|
|
)
|
|
self.assertEqual(objectchange.changed_object, sites[0])
|
|
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_UPDATE)
|
|
self.assertEqual(objectchange.prechange_data['name'], 'Site 1')
|
|
self.assertEqual(objectchange.prechange_data['slug'], 'site-1')
|
|
self.assertEqual(objectchange.postchange_data['name'], data[0]['name'])
|
|
self.assertEqual(objectchange.postchange_data['slug'], data[0]['slug'])
|
|
|
|
def test_bulk_delete_objects(self):
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1'),
|
|
Site(name='Site 2', slug='site-2'),
|
|
Site(name='Site 3', slug='site-3'),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
data = (
|
|
{
|
|
'id': sites[0].pk,
|
|
},
|
|
{
|
|
'id': sites[1].pk,
|
|
},
|
|
{
|
|
'id': sites[2].pk,
|
|
},
|
|
)
|
|
self.assertEqual(ObjectChange.objects.count(), 0)
|
|
url = reverse('dcim-api:site-list')
|
|
self.add_permissions('dcim.delete_site')
|
|
|
|
response = self.client.delete(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
|
|
self.assertEqual(ObjectChange.objects.count(), 3)
|
|
|
|
objectchange = ObjectChange.objects.get(
|
|
changed_object_type=ContentType.objects.get_for_model(Site),
|
|
changed_object_id=sites[0].pk
|
|
)
|
|
self.assertEqual(objectchange.changed_object_type, ContentType.objects.get_for_model(Site))
|
|
self.assertEqual(objectchange.changed_object_id, sites[0].pk)
|
|
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_DELETE)
|
|
self.assertEqual(objectchange.prechange_data['name'], 'Site 1')
|
|
self.assertEqual(objectchange.prechange_data['slug'], 'site-1')
|
|
self.assertEqual(objectchange.postchange_data, None)
|
|
|
|
def test_deletion_ordering(self):
|
|
"""
|
|
Check that the cascading deletion of dependent objects is recorded in the correct order.
|
|
"""
|
|
device = create_test_device('device1')
|
|
module_bay = ModuleBay.objects.create(device=device, name='Module Bay 1')
|
|
module_type = ModuleType.objects.create(manufacturer=Manufacturer.objects.first(), model='Module Type 1')
|
|
self.add_permissions('dcim.add_module', 'dcim.add_interface', 'dcim.delete_module')
|
|
self.assertEqual(ObjectChange.objects.count(), 0) # Sanity check
|
|
|
|
# Create a new Module
|
|
data = {
|
|
'device': device.pk,
|
|
'module_bay': module_bay.pk,
|
|
'module_type': module_type.pk,
|
|
'status': ModuleStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
url = reverse('dcim-api:module-list')
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
module = device.modules.first()
|
|
|
|
# Create an Interface on the Module
|
|
data = {
|
|
'device': device.pk,
|
|
'module': module.pk,
|
|
'name': 'Interface 1',
|
|
'type': InterfaceTypeChoices.TYPE_1GE_FIXED,
|
|
}
|
|
url = reverse('dcim-api:interface-list')
|
|
response = self.client.post(url, data, format='json', **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
|
interface = device.interfaces.first()
|
|
|
|
# Delete the Module
|
|
url = reverse('dcim-api:module-detail', kwargs={'pk': module.pk})
|
|
response = self.client.delete(url, **self.header)
|
|
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
|
|
self.assertEqual(Module.objects.count(), 0)
|
|
self.assertEqual(Interface.objects.count(), 0)
|
|
|
|
# Verify the creation of the expected ObjectChange records. We should see four total records, in this order:
|
|
# 1. Module created
|
|
# 2. Interface created
|
|
# 3. Interface deleted
|
|
# 4. Module deleted
|
|
changes = ObjectChange.objects.order_by('time')
|
|
self.assertEqual(len(changes), 4)
|
|
self.assertEqual(changes[0].changed_object_type, ContentType.objects.get_for_model(Module))
|
|
self.assertEqual(changes[0].changed_object_id, module.pk)
|
|
self.assertEqual(changes[0].action, ObjectChangeActionChoices.ACTION_CREATE)
|
|
self.assertEqual(changes[1].changed_object_type, ContentType.objects.get_for_model(Interface))
|
|
self.assertEqual(changes[1].changed_object_id, interface.pk)
|
|
self.assertEqual(changes[1].action, ObjectChangeActionChoices.ACTION_CREATE)
|
|
self.assertEqual(changes[2].changed_object_type, ContentType.objects.get_for_model(Interface))
|
|
self.assertEqual(changes[2].changed_object_id, interface.pk)
|
|
self.assertEqual(changes[2].action, ObjectChangeActionChoices.ACTION_DELETE)
|
|
self.assertEqual(changes[3].changed_object_type, ContentType.objects.get_for_model(Module))
|
|
self.assertEqual(changes[3].changed_object_id, module.pk)
|
|
self.assertEqual(changes[3].action, ObjectChangeActionChoices.ACTION_DELETE)
|
|
|
|
|
|
class ChangelogPruneRetentionTest(TestCase):
|
|
"""Test suite for Changelog pruning retention settings."""
|
|
|
|
@staticmethod
|
|
def _make_oc(*, ct, obj_id, action, ts):
|
|
oc = ObjectChange.objects.create(
|
|
changed_object_type=ct,
|
|
changed_object_id=obj_id,
|
|
action=action,
|
|
user_name='test',
|
|
request_id=uuid.uuid4(),
|
|
object_repr=f'Object {obj_id}',
|
|
)
|
|
ObjectChange.objects.filter(pk=oc.pk).update(time=ts)
|
|
return oc.pk
|
|
|
|
@staticmethod
|
|
def _run_prune(*, retention_days, retain_create_last_update):
|
|
job = SystemHousekeepingJob.__new__(SystemHousekeepingJob)
|
|
job.logger = logging.getLogger('netbox.tests.changelog_prune')
|
|
|
|
with patch('core.jobs.Config') as MockConfig:
|
|
cfg = MockConfig.return_value
|
|
cfg.CHANGELOG_RETENTION = retention_days
|
|
cfg.CHANGELOG_RETAIN_CREATE_LAST_UPDATE = retain_create_last_update
|
|
job.prune_changelog()
|
|
|
|
def test_prune_retain_create_last_update_excludes_deleted_objects(self):
|
|
ct = ContentType.objects.get_for_model(Site)
|
|
|
|
retention_days = 90
|
|
now = timezone.now()
|
|
cutoff = now - timedelta(days=retention_days)
|
|
|
|
expired_old = cutoff - timedelta(days=10)
|
|
expired_newer = cutoff - timedelta(days=1)
|
|
not_expired = cutoff + timedelta(days=1)
|
|
|
|
# A) Not deleted: should keep CREATE + latest UPDATE, prune intermediate UPDATEs
|
|
a_create = self._make_oc(ct=ct, obj_id=1, action=ObjectChangeActionChoices.ACTION_CREATE, ts=expired_old)
|
|
a_update1 = self._make_oc(ct=ct, obj_id=1, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired_old)
|
|
a_update2 = self._make_oc(ct=ct, obj_id=1, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired_newer)
|
|
|
|
# B) Deleted (all expired): should keep NOTHING
|
|
b_create = self._make_oc(ct=ct, obj_id=2, action=ObjectChangeActionChoices.ACTION_CREATE, ts=expired_old)
|
|
b_update = self._make_oc(ct=ct, obj_id=2, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired_newer)
|
|
b_delete = self._make_oc(ct=ct, obj_id=2, action=ObjectChangeActionChoices.ACTION_DELETE, ts=expired_newer)
|
|
|
|
# C) Deleted but delete is not expired: create/update expired should be pruned; delete remains
|
|
c_create = self._make_oc(ct=ct, obj_id=3, action=ObjectChangeActionChoices.ACTION_CREATE, ts=expired_old)
|
|
c_update = self._make_oc(ct=ct, obj_id=3, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired_newer)
|
|
c_delete = self._make_oc(ct=ct, obj_id=3, action=ObjectChangeActionChoices.ACTION_DELETE, ts=not_expired)
|
|
|
|
self._run_prune(retention_days=retention_days, retain_create_last_update=True)
|
|
|
|
remaining = set(ObjectChange.objects.values_list('pk', flat=True))
|
|
|
|
# A) Not deleted -> create + latest update remain
|
|
self.assertIn(a_create, remaining)
|
|
self.assertIn(a_update2, remaining)
|
|
self.assertNotIn(a_update1, remaining)
|
|
|
|
# B) Deleted (all expired) -> nothing remains
|
|
self.assertNotIn(b_create, remaining)
|
|
self.assertNotIn(b_update, remaining)
|
|
self.assertNotIn(b_delete, remaining)
|
|
|
|
# C) Deleted, delete not expired -> delete remains, but create/update are pruned
|
|
self.assertNotIn(c_create, remaining)
|
|
self.assertNotIn(c_update, remaining)
|
|
self.assertIn(c_delete, remaining)
|
|
|
|
def test_prune_disabled_deletes_all_expired(self):
|
|
ct = ContentType.objects.get_for_model(Site)
|
|
|
|
retention_days = 90
|
|
now = timezone.now()
|
|
cutoff = now - timedelta(days=retention_days)
|
|
expired = cutoff - timedelta(days=1)
|
|
not_expired = cutoff + timedelta(days=1)
|
|
|
|
# expired create/update should be deleted when feature disabled
|
|
x_create = self._make_oc(ct=ct, obj_id=10, action=ObjectChangeActionChoices.ACTION_CREATE, ts=expired)
|
|
x_update = self._make_oc(ct=ct, obj_id=10, action=ObjectChangeActionChoices.ACTION_UPDATE, ts=expired)
|
|
|
|
# non-expired delete should remain regardless
|
|
y_delete = self._make_oc(ct=ct, obj_id=11, action=ObjectChangeActionChoices.ACTION_DELETE, ts=not_expired)
|
|
|
|
self._run_prune(retention_days=retention_days, retain_create_last_update=False)
|
|
|
|
remaining = set(ObjectChange.objects.values_list('pk', flat=True))
|
|
self.assertNotIn(x_create, remaining)
|
|
self.assertNotIn(x_update, remaining)
|
|
self.assertIn(y_delete, remaining)
|