fix(extras): Honor Script defaults when triggered by Event Rules

Scripts triggered by Event Rules now respect notifications_default and
job_timeout from script Meta class. Updates documentation to clarify
this behavior and adds regression test coverage.

Fixes #22852

docs/customization/custom-scripts.md

@@ -108,7 +108,7 @@ class MyScript(Script):
 
 ### `commit_default`
 
-The checkbox to commit database changes when executing a script is checked by default. Set `commit_default` to False under the script's Meta class to leave this option unchecked by default.
+The checkbox to commit database changes when executing a script is checked by default. Set `commit_default` to False under the script's Meta class to leave this option unchecked by default. This setting controls only the initial state of the execution form.
 
 ```python
 commit_default = False
@@ -120,7 +120,9 @@ By default, a script can be scheduled for execution at a later time. Setting `sc
 
 ### `notifications_default`
 
-By default, a notification is generated for the requesting user each time a script finishes running. This attribute sets the initial value for the notifications field when running a script. Valid values are `always` (default), `on_failure`, and `never`.
+By default, a notification is generated for the user associated with the script's job each time the script finishes running. This attribute sets the initial value for the notifications field when running a script. Valid values are `always` (default), `on_failure`, and `never`.
+
+Scripts run from an event rule or the `runscript` management command use this value as their notification policy. For an event rule, the notification goes to the user associated with the triggering event, if there is one.
 
 ```python
 notifications_default = 'on_failure'
@@ -134,7 +136,7 @@ notifications_default = 'on_failure'
 
 ### `job_timeout`
 
-Set the maximum allowed runtime for the script. If not set, `RQ_DEFAULT_TIMEOUT` will be used.
+Set the maximum allowed runtime for the script. If not set, `RQ_DEFAULT_TIMEOUT` will be used. Scripts run from an event rule use this value as their execution timeout.
 
 ## Accessing Request Data
 

netbox/extras/events.py

@@ -253,6 +253,8 @@ def process_event_rules(event_rules, object_type, event):
                 'name': script.name,
                 'user': event['user'],
                 'data': event_data,
+                'notifications': script.notifications_default,
+                'job_timeout': script.job_timeout,
             }
             if 'snapshots' in event:
                 params['snapshots'] = event['snapshots']

netbox/extras/management/commands/runscript.py

@@ -85,6 +85,7 @@ class Command(BaseCommand):
         form.cleaned_data.pop('_schedule_at')
         form.cleaned_data.pop('_interval')
         form.cleaned_data.pop('_commit')
+        notifications = form.cleaned_data.pop('_notifications')
 
         # Execute the script.
         job = ScriptJob.enqueue(
@@ -92,6 +93,7 @@ class Command(BaseCommand):
             user=user,
             immediate=True,
             data=form.cleaned_data,
+            notifications=notifications,
             request=NetBoxFakeRequest({
                 'META': {},
                 'COOKIES': {},

netbox/extras/tests/test_event_rules.py

@@ -14,14 +14,14 @@ from PIL import Image
 from requests import Session
 from rest_framework import status
 
-from core.choices import ManagedFileRootPathChoices
+from core.choices import JobNotificationChoices, ManagedFileRootPathChoices
 from core.events import *
 from core.models import Job, ObjectType
 from dcim.choices import SiteStatusChoices
 from dcim.models import DeviceType, Interface, Manufacturer, Site
 from extras.choices import EventRuleActionChoices
 from extras.events import enqueue_event, flush_events, serialize_for_event
-from extras.models import EventRule, Script, ScriptModule, Tag, Webhook
+from extras.models import EventRule, Notification, Script, ScriptModule, Tag, Webhook
 from extras.scripts import Script as ScriptBase
 from extras.signals import process_job_end_event_rules
 from extras.webhooks import generate_signature, send_webhook
@@ -756,6 +756,79 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(script_job.status, "completed")
         self.assertEqual(script_job.data.get('output', ''), "finished successfully")
 
+    @tag('regression')  # Issue #22852
+    def test_eventrule_script_action_honors_script_defaults(self):
+        """A script run from an event rule uses the notification policy and job timeout from its Meta class."""
+        class DummyScript(ScriptBase):
+            class Meta:
+                name = 'Dummy Defaults Script'
+                notifications_default = JobNotificationChoices.NOTIFICATION_ON_FAILURE
+                job_timeout = 600
+
+            def run(self, data, commit=True):
+                return 'finished successfully'
+
+        dummy_script = DummyScript()
+
+        with patch.object(ScriptModule, 'sync_classes'):
+            module = ScriptModule.objects.create(
+                file_root=ManagedFileRootPathChoices.SCRIPTS,
+                file_path='dummy_defaults_script.py',
+            )
+        script = Script.objects.create(
+            module=module,
+            name=dummy_script.name,
+            is_executable=True,
+        )
+
+        event_rule = EventRule.objects.create(
+            name='Test Script Defaults Event Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=EventRuleActionChoices.SCRIPT,
+            action_object_type=ObjectType.objects.get_for_model(Script),
+            action_object_id=script.pk,
+        )
+        event_rule.object_types.set([ObjectType.objects.get_for_model(DeviceType)])
+
+        manufacturer = Manufacturer.objects.create(name='Test Manufacturer', slug='test-manufacturer')
+        self.add_permissions('dcim.add_devicetype')
+
+        with patch.object(Script, 'python_class') as mock:
+            mock.return_value = dummy_script
+            with self.captureOnCommitCallbacks(execute=True):
+                response = self.client.post(
+                    reverse('dcim-api:devicetype-list'),
+                    {
+                        'manufacturer': manufacturer.pk,
+                        'model': 'Test DeviceType',
+                        'slug': 'test-devicetype',
+                    },
+                    format='json',
+                    **self.header,
+                )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+
+            self.assertEqual(self.queue.count, 1)
+            self.assertEqual(self.queue.jobs[0].timeout, 600)
+            script_job = Job.objects.get(name=dummy_script.name)
+            self.assertEqual(script_job.notifications, JobNotificationChoices.NOTIFICATION_ON_FAILURE)
+
+            # silence rqworker (cleaner output) and trigger job execution
+            rq_logger = logging.getLogger('rq.worker')
+            self.addCleanup(rq_logger.setLevel, rq_logger.level)
+            rq_logger.setLevel(logging.ERROR)
+            self.run_rq_jobs('default')
+
+        script_job.refresh_from_db()
+        self.assertEqual(script_job.status, "completed")
+        self.assertFalse(
+            Notification.objects.filter(
+                user=self.user,
+                object_type=ObjectType.objects.get_for_model(Job),
+                object_id=script_job.pk,
+            ).exists()
+        )
+
     @tag('regression')
     def test_eventrule_webhook_action_with_object_image_files(self):
         """

netbox/extras/tests/test_management_commands.py

@@ -7,11 +7,13 @@ from django.core.management import call_command
 from django.core.management.base import CommandError
 from django.test import TestCase
 
+from core.choices import JobNotificationChoices
 from dcim.choices import InterfaceTypeChoices
 from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site
 from extras.management.commands import renaturalize, webhook_receiver
 from extras.management.commands.webhook_receiver import WebhookHandler
 from extras.models import ImageAttachment
+from extras.scripts import Script, StringVar
 from extras.tests.test_models import OverwriteStyleMemoryStorage, UnreadableSizeMemoryStorage
 from users.models import User
 from utilities.fields import NaturalOrderingField
@@ -255,20 +257,14 @@ class RunScriptTestCase(TestCase):
         )
 
     def test_enqueues_script_job(self):
-        class TestScript:
-            full_name = 'test.Script'
+        class TestScript(Script):
+            value = StringVar()
 
-            def as_form(self, data, files):
-                form = MagicMock()
-                form.is_valid.return_value = True
-                form.cleaned_data = {
-                    '_schedule_at': None,
-                    '_interval': None,
-                    '_commit': None,
-                    'name': data['name'],
-                }
-                form.errors.get_json_data.return_value = {}
-                return form
+            class Meta:
+                notifications_default = JobNotificationChoices.NOTIFICATION_ON_FAILURE
+
+            def run(self, data, commit):
+                return None
 
         script_obj = SimpleNamespace(python_class=TestScript)
         job = SimpleNamespace(duration='0 seconds')
@@ -288,7 +284,7 @@ class RunScriptTestCase(TestCase):
                 'runscript',
                 'test.Script',
                 user='admin',
-                data='{"name": "test"}',
+                data='{"value": "test"}',
                 stdout=StringIO(),
             )
 
@@ -298,8 +294,9 @@ class RunScriptTestCase(TestCase):
         self.assertEqual(kwargs['instance'], script_obj)
         self.assertEqual(kwargs['user'], self.user)
         self.assertTrue(kwargs['immediate'])
-        self.assertEqual(kwargs['data'], {'name': 'test'})
+        self.assertEqual(kwargs['data'], {'value': 'test'})
         self.assertFalse(kwargs['commit'])
+        self.assertEqual(kwargs['notifications'], JobNotificationChoices.NOTIFICATION_ON_FAILURE)
 
     def test_invalid_script_data_raises_error_without_enqueueing_job(self):
         class TestScript:
@@ -351,6 +348,7 @@ class RunScriptTestCase(TestCase):
                     '_schedule_at': None,
                     '_interval': None,
                     '_commit': None,
+                    '_notifications': JobNotificationChoices.NOTIFICATION_ALWAYS,
                 }
                 form.errors.get_json_data.return_value = {}
                 return form
@@ -391,6 +389,7 @@ class RunScriptTestCase(TestCase):
                     '_schedule_at': None,
                     '_interval': None,
                     '_commit': None,
+                    '_notifications': JobNotificationChoices.NOTIFICATION_ALWAYS,
                 }
                 form.errors.get_json_data.return_value = {}
                 return form