import uuid from unittest import mock from django.contrib.contenttypes.models import ContentType from django.db import router from django.db.models import QuerySet from django.test import TestCase, override_settings from django.urls import reverse from core.choices import JobStatusChoices, ManagedFileRootPathChoices from core.models import DataSource, Job from extras.models import Script, ScriptModule from extras.validators import CustomValidator from netbox.models.deletion import ConfirmCollector, CountOnly from utilities.exceptions import AbortRequest from utilities.testing import TestCase as ViewTestCase class ScriptDeletionTestCase(TestCase): """ Regression tests for #22812: deleting a JobsMixin object (Script, ScriptModule, DataSource) with many associated Jobs must not load every Job into memory at once. """ @classmethod def setUpTestData(cls): cls.script_ct = ContentType.objects.get_for_model(Script, for_concrete_model=False) def _create_module(self): return ScriptModule.objects.create( file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=f'test_{uuid.uuid4().hex[:8]}.py', ) def _create_script(self, module=None): module = module or self._create_module() script = Script.objects.create(module=module, name=f'S{uuid.uuid4().hex[:8]}') return module, script def _add_jobs(self, obj, count, object_type=None): object_type = object_type or ContentType.objects.get_for_model(type(obj), for_concrete_model=False) Job.objects.bulk_create([ Job( object_type=object_type, object_id=obj.pk, name='testjob', status=JobStatusChoices.STATUS_COMPLETED, job_id=uuid.uuid4(), data={'output': 'x' * 50}, ) for _ in range(count) ]) def test_delete_script_deletes_all_jobs(self): _, script = self._create_script() self._add_jobs(script, 2500) self.assertEqual(script.jobs.count(), 2500) script.delete() self.assertFalse(Script.objects.filter(pk=script.pk).exists()) self.assertEqual(Job.objects.filter(object_type=self.script_ct, object_id=script.pk).count(), 0) def test_delete_script_batches_jobs(self): _, script = self._create_script() self._add_jobs(script, 5) job_delete_calls = [] original_delete = QuerySet.delete def counting_delete(qs, *args, **kwargs): if qs.model is Job: job_delete_calls.append(len(qs)) return original_delete(qs, *args, **kwargs) with mock.patch('netbox.models.features.JOB_DELETE_BATCH_SIZE', 2): with mock.patch.object(QuerySet, 'delete', counting_delete): script.delete() # 5 jobs at a batch size of 2 => three batched deletes (2, 2, 1) self.assertEqual(job_delete_calls, [2, 2, 1]) self.assertEqual(Job.objects.filter(object_type=self.script_ct, object_id=script.pk).count(), 0) def test_delete_scriptmodule_cascades_to_scripts_and_jobs(self): module, script = self._create_script() self._add_jobs(script, 100) module.delete() self.assertFalse(ScriptModule.objects.filter(pk=module.pk).exists()) self.assertFalse(Script.objects.filter(pk=script.pk).exists()) self.assertEqual(Job.objects.filter(object_type=self.script_ct, object_id=script.pk).count(), 0) def test_delete_scriptmodule_batches_child_script_jobs(self): # The reporter's actual path: a script is only removable via the UI by deleting its # ScriptModule. The module's delete must batch the child Script's jobs. module, script = self._create_script() self._add_jobs(script, 5) job_delete_calls = [] original_delete = QuerySet.delete def counting_delete(qs, *args, **kwargs): if qs.model is Job: job_delete_calls.append(len(qs)) return original_delete(qs, *args, **kwargs) with mock.patch('netbox.models.features.JOB_DELETE_BATCH_SIZE', 2): with mock.patch.object(QuerySet, 'delete', counting_delete): module.delete() # 5 child-script jobs at a batch size of 2 => three batched deletes (2, 2, 1). The module # has no jobs of its own, so JobsMixin.delete adds no further Job deletes. self.assertEqual(job_delete_calls, [2, 2, 1]) self.assertFalse(Script.objects.filter(pk=script.pk).exists()) self.assertEqual(Job.objects.filter(object_type=self.script_ct, object_id=script.pk).count(), 0) def test_delete_datasource_deletes_jobs(self): datasource = DataSource.objects.create(name='DS', type='local', source_url='/tmp/test') self._add_jobs(datasource, 100) ds_ct = ContentType.objects.get_for_model(DataSource, for_concrete_model=False) self.assertEqual(Job.objects.filter(object_type=ds_ct, object_id=datasource.pk).count(), 100) datasource.delete() self.assertFalse(DataSource.objects.filter(pk=datasource.pk).exists()) self.assertEqual(Job.objects.filter(object_type=ds_ct, object_id=datasource.pk).count(), 0) def test_soft_delete_preserves_jobs(self): _, script = self._create_script() self._add_jobs(script, 10) script.delete(soft_delete=True) script.refresh_from_db() self.assertFalse(script.is_executable) self.assertEqual(Job.objects.filter(object_type=self.script_ct, object_id=script.pk).count(), 10) @override_settings(PROTECTION_RULES={'extras.script': [CustomValidator({'name': {'eq': ''}})]}) def test_delete_rolls_back_jobs_on_parent_failure(self): # A protection rule that no real script can satisfy (name must be empty) makes the # cascade's pre_delete handler raise AbortRequest *after* JobsMixin.delete has already # batch-deleted the jobs. JobsMixin.delete wraps the batch loop and super().delete() in a # transaction, so the job deletions must roll back, leaving no orphaned partial state. # This exercises the real deletion-abort path rather than mocking Django internals. _, script = self._create_script() self._add_jobs(script, 10) with self.assertRaises(AbortRequest): script.delete() self.assertTrue(Script.objects.filter(pk=script.pk).exists()) self.assertEqual(Job.objects.filter(object_type=self.script_ct, object_id=script.pk).count(), 10) @override_settings(PROTECTION_RULES={'extras.script': [CustomValidator({'name': {'eq': ''}})]}) def test_delete_scriptmodule_rolls_back_child_jobs_on_failure(self): # Same abort path via the module: the protection rule fires when the cascade pre_deletes # the child Script, after ScriptModule.delete has already batch-deleted that script's jobs. # The transaction must roll those job deletions back, leaving no orphaned partial state. module, script = self._create_script() self._add_jobs(script, 10) with self.assertRaises(AbortRequest): module.delete() self.assertTrue(ScriptModule.objects.filter(pk=module.pk).exists()) self.assertTrue(Script.objects.filter(pk=script.pk).exists()) self.assertEqual(Job.objects.filter(object_type=self.script_ct, object_id=script.pk).count(), 10) class ConfirmCollectorTestCase(TestCase): """ #22812: the delete-confirmation page must not materialize every dependent Job. """ def _create_script_with_jobs(self, count): module = ScriptModule.objects.create( file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=f'test_{uuid.uuid4().hex[:8]}.py', ) script = Script.objects.create(module=module, name=f'S{uuid.uuid4().hex[:8]}') ct = ContentType.objects.get_for_model(Script, for_concrete_model=False) Job.objects.bulk_create([ Job(object_type=ct, object_id=script.pk, name='j', status=JobStatusChoices.STATUS_COMPLETED, job_id=uuid.uuid4(), data={'output': 'x' * 50}) for _ in range(count) ]) return script def test_confirm_collector_counts_jobs_without_instantiating(self): script = self._create_script_with_jobs(500) init_calls = [] original_init = Job.__init__ def counting_init(self, *args, **kwargs): init_calls.append(1) original_init(self, *args, **kwargs) with mock.patch.object(Job, '__init__', counting_init): collector = ConfirmCollector(using=router.db_for_write(Script)) collector.collect([script]) # No Job rows were instantiated; the relation was counted instead. self.assertEqual(len(init_calls), 0) self.assertNotIn(Job, collector.data) self.assertEqual(collector.generic_relation_counts.get(Job), 500) # The non-job cascade (the Script itself) is still collected. self.assertIn(Script, collector.data) def test_count_only_wrapper(self): # CountOnly reports its count via len() but iterates empty, so it slots into the # dependent-objects mapping as a non-expandable, non-materializing row. wrapper = CountOnly(3000) self.assertEqual(len(wrapper), 3000) self.assertEqual(list(wrapper), []) self.assertTrue(wrapper.count_only) def test_confirm_collector_omits_jobs_when_none(self): # A jobless object must not record a zero count, or the confirmation page would show a # spurious "0 jobs" row (#22812 regression). datasource = DataSource.objects.create(name='DS', type='local', source_url='/tmp/test') collector = ConfirmCollector(using=router.db_for_write(DataSource)) collector.collect([datasource]) self.assertNotIn(Job, collector.generic_relation_counts) class ObjectDeleteViewCountsTestCase(ViewTestCase): """ #22812: the delete-confirmation view must report a JobsMixin object's jobs as a count (via CountOnly) without materializing them, and _get_dependent_objects must keep returning a single dict. """ def test_get_dependent_objects_returns_count_only_for_jobs(self): from netbox.views.generic.object_views import ObjectDeleteView module = ScriptModule.objects.create( file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=f'test_{uuid.uuid4().hex[:8]}.py', ) script = Script.objects.create(module=module, name=f'S{uuid.uuid4().hex[:8]}') ct = ContentType.objects.get_for_model(Script, for_concrete_model=False) Job.objects.bulk_create([ Job( object_type=ct, object_id=script.pk, name='j', status=JobStatusChoices.STATUS_COMPLETED, job_id=uuid.uuid4(), ) for _ in range(50) ]) view = ObjectDeleteView() view.queryset = ScriptModule.objects.all() dependent_objects = view._get_dependent_objects(module) # Single dict returned (not a tuple); jobs represented as a CountOnly. self.assertIsInstance(dependent_objects, dict) self.assertIn(Job, dependent_objects) self.assertIsInstance(dependent_objects[Job], CountOnly) self.assertEqual(len(dependent_objects[Job]), 50) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_confirm_page_renders_job_count(self): module = ScriptModule.objects.create( file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=f'test_{uuid.uuid4().hex[:8]}.py', ) script = Script.objects.create(module=module, name=f'S{uuid.uuid4().hex[:8]}') ct = ContentType.objects.get_for_model(Script, for_concrete_model=False) Job.objects.bulk_create([ Job( object_type=ct, object_id=script.pk, name='j', status=JobStatusChoices.STATUS_COMPLETED, job_id=uuid.uuid4(), ) for _ in range(50) ]) # ScriptModule is a proxy over core.ManagedFile, so the delete view requires the # concrete model's permission (core.delete_managedfile), not extras.delete_scriptmodule. self.add_permissions('core.delete_managedfile') url = reverse('extras:scriptmodule_delete', kwargs={'pk': module.pk}) response = self.client.get(url) self.assertEqual(response.status_code, 200) # Assert on the rendered context, not brittle HTML substrings: Job is present in # dependent_objects as a CountOnly reporting the true count, so the confirmation page # renders it as a summarized (non-expandable) row without materializing 50 Job rows. dependent_objects = response.context['dependent_objects'] self.assertIn(Job, dependent_objects) self.assertIsInstance(dependent_objects[Job], CountOnly) self.assertEqual(len(dependent_objects[Job]), 50) self.assertTrue(dependent_objects[Job].count_only) def test_get_dependent_objects_omits_jobs_when_none(self): from netbox.views.generic.object_views import ObjectDeleteView # A module with no jobs must not produce a CountOnly(0) entry (#22812 regression). module = ScriptModule.objects.create( file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=f'test_{uuid.uuid4().hex[:8]}.py', ) view = ObjectDeleteView() view.queryset = ScriptModule.objects.all() dependent_objects = view._get_dependent_objects(module) self.assertNotIn(Job, dependent_objects)