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 ChangeLogViewTestCase(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 ChangeLogAPITestCase(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', 'extras.view_tag') 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', 'extras.view_tag') 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 ChangelogPruneRetentionTestCase(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)