RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #22870 from netbox-community/22852-scripts-run-from-event-rules-ignore-notifications_default 

Fixes #22852: Honor Custom Script execution defaults for Event Rules and `runscript`
This commit is contained in:
bctiemann 2026-08-10 14:02:25 -04:00 committed by GitHub
parent a4dcd82606 75ceb55754
commit ae8fa4c074
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 98 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
docs/customization/custom-scripts.md
View File

@ -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

2
netbox/extras/events.py
View File

@ -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']

2
netbox/extras/management/commands/runscript.py
View File

@ -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': {},

77
netbox/extras/tests/test_event_rules.py
View File

@ -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):
"""

29
netbox/extras/tests/test_management_commands.py
View File

@ -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