diff --git a/docs/models/extras/eventrule.md b/docs/models/extras/eventrule.md index 9dca04529..b3232105c 100644 --- a/docs/models/extras/eventrule.md +++ b/docs/models/extras/eventrule.md @@ -47,6 +47,11 @@ The type of action to take when the rule triggers. This must be one of the follo * Custom script * Notification +!!! tip "Custom Action Types" + The above list includes only built-in action types. NetBox plugins can also [register their own custom action types](../../plugins/development/event-rule-actions.md). + + If the plugin providing an event rule's action type is uninstalled or disabled, the event rule is not deleted, but it is marked as unavailable and will not run. It also cannot be saved -- even to edit an unrelated field -- until either the plugin is reinstalled or the action type is changed to a currently-available one. + ### Action Data An optional dictionary of JSON data to pass when executing the rule. This can be useful to include additional context data, e.g. when transmitting a webhook. diff --git a/docs/plugins/development/event-rule-actions.md b/docs/plugins/development/event-rule-actions.md new file mode 100644 index 000000000..f795ffa02 --- /dev/null +++ b/docs/plugins/development/event-rule-actions.md @@ -0,0 +1,62 @@ +# Event Rule Actions + +[Event rules](../../models/extras/eventrule.md) dispatch to an *action* when a matching event occurs, such as sending a webhook request or running a script. Plugins can register their own action types to extend the list of actions an event rule can perform, by subclassing NetBox's `EventRuleAction` class. + +```python title="event_rules.py" +from django.utils.translation import gettext_lazy as _ +from netbox.event_rules import EventRuleAction + +from .models import Ticket + +class OpenTicketAction(EventRuleAction): + slug = 'my_plugin.open_ticket' + label = _('Open ticket') + description = _('Open a ticket in the external ticketing system') + object_model = Ticket + object_required = True + + def enqueue(self, *, event_rule, event_context, action_object, action_data): + ... +``` + +To register one or more event rule actions with NetBox, define a list named `event_rule_actions` at the end of this file: + +```python title="event_rules.py" +event_rule_actions = [OpenTicketAction] +``` + +!!! tip + The path to the list of event rule actions can be modified by setting `event_rule_actions` in the PluginConfig instance. + +A dotted namespace prefix (e.g. `my_plugin.open_ticket`) is strongly recommended for `slug` to avoid collisions with other plugins or with action types added to NetBox core in the future. + +`slug` must begin with a lowercase letter, and may contain only letters, digits, underscores, and dot-separated segments thereafter. **Hyphens are not allowed**, even though they're common in plugin/package names -- use an underscore instead, e.g. `my_plugin.open_ticket` as in the example above. `register_event_rule_action()` raises `ImproperlyConfigured` immediately for a slug outside this pattern, rather than allowing it to fail later during GraphQL schema assembly. + +`slug`/`label` are only required at registration time, not at class definition, so an intermediate base class shared by several concrete actions may leave them unset. + +!!! warning "Actions must be stateless" + Registration instantiates the class once, and that single instance serves every event rule, request, and background worker thread for the lifetime of the process. Do not stash per-event data on `self` in `enqueue()` or `validate()` -- concurrent dispatches would race over it. Everything an action needs is passed in as an argument. + +## Target Objects + +If an action operates against a specific object (e.g. a webhook targets a `Webhook` instance, and a script targets a `Script` instance), set `object_model` to the relevant model class. NetBox uses this to render the object-selection field on the event rule form and to validate the selected object's type. `object_required` defaults to `False` (matching `object_model`'s default of `None`); set it to `True` alongside `object_model` if the target object must always be selected. (Setting `object_required` *without* an `object_model` raises `ImproperlyConfigured` at registration, as it could never be satisfied.) Override `get_object_queryset()` to customize which objects are eligible for selection (e.g. to filter or further restrict the queryset). + +The object-selection field is labeled with `object_model`'s verbose name; set `object_label` to override it. + +If an action leaves `object_model` as `None`, event rules using it must not specify a target object: supplying one is rejected as a validation error rather than being silently stored. + +## Bulk Import + +To support resolving a target object from a CSV value during bulk import of event rules, override `resolve_import_object()`. Raise `django.core.exceptions.ObjectDoesNotExist` (or a subclass) if the supplied value doesn't resolve to an object. If this method is not overridden, event rules using this action type cannot be targeted at an object via bulk import. + +## Unregistered Actions + +An event rule's `action_type` is stored as a plain string, and is not validated against the set of currently-registered actions at the database level. This means an event rule can reference an action type provided by a plugin that is later uninstalled or disabled, without the row being deleted or corrupted. While its action type is unavailable: + +* The event rule is skipped during event processing (it does not raise an error, and does not prevent other event rules from being processed). +* It is displayed with an "unavailable" indicator in the UI. `action_is_available` is exposed as a read-only field via the REST API, and as a filter (`?action_is_available=false`), so affected event rules can be found in bulk. +* It cannot be saved via the UI or REST API -- even to edit an unrelated field -- until its `action_type` is changed to a currently-registered value. + +Reinstalling the plugin (and thereby re-registering the action type) automatically restores the event rule to working order, with no need to re-save it. + +::: netbox.event_rules.EventRuleAction diff --git a/docs/plugins/development/index.md b/docs/plugins/development/index.md index ec85b50c2..5e23167c4 100644 --- a/docs/plugins/development/index.md +++ b/docs/plugins/development/index.md @@ -118,6 +118,7 @@ NetBox looks for the `config` variable within a plugin's `__init__.py` to load i | `events_pipeline` | A list of handlers to add to [`EVENTS_PIPELINE`](../../configuration/miscellaneous.md#events_pipeline), identified by dotted paths | | `search_indexes` | The dotted path to the list of search index classes (default: `search.indexes`) | | `data_backends` | The dotted path to the list of data source backend classes (default: `data_backends.backends`) | +| `event_rule_actions` | The dotted path to the list of event rule action classes (default: `event_rules.event_rule_actions`) | | `template_extensions` | The dotted path to the list of template extension classes (default: `template_content.template_extensions`) | | `jinja_filters` | The dotted path to a dict of custom Jinja filter functions for use in config templates (default: `jinja_env.filters`) | | `menu` | The dotted path to a top-level navigation menu provided by the plugin (default: `navigation.menu`) | diff --git a/mkdocs.yml b/mkdocs.yml index 67a6690e3..456989c5a 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -154,6 +154,7 @@ nav: - Filters & Filter Sets: 'plugins/development/filtersets.md' - Search: 'plugins/development/search.md' - Event Types: 'plugins/development/event-types.md' + - Event Rule Actions: 'plugins/development/event-rule-actions.md' - Permissions: 'plugins/development/permissions.md' - Data Backends: 'plugins/development/data-backends.md' - Webhooks: 'plugins/development/webhooks.md' diff --git a/netbox/extras/api/serializers_/events.py b/netbox/extras/api/serializers_/events.py index 29b8d3e5b..697eb50dd 100644 --- a/netbox/extras/api/serializers_/events.py +++ b/netbox/extras/api/serializers_/events.py @@ -1,9 +1,12 @@ +from rest_framework import serializers + from core.models import ObjectType from extras.choices import * from extras.models import EventRule, Webhook from netbox.api.fields import ChoiceField, ContentTypeField from netbox.api.gfk_fields import GFKSerializerField from netbox.api.serializers import NetBoxModelSerializer +from netbox.event_rules import get_event_rule_action_choices from users.api.serializers_.mixins import OwnerMixin __all__ = ( @@ -21,21 +24,34 @@ class EventRuleSerializer(OwnerMixin, NetBoxModelSerializer): queryset=ObjectType.objects.with_feature('event_rules'), many=True ) - action_type = ChoiceField(choices=EventRuleActionChoices) + action_type = ChoiceField(choices=[]) # Choices are set by get_fields() action_object_type = ContentTypeField( - queryset=ObjectType.objects.with_feature('event_rules'), + queryset=ObjectType.objects.all(), + required=False, + allow_null=True, ) action_object = GFKSerializerField(read_only=True) + action_is_available = serializers.BooleanField(read_only=True) class Meta: model = EventRule fields = [ 'id', 'url', 'display_url', 'display', 'object_types', 'name', 'enabled', 'event_types', 'conditions', - 'action_type', 'action_object_type', 'action_object_id', 'action_object', 'description', 'custom_fields', - 'owner', 'tags', 'created', 'last_updated', + 'action_type', 'action_object_type', 'action_object_id', 'action_object', 'action_is_available', + 'description', 'custom_fields', 'owner', 'tags', 'created', 'last_updated', ] brief_fields = ('id', 'url', 'display', 'name', 'description') + def get_fields(self): + fields = super().get_fields() + + # Rebuild action_type from the live registry on each instantiation to ensure all registered + # actions are captured as choices. + if 'action_type' in fields: + fields['action_type'] = ChoiceField(choices=get_event_rule_action_choices()) + + return fields + # # Webhooks diff --git a/netbox/extras/apps.py b/netbox/extras/apps.py index 8aad97cd2..afa5b91e7 100644 --- a/netbox/extras/apps.py +++ b/netbox/extras/apps.py @@ -5,9 +5,16 @@ class ExtrasConfig(AppConfig): name = "extras" def ready(self): + from netbox.event_rules import register_event_rule_action from netbox.models.features import register_models from . import dashboard, lookups, search, signals # noqa: F401 + from .event_rules import NotificationAction, ScriptAction, WebhookAction # Register models register_models(*self.get_models()) + + # Register core event rule actions + register_event_rule_action(WebhookAction, is_plugin_provided=False) + register_event_rule_action(ScriptAction, is_plugin_provided=False) + register_event_rule_action(NotificationAction, is_plugin_provided=False) diff --git a/netbox/extras/choices.py b/netbox/extras/choices.py index ae7e28e8a..79bd51097 100644 --- a/netbox/extras/choices.py +++ b/netbox/extras/choices.py @@ -276,18 +276,14 @@ class DashboardWidgetColorChoices(ChoiceSet): # Event Rules # -class EventRuleActionChoices(ChoiceSet): +class EventRuleActionChoices: + """ + The slugs of NetBox's built-in event rule actions. Not a ChoiceSet: the full, current set of + valid action_type values is plugin-extensible and lives in the netbox.event_rules registry, + not here -- see get_event_rule_action_choices(). Use these constants for the three built-in + actions; do not pass this class itself as a Django/DRF field's `choices=`. + """ WEBHOOK = 'webhook' SCRIPT = 'script' NOTIFICATION = 'notification' - - CHOICES = ( - Choice(WEBHOOK, _('Webhook'), description=_('Send an outgoing HTTP request to a remote endpoint')), - Choice(SCRIPT, _('Script'), description=_('Execute a custom script')), - Choice( - NOTIFICATION, - _('Notification'), - description=_('Generate a notification for one or more users or groups') - ), - ) diff --git a/netbox/extras/event_rules.py b/netbox/extras/event_rules.py new file mode 100644 index 000000000..e5ee5442e --- /dev/null +++ b/netbox/extras/event_rules.py @@ -0,0 +1,106 @@ +from django.utils import timezone +from django.utils.translation import gettext_lazy as _ +from django_rq import get_queue + +from netbox.config import get_config +from netbox.constants import RQ_QUEUE_DEFAULT +from netbox.event_rules import EventRuleAction +from utilities.request import copy_safe_request +from utilities.rqworker import get_rq_retry + +from .choices import EventRuleActionChoices +from .models import NotificationGroup, Script, Webhook + +__all__ = ( + 'NotificationAction', + 'ScriptAction', + 'WebhookAction', +) + + +class WebhookAction(EventRuleAction): + slug = EventRuleActionChoices.WEBHOOK + label = _('Webhook') + description = _('Send an outgoing HTTP request to a remote endpoint') + object_model = Webhook + object_required = True + + def enqueue(self, *, event_rule, event_context, action_object, action_data): + # Select the appropriate RQ queue + queue_name = get_config().QUEUE_MAPPINGS.get('webhook', RQ_QUEUE_DEFAULT) + rq_queue = get_queue(queue_name) + + # Compile the task parameters + params = { + 'event_rule': event_rule, + 'object_type': event_context['object_type'], + 'event_type': event_context['event_type'], + 'data': action_data, + 'snapshots': event_context.get('snapshots'), + 'timestamp': timezone.now().isoformat(), + 'retry': get_rq_retry(), + } + if 'request' in event_context: + # Exclude FILES - webhooks don't need uploaded files, + # which can cause pickle errors with Pillow. + params['request'] = copy_safe_request(event_context['request'], include_files=False) + + # Enqueue the task + rq_queue.enqueue('extras.webhooks.send_webhook', **params) + + def resolve_import_object(self, value): + return Webhook.objects.get(name=value) + + +class ScriptAction(EventRuleAction): + slug = EventRuleActionChoices.SCRIPT + label = _('Script') + description = _('Execute a custom script') + object_model = Script + object_required = True + + def enqueue(self, *, event_rule, event_context, action_object, action_data): + # Resolve the script from action parameters + script = action_object.python_class() + + # Enqueue a Job to record the script's execution + from extras.jobs import ScriptJob + + params = { + 'instance': action_object, + 'name': script.name, + 'user': event_context['user'], + 'data': action_data, + } + if 'snapshots' in event_context: + params['snapshots'] = event_context['snapshots'] + if 'request' in event_context: + params['request'] = copy_safe_request(event_context['request'], include_files=False) + + # Enqueue the job + ScriptJob.enqueue(**params) + + def resolve_import_object(self, value): + from extras.scripts import get_module_and_script + module_name, script_name = value.split('.', 1) + return get_module_and_script(module_name, script_name)[1] + + +class NotificationAction(EventRuleAction): + slug = EventRuleActionChoices.NOTIFICATION + label = _('Notification') + description = _('Generate a notification for one or more users or groups') + object_model = NotificationGroup + object_required = True + + def enqueue(self, *, event_rule, event_context, action_object, action_data): + # Bulk-create notifications for all members of the notification group + action_object.notify( + object_type=event_context['object_type'], + object_id=action_data['id'], + object_repr=action_data.get('display'), + event_type=event_context['event_type'], + ) + + def resolve_import_object(self, value): + return NotificationGroup.objects.get(name=value) diff --git a/netbox/extras/events.py b/netbox/extras/events.py index 2bc1a61e2..2c64c1d30 100644 --- a/netbox/extras/events.py +++ b/netbox/extras/events.py @@ -2,22 +2,15 @@ import logging from collections import UserDict, defaultdict from django.conf import settings -from django.utils import timezone from django.utils.module_loading import import_string from django.utils.translation import gettext as _ -from django_rq import get_queue from core.events import * from core.models import ObjectType -from netbox.config import get_config -from netbox.constants import RQ_QUEUE_DEFAULT from netbox.models.features import has_feature from utilities.api import get_serializer_for_model -from utilities.request import copy_safe_request -from utilities.rqworker import get_rq_retry from utilities.serialization import serialize_object -from .choices import EventRuleActionChoices from .models import EventRule logger = logging.getLogger('netbox.events_processor') @@ -174,6 +167,10 @@ def process_event_rules(event_rules, object_type, event): that a request is always present. """ + # Normalize object_type onto the event context so that an action's enqueue() can always read + # event_context['object_type']: job-lifecycle events pass it only as this parameter. + event['object_type'] = object_type + for event_rule in event_rules: # Evaluate event rule conditions (if any). @@ -206,67 +203,35 @@ def process_event_rules(event_rules, object_type, event): # Copy to avoid mutating the rule's stored action_data dict. event_data = {**action_data, **event['data']} - # Webhooks - if event_rule.action_type == EventRuleActionChoices.WEBHOOK: - - # Select the appropriate RQ queue - queue_name = get_config().QUEUE_MAPPINGS.get('webhook', RQ_QUEUE_DEFAULT) - rq_queue = get_queue(queue_name) - - # Compile the task parameters - params = { - 'event_rule': event_rule, - 'object_type': object_type, - 'event_type': event['event_type'], - 'data': event_data, - 'snapshots': event.get('snapshots'), - 'timestamp': timezone.now().isoformat(), - 'retry': get_rq_retry(), - } - if 'request' in event: - # Exclude FILES - webhooks don't need uploaded files, - # which can cause pickle errors with Pillow. - params['request'] = copy_safe_request(event['request'], include_files=False) - - # Enqueue the task - rq_queue.enqueue('extras.webhooks.send_webhook', **params) - - # Scripts - elif event_rule.action_type == EventRuleActionChoices.SCRIPT: - # Resolve the script from action parameters - script = event_rule.action_object.python_class() - - # Enqueue a Job to record the script's execution - from extras.jobs import ScriptJob - - params = { - 'instance': event_rule.action_object, - 'name': script.name, - 'user': event['user'], - 'data': event_data, - } - if 'snapshots' in event: - params['snapshots'] = event['snapshots'] - if 'request' in event: - params['request'] = copy_safe_request(event['request'], include_files=False) - - # Enqueue the job - ScriptJob.enqueue(**params) - - # Notification groups - elif event_rule.action_type == EventRuleActionChoices.NOTIFICATION: - # Bulk-create notifications for all members of the notification group - event_rule.action_object.notify( - object_type=object_type, - object_id=event_data['id'], - object_repr=event_data.get('display'), - event_type=event['event_type'], + action = event_rule.action_provider + if action is None: + # The plugin providing this action type may not be installed. Log and move on to the + # next rule rather than raising: one rule's unavailable action must not prevent any + # other rule in this batch from being processed. + logger.warning( + _('Skipping event rule "{rule}": action type "{action_type}" is not registered ' + '(the providing plugin may not be installed).').format( + rule=event_rule, action_type=event_rule.action_type, + ) ) + continue - else: - raise ValueError(_("Unknown action type for an event rule: {action_type}").format( - action_type=event_rule.action_type - )) + try: + action.enqueue( + event_rule=event_rule, + event_context=event, + action_object=event_rule.action_object, + action_data=event_data, + ) + except Exception: + # Isolate third-party bugs; a core action's own bugs should propagate instead. + if not action.is_plugin_provided: + raise + logger.exception( + _('Error processing event rule "{rule}" (action: {action_type})').format( + rule=event_rule, action_type=event_rule.action_type, + ) + ) def process_event_queue(events): diff --git a/netbox/extras/filtersets.py b/netbox/extras/filtersets.py index 3ee7e5167..eab172b22 100644 --- a/netbox/extras/filtersets.py +++ b/netbox/extras/filtersets.py @@ -5,6 +5,7 @@ from django.utils.translation import gettext as _ from core.models import DataSource, ObjectType from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site, SiteGroup +from netbox.event_rules import get_event_rule_action_choices, get_event_rule_action_slugs from netbox.filtersets import BaseFilterSet, ChangeLoggedModelFilterSet, NetBoxModelFilterSet, PrimaryModelFilterSet from tenancy.models import Tenant, TenantGroup from users.filterset_mixins import OwnerFilterMixin @@ -112,9 +113,13 @@ class EventRuleFilterSet(OwnerFilterMixin, NetBoxModelFilterSet): method='filter_event_type' ) action_type = django_filters.MultipleChoiceFilter( - choices=EventRuleActionChoices, + choices=get_event_rule_action_choices, distinct=False, ) + action_is_available = django_filters.BooleanFilter( + method='filter_action_is_available', + label=_('Action available'), + ) action_object_type = MultiValueContentTypeFilter() action_object_id = MultiValueNumberFilter() @@ -136,6 +141,12 @@ class EventRuleFilterSet(OwnerFilterMixin, NetBoxModelFilterSet): def filter_event_type(self, queryset, name, value): return queryset.filter(event_types__overlap=value) + def filter_action_is_available(self, queryset, name, value): + registered_slugs = get_event_rule_action_slugs() + if value: + return queryset.filter(action_type__in=registered_slugs) + return queryset.exclude(action_type__in=registered_slugs) + @register_filterset class CustomFieldFilterSet(OwnerFilterMixin, ChangeLoggedModelFilterSet): diff --git a/netbox/extras/forms/bulk_import.py b/netbox/extras/forms/bulk_import.py index a8a1eda34..81ea8e6d2 100644 --- a/netbox/extras/forms/bulk_import.py +++ b/netbox/extras/forms/bulk_import.py @@ -2,12 +2,13 @@ import re from django import forms from django.contrib.postgres.forms import SimpleArrayField -from django.core.exceptions import ObjectDoesNotExist +from django.core.exceptions import NON_FIELD_ERRORS, ObjectDoesNotExist, ValidationError from django.utils.translation import gettext_lazy as _ from core.models import DataFile, DataSource, ObjectType from extras.choices import * from extras.models import * +from netbox.event_rules import get_event_rule_action from netbox.events import get_event_type_choices from netbox.forms import NetBoxModelImportForm, OwnerCSVMixin, PrimaryModelImportForm from users.models import Group, User @@ -271,8 +272,11 @@ class EventRuleImportForm(OwnerCSVMixin, NetBoxModelImportForm): ) action_object = forms.CharField( label=_('Action object'), - required=True, - help_text=_('Webhook name or script as dotted path module.Class') + required=False, + help_text=_( + 'The target object for the action, if it requires one. The expected format depends on the action type ' + '(e.g. a webhook or notification group name, or a script as dotted path module.Class).' + ) ) class Meta: @@ -287,24 +291,58 @@ class EventRuleImportForm(OwnerCSVMixin, NetBoxModelImportForm): action_object = self.cleaned_data.get('action_object') action_type = self.cleaned_data.get('action_type') - if action_object and action_type: - # Webhook - if action_type == EventRuleActionChoices.WEBHOOK: - try: - webhook = Webhook.objects.get(name=action_object) - except Webhook.DoesNotExist: - raise forms.ValidationError(_("Webhook {name} not found").format(name=action_object)) - self.instance.action_object = webhook - # Script - elif action_type == EventRuleActionChoices.SCRIPT: - from extras.scripts import get_module_and_script - module_name, script_name = action_object.split('.', 1) - try: - script = get_module_and_script(module_name, script_name)[1] - except ObjectDoesNotExist: - raise forms.ValidationError(_("Script {name} not found").format(name=action_object)) - self.instance.action_object = script - self.instance.action_object_type = ObjectType.objects.get_for_model(script, for_concrete_model=False) + if not action_type: + return + + action = get_event_rule_action(action_type) + if action is None: + raise forms.ValidationError({ + 'action_type': _('"{action_type}" is not a registered action type.').format(action_type=action_type) + }) + + if not action_object: + if action.object_required: + raise forms.ValidationError({ + 'action_object': _("This action type requires a target object."), + }) + # Clear any action_object this instance previously had (relevant for a CSV row that + # updates an existing rule, matched by id, to a now-object-less action_type). + self.instance.action_object_type = None + self.instance.action_object_id = None + return + + if action.object_model is None: + raise forms.ValidationError({ + 'action_object': _("This action type does not operate against a target object."), + }) + + try: + obj = action.resolve_import_object(action_object) + except ObjectDoesNotExist: + raise forms.ValidationError({ + 'action_object': _("{name} not found").format(name=action_object) + }) + if obj is None: + raise forms.ValidationError({ + 'action_object': _("This action type does not support bulk import.") + }) + + # Assign the GFK itself (not just action_object_type/id) so EventRule.clean()'s later + # access to self.action_object hits the descriptor cache instead of a fresh SELECT -- + # for a non-proxy object_model, where the concrete and non-concrete content types match. + self.instance.action_object = obj + self.instance.action_object_type = ObjectType.objects.get_for_model(obj, for_concrete_model=False) + + def _update_errors(self, errors): + # Remap errors keyed by fields this form doesn't expose (e.g. action_object_id) to + # NON_FIELD_ERRORS; otherwise Django's add_error() raises ValueError instead of failing validation normally. + if hasattr(errors, 'error_dict'): + remapped = {} + for field, messages in errors.error_dict.items(): + key = field if field == NON_FIELD_ERRORS or field in self.fields else NON_FIELD_ERRORS + remapped.setdefault(key, []).extend(messages) + errors = ValidationError(remapped) + super()._update_errors(errors) class TagImportForm(OwnerCSVMixin, CSVModelForm): diff --git a/netbox/extras/forms/filtersets.py b/netbox/extras/forms/filtersets.py index 4cdf601f9..0ce12cab3 100644 --- a/netbox/extras/forms/filtersets.py +++ b/netbox/extras/forms/filtersets.py @@ -5,6 +5,7 @@ from core.models import DataFile, DataSource, ObjectType from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site, SiteGroup from extras.choices import * from extras.models import * +from netbox.event_rules import get_event_rule_action_choices from netbox.events import get_event_type_choices from netbox.forms import NetBoxModelFilterSetForm, PrimaryModelFilterSetForm from netbox.forms.mixins import OwnerFilterMixin, SavedFiltersMixin @@ -338,7 +339,9 @@ class EventRuleFilterForm(OwnerFilterMixin, NetBoxModelFilterSetForm): model = EventRule fieldsets = ( FieldSet('q', 'filter_id', 'tag'), - FieldSet('object_type_id', 'event_type', 'action_type', 'enabled', name=_('Attributes')), + FieldSet( + 'object_type_id', 'event_type', 'action_type', 'action_is_available', 'enabled', name=_('Attributes') + ), FieldSet('owner_group_id', 'owner_id', name=_('Ownership')), ) object_type_id = ContentTypeMultipleChoiceField( @@ -352,10 +355,19 @@ class EventRuleFilterForm(OwnerFilterMixin, NetBoxModelFilterSetForm): label=_('Event type') ) action_type = forms.ChoiceField( - choices=add_blank_choice(EventRuleActionChoices), + # Wrapped in a callable so the registry is read on each access, rather than frozen at the + # time this module is first imported (see EventRule.action_type). + choices=lambda: add_blank_choice(get_event_rule_action_choices()), required=False, label=_('Action type') ) + action_is_available = forms.NullBooleanField( + label=_('Action available'), + required=False, + widget=forms.Select( + choices=BOOLEAN_WITH_BLANK_CHOICES + ) + ) enabled = forms.NullBooleanField( label=_('Enabled'), required=False, diff --git a/netbox/extras/forms/model_forms.py b/netbox/extras/forms/model_forms.py index b6d23ecaa..41e00f9ce 100644 --- a/netbox/extras/forms/model_forms.py +++ b/netbox/extras/forms/model_forms.py @@ -12,6 +12,7 @@ from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site from extras.choices import * from extras.constants import IMAGE_ATTACHMENT_IMAGE_FORMATS from extras.models import * +from netbox.event_rules import get_event_rule_action, get_event_rule_action_choices from netbox.events import get_event_type_choices from netbox.forms import NetBoxModelForm, PrimaryModelForm from netbox.forms.mixins import ChangelogMessageMixin, OwnerMixin @@ -601,8 +602,9 @@ class WebhookForm(OwnerMixin, NetBoxModelForm): class EventRuleForm(OwnerMixin, NetBoxModelForm): action_type = ChoiceField( label=_('Action type'), - choices=EventRuleActionChoices, + choices=get_event_rule_action_choices, initial=EventRuleActionChoices.WEBHOOK, + widget=HTMXSelect(hx_target_id='event-rule-action'), ) object_types = ContentTypeMultipleChoiceField( label=_('Object types'), @@ -640,45 +642,31 @@ class EventRuleForm(OwnerMixin, NetBoxModelForm): ) widgets = { 'conditions': forms.Textarea(attrs={'class': 'font-monospace'}), - 'action_type': HTMXSelect(hx_target_id='event-rule-action'), 'action_object_type': forms.HiddenInput, 'action_object_id': forms.HiddenInput, } - def init_script_choice(self): - initial = None - if self.instance.action_type == EventRuleActionChoices.SCRIPT: - script_id = get_field_value(self, 'action_object_id') - initial = Script.objects.get(pk=script_id) if script_id else None - self.fields['action_choice'] = DynamicModelChoiceField( - label=_('Script'), - queryset=Script.objects.all(), - required=True, - initial=initial - ) + def init_action_choice(self): + action_type = get_field_value(self, 'action_type') + action = get_event_rule_action(action_type) - def init_webhook_choice(self): - initial = None - if self.instance.action_type == EventRuleActionChoices.WEBHOOK: - webhook_id = get_field_value(self, 'action_object_id') - initial = Webhook.objects.get(pk=webhook_id) if webhook_id else None - self.fields['action_choice'] = DynamicModelChoiceField( - label=_('Webhook'), - queryset=Webhook.objects.all(), - required=True, - initial=initial - ) + if action is None or action.object_model is None: + # Either an unregistered action_type (e.g. the providing plugin is not installed), or + # an action that doesn't operate on a target object at all -- no object picker needed. + self.fields.pop('action_choice', None) + return - def init_notificationgroup_choice(self): initial = None - if self.instance.action_type == EventRuleActionChoices.NOTIFICATION: - notificationgroup_id = get_field_value(self, 'action_object_id') - initial = NotificationGroup.objects.get(pk=notificationgroup_id) if notificationgroup_id else None + if self.instance.action_type == action_type: + object_id = get_field_value(self, 'action_object_id') + if object_id: + initial = action.get_object_queryset().filter(pk=object_id).first() + self.fields['action_choice'] = DynamicModelChoiceField( - label=_('Notification group'), - queryset=NotificationGroup.objects.all(), - required=True, - initial=initial + label=action.get_object_label(), + queryset=action.get_object_queryset(), + required=action.object_required, + initial=initial, ) def __init__(self, *args, **kwargs): @@ -686,35 +674,24 @@ class EventRuleForm(OwnerMixin, NetBoxModelForm): self.fields['action_object_type'].required = False self.fields['action_object_id'].required = False - # Determine the action type - action_type = get_field_value(self, 'action_type') - - if action_type == EventRuleActionChoices.WEBHOOK: - self.init_webhook_choice() - elif action_type == EventRuleActionChoices.SCRIPT: - self.init_script_choice() - elif action_type == EventRuleActionChoices.NOTIFICATION: - self.init_notificationgroup_choice() + self.init_action_choice() def clean(self): super().clean() + action = get_event_rule_action(self.cleaned_data.get('action_type')) action_choice = self.cleaned_data.get('action_choice') - # Webhook - if self.cleaned_data.get('action_type') == EventRuleActionChoices.WEBHOOK: - self.cleaned_data['action_object_type'] = ObjectType.objects.get_for_model(action_choice) - self.cleaned_data['action_object_id'] = action_choice.id - # Script - elif self.cleaned_data.get('action_type') == EventRuleActionChoices.SCRIPT: + + if action and action.object_model and action_choice: self.cleaned_data['action_object_type'] = ObjectType.objects.get_for_model( - Script, - for_concrete_model=False + action_choice, for_concrete_model=False ) - self.cleaned_data['action_object_id'] = action_choice.id - # Notification - elif self.cleaned_data.get('action_type') == EventRuleActionChoices.NOTIFICATION: - self.cleaned_data['action_object_type'] = ObjectType.objects.get_for_model(action_choice) - self.cleaned_data['action_object_id'] = action_choice.id + self.cleaned_data['action_object_id'] = action_choice.pk + elif action: + # A no-object action, or an optional object left unselected: store no action_object + self.cleaned_data['action_object_type'] = None + self.cleaned_data['action_object_id'] = None + # An unregistered action_type leaves the stored action_object untouched return self.cleaned_data diff --git a/netbox/extras/graphql/enums.py b/netbox/extras/graphql/enums.py index 1f95fa7be..fc9636d5f 100644 --- a/netbox/extras/graphql/enums.py +++ b/netbox/extras/graphql/enums.py @@ -1,6 +1,10 @@ +import enum + import strawberry from extras.choices import * +from netbox.event_rules import get_event_rule_action_choices +from utilities.string import enum_key __all__ = ( 'CustomFieldChoiceColorEnum', @@ -23,6 +27,10 @@ CustomFieldTypeEnum = strawberry.enum(CustomFieldTypeChoices.as_enum(prefix='typ CustomFieldUIEditableEnum = strawberry.enum(CustomFieldUIEditableChoices.as_enum()) CustomFieldUIVisibleEnum = strawberry.enum(CustomFieldUIVisibleChoices.as_enum()) CustomLinkButtonClassEnum = strawberry.enum(CustomLinkButtonClassChoices.as_enum()) -EventRuleActionEnum = strawberry.enum(EventRuleActionChoices.as_enum()) +# Built from the event_rule_actions registry, which is fully populated by the time the schema is +# assembled. Fixed for the process's lifetime, as any Strawberry enum is. +EventRuleActionEnum = strawberry.enum(enum.Enum('EventRuleActionEnum', { + enum_key(choice.value): choice.value for choice in get_event_rule_action_choices() +})) JournalEntryKindEnum = strawberry.enum(JournalEntryKindChoices.as_enum(prefix='kind')) WebhookHttpMethodEnum = strawberry.enum(WebhookHttpMethodChoices.as_enum()) diff --git a/netbox/extras/migrations/0143_event_rule_action_registry.py b/netbox/extras/migrations/0143_event_rule_action_registry.py new file mode 100644 index 000000000..f97901577 --- /dev/null +++ b/netbox/extras/migrations/0143_event_rule_action_registry.py @@ -0,0 +1,29 @@ +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('contenttypes', '0002_remove_content_type_name'), + ('extras', '0142_webhook_timeout'), + ] + + operations = [ + migrations.AlterField( + model_name='eventrule', + name='action_object_type', + field=models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name='eventrule_actions', + to='contenttypes.contenttype', + ), + ), + migrations.AlterField( + model_name='eventrule', + name='action_type', + field=models.CharField(default='webhook', max_length=100), + ), + ] diff --git a/netbox/extras/models/models.py b/netbox/extras/models/models.py index 3cf2b46f1..ea65c0c92 100644 --- a/netbox/extras/models/models.py +++ b/netbox/extras/models/models.py @@ -21,6 +21,7 @@ from extras.models.mixins import RenderTemplateMixin from extras.querysets import SharedObjectQuerySet from extras.utils import image_upload from netbox.config import get_config +from netbox.event_rules import get_event_rule_action, get_event_rule_action_choices from netbox.events import get_event_type_choices from netbox.models import ChangeLoggedModel from netbox.models.features import ( @@ -91,15 +92,19 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, OwnerMixin, TagsMixin, # Action to take action_type = models.CharField( - max_length=30, - choices=EventRuleActionChoices, + max_length=100, + # Bare callable, re-evaluated fresh on each access via Django's CallableChoiceIterator, + # so a plugin action registered after this module was first imported is still reflected. + choices=get_event_rule_action_choices, default=EventRuleActionChoices.WEBHOOK, verbose_name=_('action type') ) action_object_type = models.ForeignKey( to='contenttypes.ContentType', related_name='eventrule_actions', - on_delete=models.CASCADE + on_delete=models.CASCADE, + blank=True, + null=True, ) action_object_id = models.PositiveBigIntegerField( blank=True, @@ -134,6 +139,26 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, OwnerMixin, TagsMixin, def get_absolute_url(self): return reverse('extras:eventrule', args=[self.pk]) + @property + def action_provider(self): + """ + Return the registered EventRuleAction instance for this rule's action_type, or None if it + is not currently registered (e.g. the providing plugin is not installed). + """ + return get_event_rule_action(self.action_type) + + @property + def action_is_available(self): + return self.action_provider is not None + + def get_action_type_display(self): + if action := self.action_provider: + return action.label + return _('{slug} (unavailable)').format(slug=self.action_type) + + def get_action_type_color(self): + return None if self.action_is_available else 'red' + def clean(self): super().clean() @@ -148,6 +173,11 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, OwnerMixin, TagsMixin, if self.action_data is not None and not isinstance(self.action_data, dict): raise ValidationError({'action_data': _('Action data must be a JSON object or null.')}) + # action_type's own validity is already enforced by the field's dynamic choices= (Field. + # validate(), earlier in full_clean()); guard here only in case clean() ran standalone. + if self.action_is_available: + self.action_provider._validate(action_object=self.action_object, action_data=self.action_data) + def eval_conditions(self, data): """ Test whether the given data meets the conditions of the event rule (if any). Return True diff --git a/netbox/extras/tables/tables.py b/netbox/extras/tables/tables.py index 79f9f5a1f..8a6ccffcd 100644 --- a/netbox/extras/tables/tables.py +++ b/netbox/extras/tables/tables.py @@ -552,6 +552,19 @@ class EventRuleTable(NetBoxTable): 'pk', 'name', 'enabled', 'action_type', 'action_object', 'object_types', 'event_types', ) + def render_action_type(self, record): + # Render explicitly (rather than relying on django-tables2's built-in choices-driven + # get_FOO_display() auto-rendering) so an unavailable action type gets a red badge. + label = record.get_action_type_display() + if not record.action_is_available: + return format_html('{}', label) + return label + + def value_action_type(self, record): + # Raw value for non-HTML output (e.g. CSV/table-config export), so the badge's HTML + # markup from render_action_type() above isn't leaked into it. + return record.get_action_type_display() + class TagTable(NetBoxTable): name = tables.Column( diff --git a/netbox/extras/tests/test_api.py b/netbox/extras/tests/test_api.py index cabe117b1..67dbb527a 100644 --- a/netbox/extras/tests/test_api.py +++ b/netbox/extras/tests/test_api.py @@ -16,10 +16,13 @@ from core.choices import ManagedFileRootPathChoices from core.events import * from core.models import DataFile, DataSource, ObjectType from dcim.models import Device, DeviceRole, DeviceType, Location, Manufacturer, Rack, RackRole, Site +from extras.api.serializers import EventRuleSerializer from extras.choices import * from extras.models import * from extras.scripts import BooleanVar, IntegerVar, StringVar from extras.scripts import Script as PythonClass +from netbox.event_rules import EventRuleAction, register_event_rule_action +from netbox.registry import registry from users.constants import TOKEN_PREFIX from users.models import Group, ObjectPermission, Token, User from utilities.tables import get_table_for_model @@ -157,6 +160,129 @@ class EventRuleTestCase(APIViewTestCases.APIViewTestCase): ] +class EventRuleActionAPITestCase(APITestCase): + """ + REST API tests for EventRule's registry-driven action_type. + """ + + def test_create_event_rule_with_unregistered_action_type_fails(self): + self.add_permissions('extras.add_eventrule') + url = reverse('extras-api:eventrule-list') + data = { + 'name': 'API Bad Action Rule', + 'object_types': ['dcim.site'], + 'event_types': [OBJECT_CREATED], + 'action_type': 'this.is.not.registered', + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertIn('action_type', response.data) + + def test_update_unrelated_field_on_unavailable_action_rule_fails(self): + """PATCHing a rule with an unavailable action_type is rejected, even for an unrelated field.""" + rule = EventRule.objects.create( + name='API Unavailable Rule', + event_types=[OBJECT_CREATED], + action_type='some.plugin.not_installed', + ) + rule.object_types.set([ObjectType.objects.get_for_model(Site)]) + + self.add_permissions('extras.change_eventrule') + url = reverse('extras-api:eventrule-detail', kwargs={'pk': rule.pk}) + response = self.client.patch(url, {'enabled': False}, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertIn('action_type', response.data) + + rule.refresh_from_db() + self.assertTrue(rule.enabled) + + def test_action_is_available_exposed_via_api(self): + """action_is_available is exposed as a read-only field, so unavailable rules can be found in bulk.""" + available_rule = EventRule.objects.create( + name='API Available Rule', event_types=[OBJECT_CREATED], action_type='webhook', + ) + available_rule.object_types.set([ObjectType.objects.get_for_model(Site)]) + + unavailable_rule = EventRule.objects.create( + name='API Unavailable Flag Rule', + event_types=[OBJECT_CREATED], + action_type='some.plugin.not_installed_flag_test', + ) + unavailable_rule.object_types.set([ObjectType.objects.get_for_model(Site)]) + + self.add_permissions('extras.view_eventrule') + url = reverse('extras-api:eventrule-detail', kwargs={'pk': available_rule.pk}) + response = self.client.get(url, **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + self.assertTrue(response.data['action_is_available']) + + url = reverse('extras-api:eventrule-detail', kwargs={'pk': unavailable_rule.pk}) + response = self.client.get(url, **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + self.assertFalse(response.data['action_is_available']) + + def test_create_event_rule_with_runtime_registered_action(self): + """An action registered after this serializer's module was imported must still be a valid action_type.""" + class NoObjectAction(EventRuleAction): + slug = 'test.api_no_object_action' + label = 'API No-Object Action' + + register_event_rule_action(NoObjectAction) + self.addCleanup(registry['event_rule_actions'].pop, NoObjectAction.slug, None) + + self.add_permissions('extras.add_eventrule') + url = reverse('extras-api:eventrule-list') + data = { + 'name': 'API No-Object Rule', + 'object_types': ['dcim.site'], + 'event_types': [OBJECT_CREATED], + 'action_type': NoObjectAction.slug, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_201_CREATED) + + rule = EventRule.objects.get(pk=response.data['id']) + self.assertEqual(rule.action_type, NoObjectAction.slug) + self.assertIsNone(rule.action_object_type) + self.assertIsNone(rule.action_object_id) + + def test_create_event_rule_with_object_for_no_object_action_fails(self): + """An action with no object_model must reject a target object rather than storing it.""" + class NoObjectAction(EventRuleAction): + slug = 'test.api_no_object_action' + label = 'API No-Object Action' + + register_event_rule_action(NoObjectAction) + self.addCleanup(registry['event_rule_actions'].pop, NoObjectAction.slug, None) + + site = Site.objects.create(name='Action Object Site', slug='action-object-site') + + self.add_permissions('extras.add_eventrule') + url = reverse('extras-api:eventrule-list') + data = { + 'name': 'API Bogus Object Rule', + 'object_types': ['dcim.site'], + 'event_types': [OBJECT_CREATED], + 'action_type': NoObjectAction.slug, + 'action_object_type': 'dcim.site', + 'action_object_id': site.pk, + } + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertIn('action_object_id', response.data) + self.assertFalse(EventRule.objects.filter(name='API Bogus Object Rule').exists()) + + def test_action_object_type_field_accepts_any_content_type(self): + """action_object_type's queryset must not be restricted to the with_feature('event_rules') set.""" + field = EventRuleSerializer().fields['action_object_type'] + user_ct = ObjectType.objects.get_for_model(User) + self.assertFalse( + ObjectType.objects.with_feature('event_rules').filter(pk=user_ct.pk).exists(), + "auth.user must not support event_rules for this test to be meaningful; pick another type.", + ) + self.assertIn(user_ct, field.queryset) + + class CustomFieldTestCase(APIViewTestCases.APIViewTestCase): model = CustomField brief_fields = ['description', 'display', 'id', 'name', 'url'] diff --git a/netbox/extras/tests/test_event_rules.py b/netbox/extras/tests/test_event_rules.py index 69ea80121..743d1fc48 100644 --- a/netbox/extras/tests/test_event_rules.py +++ b/netbox/extras/tests/test_event_rules.py @@ -7,6 +7,7 @@ from unittest.mock import Mock, patch import django_rq from django.conf import settings +from django.core.exceptions import ImproperlyConfigured, ValidationError from django.http import HttpResponse from django.test import RequestFactory, TestCase, tag from django.urls import reverse @@ -20,12 +21,20 @@ 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.events import enqueue_event, flush_events, process_event_rules, serialize_for_event from extras.models import EventRule, 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 from netbox.context_managers import event_tracking +from netbox.event_rules import ( + EventRuleAction, + get_event_rule_action, + get_event_rule_action_choices, + register_event_rule_action, +) +from netbox.registry import registry +from netbox.tests.dummy_plugin.event_rules import DummyRaisingAction from utilities.testing import APITestCase, create_test_device from utilities.testing.mixins import RQQueueTestMixin @@ -882,6 +891,464 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase): self.assertEqual(job.kwargs['event_rule'], event_rule) self.assertEqual(job.kwargs['event_type'], OBJECT_UPDATED) + def test_unregistered_action_type_does_not_block_other_rules(self): + """ + An unregistered action_type must not block other EventRules for the same event. Kept on + this class, not a separate RQQueueTestMixin one, since two such classes in different + `--parallel` subsuites cross-flush each other's Redis queue. + """ + site_type = ObjectType.objects.get_for_model(Site) + webhook = Webhook.objects.create(name='Dispatch Test Webhook', payload_url='http://localhost:9000/') + webhook_type = ObjectType.objects.get_for_model(Webhook) + + good_rule = EventRule.objects.create( + name='Good Rule', + event_types=[OBJECT_CREATED], + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=webhook_type, + action_object_id=webhook.pk, + ) + good_rule.object_types.set([site_type]) + + bad_rule = EventRule.objects.create( + name='Bad Rule', + event_types=[OBJECT_CREATED], + action_type='someplugin.not_installed', + ) + bad_rule.object_types.set([site_type]) + + url = reverse('dcim-api:site-list') + self.add_permissions('dcim.add_site') + with self.assertLogs('netbox.events_processor', level='WARNING') as cm: + response = self.client.post( + url, {'name': 'Dispatch Site', 'slug': 'dispatch-site'}, format='json', **self.header + ) + self.assertHttpStatus(response, status.HTTP_201_CREATED) + self.assertTrue(any('someplugin.not_installed' in message for message in cm.output)) + + # The good rule's webhook must still have been enqueued despite the bad rule. + rule_jobs = [j for j in self.queue.jobs if j.kwargs['event_rule'] == good_rule] + self.assertEqual(len(rule_jobs), 1) + + def test_raising_enqueue_does_not_block_other_rules(self): + """A raising action registered as plugin-provided (the default) must not block other rules.""" + register_event_rule_action(DummyRaisingAction) + self.addCleanup(registry['event_rule_actions'].pop, DummyRaisingAction.slug, None) + + site_type = ObjectType.objects.get_for_model(Site) + webhook = Webhook.objects.create(name='Dispatch Test Webhook 2', payload_url='http://localhost:9000/') + webhook_type = ObjectType.objects.get_for_model(Webhook) + + good_rule = EventRule.objects.create( + name='Good Rule 2', + event_types=[OBJECT_CREATED], + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=webhook_type, + action_object_id=webhook.pk, + ) + good_rule.object_types.set([site_type]) + + raising_rule = EventRule.objects.create( + name='Raising Rule', + event_types=[OBJECT_CREATED], + action_type=DummyRaisingAction.slug, + ) + raising_rule.object_types.set([site_type]) + + url = reverse('dcim-api:site-list') + self.add_permissions('dcim.add_site') + with self.assertLogs('netbox.events_processor', level='ERROR') as cm: + response = self.client.post( + url, {'name': 'Dispatch Site 2', 'slug': 'dispatch-site-2'}, format='json', **self.header + ) + self.assertHttpStatus(response, status.HTTP_201_CREATED) + self.assertTrue(any('Raising Rule' in message for message in cm.output)) + + # The good rule's webhook must still have been enqueued despite the raising rule. + rule_jobs = [j for j in self.queue.jobs if j.kwargs['event_rule'] == good_rule] + self.assertEqual(len(rule_jobs), 1) + + def test_raising_action_registered_as_non_plugin_propagates(self): + """A raising action registered with is_plugin_provided=False (as core actions are) must propagate.""" + class RaisingCoreLikeAction(EventRuleAction): + slug = 'test.raising_core_like_action' + label = 'Raising Core-Like Action' + object_required = False + + def enqueue(self, **kwargs): + raise RuntimeError("intentional failure for test") + + register_event_rule_action(RaisingCoreLikeAction, is_plugin_provided=False) + self.addCleanup(registry['event_rule_actions'].pop, 'test.raising_core_like_action', None) + + site_type = ObjectType.objects.get_for_model(Site) + rule = EventRule.objects.create( + name='Raising Core-Like Rule', + event_types=[OBJECT_CREATED], + action_type='test.raising_core_like_action', + ) + rule.object_types.set([site_type]) + + with self.assertRaises(RuntimeError): + process_event_rules([rule], object_type=site_type, event={'data': {}, 'event_type': OBJECT_CREATED}) + + +class EventRuleActionRegistrationTestCase(TestCase): + """ + Unit tests for the EventRuleAction registry (netbox.event_rules). + """ + + def tearDown(self): + super().tearDown() + # The registry is a global dict; test-registered actions must not leak into other tests. + for slug in ('test.dummy_action', 'test.duplicate_action'): + registry['event_rule_actions'].pop(slug, None) + + def test_register_event_rule_action(self): + class DummyAction(EventRuleAction): + slug = 'test.dummy_action' + label = 'Dummy Action' + description = 'A dummy action for testing' + + register_event_rule_action(DummyAction) + + action = get_event_rule_action('test.dummy_action') + self.assertIsInstance(action, DummyAction) + + choices = {choice.value: choice.label for choice in get_event_rule_action_choices()} + self.assertEqual(choices.get('test.dummy_action'), 'Dummy Action') + + def test_register_event_rule_action_as_decorator(self): + @register_event_rule_action + class DummyAction(EventRuleAction): + slug = 'test.dummy_action' + label = 'Dummy Action' + + self.assertIsInstance(get_event_rule_action('test.dummy_action'), DummyAction) + + def test_duplicate_slug_raises(self): + class FirstAction(EventRuleAction): + slug = 'test.duplicate_action' + label = 'First' + + class SecondAction(EventRuleAction): + slug = 'test.duplicate_action' + label = 'Second' + + register_event_rule_action(FirstAction) + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(SecondAction) + + def test_slug_starting_with_digit_rejected(self): + """A slug starting with a digit would sanitize into a GraphQL-invalid enum member name.""" + class DigitSlugAction(EventRuleAction): + slug = '2fa.notify' + label = 'Digit Slug Action' + + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(DigitSlugAction) + self.assertIsNone(get_event_rule_action('2fa.notify')) + + def test_slug_with_hyphen_rejected(self): + """Hyphens are not permitted, though plugin distribution names conventionally use them.""" + class HyphenSlugAction(EventRuleAction): + slug = 'my-plugin.open_ticket' + label = 'Hyphen Slug Action' + + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(HyphenSlugAction) + self.assertIsNone(get_event_rule_action('my-plugin.open_ticket')) + + def test_slug_with_leading_underscore_rejected(self): + """A leading underscore sanitizes into a "__"-prefixed name, which GraphQL reserves for introspection.""" + class LeadingUnderscoreAction(EventRuleAction): + slug = '_internal.foo' + label = 'Leading Underscore Action' + + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(LeadingUnderscoreAction) + self.assertIsNone(get_event_rule_action('_internal.foo')) + + def test_slug_with_uppercase_rejected(self): + """Slugs must be lowercase, though plugin/class names conventionally are not.""" + class UppercaseSlugAction(EventRuleAction): + slug = 'MyPlugin.action' + label = 'Uppercase Slug Action' + + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(UppercaseSlugAction) + self.assertIsNone(get_event_rule_action('MyPlugin.action')) + + def test_slug_enum_key_collision_rejected(self): + """Two distinct slugs that sanitize to the same GraphQL enum member name must not both register.""" + class DotAction(EventRuleAction): + slug = 'test.collision_action' + label = 'Dot Action' + + class UnderscoreAction(EventRuleAction): + slug = 'test_collision_action' + label = 'Underscore Action' + + register_event_rule_action(DotAction) + self.addCleanup(registry['event_rule_actions'].pop, 'test.collision_action', None) + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(UnderscoreAction) + self.assertIsNone(get_event_rule_action('test_collision_action')) + + def test_missing_slug_raises_at_registration(self): + # Class definition itself must succeed; only registration checks slug/label. + class NoSlugAction(EventRuleAction): + label = 'No Slug' + + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(NoSlugAction) + + def test_missing_label_raises_at_registration(self): + class NoLabelAction(EventRuleAction): + slug = 'test.no_label' + + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(NoLabelAction) + + def test_intermediate_base_class_without_slug_or_label_is_definable(self): + """ + slug/label are checked at registration, not class definition, so several concrete actions + can share an intermediate base class which sets neither. + """ + class PluginActionBase(EventRuleAction): + object_required = False + + def enqueue(self, **kwargs): + pass + + class ConcreteAction(PluginActionBase): + slug = 'test.intermediate_base_concrete_action' + label = 'Concrete Action' + + register_event_rule_action(ConcreteAction) + self.addCleanup(registry['event_rule_actions'].pop, 'test.intermediate_base_concrete_action', None) + + self.assertIsInstance(get_event_rule_action('test.intermediate_base_concrete_action'), ConcreteAction) + + def test_unregistered_slug_returns_none(self): + self.assertIsNone(get_event_rule_action('this.does.not.exist')) + + def test_core_actions_are_registered(self): + """WebhookAction/ScriptAction/NotificationAction are registered at app startup.""" + core_slugs = ( + EventRuleActionChoices.WEBHOOK, EventRuleActionChoices.SCRIPT, EventRuleActionChoices.NOTIFICATION, + ) + for slug in core_slugs: + self.assertIsNotNone(get_event_rule_action(slug)) + + def test_get_object_queryset_returns_none_without_object_model(self): + action = EventRuleAction() + self.assertIsNone(action.get_object_queryset()) + + def test_internal_validate_requires_object_when_object_required(self): + action = EventRuleAction() + action.object_required = True + with self.assertRaises(ValidationError): + action._validate(action_object=None, action_data={}) + + def test_internal_validate_passes_when_object_not_required(self): + action = EventRuleAction() + action.object_required = False + # Must not raise + action._validate(action_object=None, action_data={}) + + def test_internal_validate_rejects_wrong_object_type(self): + action = EventRuleAction() + action.object_model = Webhook + action.object_required = True + site = Site(name='Not A Webhook') + with self.assertRaises(ValidationError): + action._validate(action_object=site, action_data={}) + + def test_internal_validate_rejects_object_for_action_without_object_model(self): + """An action which declares no object_model must reject a target object outright.""" + action = EventRuleAction() + with self.assertRaises(ValidationError): + action._validate(action_object=Webhook(), action_data={}) + + def test_object_required_without_object_model_rejected_at_registration(self): + """object_required with no object_model could never be satisfied, so it's caught early.""" + class ImpossibleAction(EventRuleAction): + slug = 'test.impossible_action' + label = 'Impossible Action' + object_required = True + + with self.assertRaises(ImproperlyConfigured): + register_event_rule_action(ImpossibleAction) + self.assertIsNone(get_event_rule_action('test.impossible_action')) + + def test_get_object_label_defaults_to_object_model_verbose_name(self): + """The object picker's label defaults to the model's verbose name, capitalized.""" + self.assertEqual(get_event_rule_action(EventRuleActionChoices.NOTIFICATION).get_object_label(), + 'Notification group') + self.assertEqual(get_event_rule_action(EventRuleActionChoices.WEBHOOK).get_object_label(), 'Webhook') + + def test_get_object_label_honors_explicit_override(self): + class LabeledAction(EventRuleAction): + object_model = Webhook + object_label = 'Destination' + + self.assertEqual(LabeledAction().get_object_label(), 'Destination') + + def test_get_object_label_is_none_without_object_model(self): + self.assertIsNone(EventRuleAction().get_object_label()) + + def test_validate_is_noop_by_default(self): + action = EventRuleAction() + # Must not raise + action.validate(action_object=None, action_data={}) + + def test_validate_override_does_not_need_super(self): + """A subclass overriding validate() gets the base object_required check for free, no super() needed.""" + class CustomValidatingAction(EventRuleAction): + slug = 'test.custom_validating_action' + label = 'Custom Validating Action' + object_model = Webhook + object_required = True + + def validate(self, *, action_object, action_data): + if action_data.get('bad'): + raise ValidationError({'action_data': 'bad action_data for test'}) + + action = CustomValidatingAction() + + # The subclass's own check fires + with self.assertRaises(ValidationError): + action._validate(action_object=Webhook(), action_data={'bad': True}) + + # ...as does the base object_required check + with self.assertRaises(ValidationError): + action._validate(action_object=None, action_data={}) + + action._validate(action_object=Webhook(), action_data={}) # must not raise + + def test_enqueue_not_implemented_by_default(self): + action = EventRuleAction() + with self.assertRaises(NotImplementedError): + action.enqueue(event_rule=None, event_context={}, action_object=None, action_data={}) + + def test_is_plugin_provided_defaults_true_before_registration(self): + """is_plugin_provided is True on an instance that never goes through registration.""" + self.assertTrue(EventRuleAction().is_plugin_provided) + + +class EventRuleActionAvailabilityTestCase(TestCase): + """ + An EventRule with an unregistered action_type must remain loadable, skip gracefully during + processing, and display as "unavailable" -- but reject full_clean() until action_type changes. + """ + + @classmethod + def setUpTestData(cls): + site_type = ObjectType.objects.get_for_model(Site) + webhook = Webhook.objects.create(name='Availability Test Webhook', payload_url='http://localhost:9000/') + webhook_type = ObjectType.objects.get_for_model(Webhook) + + cls.healthy_rule = EventRule.objects.create( + name='Healthy Rule', + event_types=[OBJECT_CREATED], + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=webhook_type, + action_object_id=webhook.pk, + ) + cls.healthy_rule.object_types.set([site_type]) + + # .objects.create() calls save(), not full_clean(), so an unregistered action_type can be + # persisted directly, matching the state of a row whose providing plugin was uninstalled. + cls.unavailable_rule = EventRule.objects.create( + name='Unavailable Rule', + event_types=[OBJECT_CREATED], + action_type='someplugin.not_installed', + ) + cls.unavailable_rule.object_types.set([site_type]) + + def test_action_is_available_true_for_registered_action(self): + self.assertTrue(self.healthy_rule.action_is_available) + self.assertIsNotNone(self.healthy_rule.action_provider) + + def test_action_is_available_false_for_unregistered_action(self): + self.assertFalse(self.unavailable_rule.action_is_available) + self.assertIsNone(self.unavailable_rule.action_provider) + + def test_get_action_type_display_for_registered_action(self): + self.assertEqual(self.healthy_rule.get_action_type_display(), 'Webhook') + + def test_get_action_type_display_for_unregistered_action(self): + self.assertEqual( + self.unavailable_rule.get_action_type_display(), + 'someplugin.not_installed (unavailable)', + ) + + def test_get_action_type_color_for_registered_action(self): + self.assertIsNone(self.healthy_rule.get_action_type_color()) + + def test_get_action_type_color_for_unregistered_action(self): + self.assertEqual(self.unavailable_rule.get_action_type_color(), 'red') + + def test_clean_rejects_unchanged_unavailable_action_type(self): + """A persisted-but-unavailable action_type is rejected by full_clean() even when left unchanged.""" + rule = EventRule.objects.get(pk=self.unavailable_rule.pk) + rule.enabled = False + with self.assertRaises(ValidationError): + rule.full_clean() + + def test_clean_rejects_new_row_with_unregistered_action_type(self): + rule = EventRule( + name='New Unregistered Rule', + event_types=[OBJECT_CREATED], + action_type='someplugin.also_not_installed', + ) + with self.assertRaises(ValidationError): + rule.full_clean() + + def test_clean_rejects_changing_to_unregistered_action_type(self): + rule = EventRule.objects.get(pk=self.healthy_rule.pk) + rule.action_type = 'someplugin.newly_unregistered' + with self.assertRaises(ValidationError): + rule.full_clean() + + def test_clean_accepts_registered_action_with_valid_object(self): + rule = EventRule.objects.get(pk=self.healthy_rule.pk) + rule.full_clean() # must not raise + + +class EventRuleNoObjectActionTestCase(TestCase): + """ + Model-layer tests for an EventRuleAction which declares object_model=None (no target object). + """ + + def tearDown(self): + super().tearDown() + registry['event_rule_actions'].pop('test.model_no_object_action', None) + + def test_full_clean_and_save_with_no_object_action(self): + class NoObjectAction(EventRuleAction): + slug = 'test.model_no_object_action' + label = 'Model No-Object Action' + object_required = False + + register_event_rule_action(NoObjectAction) + + site_type = ObjectType.objects.get_for_model(Site) + rule = EventRule( + name='Model No-Object Rule', + event_types=[OBJECT_CREATED], + action_type='test.model_no_object_action', + ) + rule.full_clean() # must not raise: no action_object required or supplied + rule.save() + rule.object_types.set([site_type]) + + rule.refresh_from_db() + self.assertIsNone(rule.action_object_type) + self.assertIsNone(rule.action_object_id) + self.assertIsNone(rule.action_object) + class WebhookRenderHeadersTest(TestCase): diff --git a/netbox/extras/tests/test_filtersets.py b/netbox/extras/tests/test_filtersets.py index 3e002864e..25f4abab6 100644 --- a/netbox/extras/tests/test_filtersets.py +++ b/netbox/extras/tests/test_filtersets.py @@ -396,6 +396,49 @@ class EventRuleTestCase(TestCase, BaseFilterSetTests): params = {'action_type': [EventRuleActionChoices.SCRIPT]} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_action_is_available(self): + unavailable_rule = EventRule.objects.create( + name='Unavailable Filterset Rule', + event_types=[OBJECT_CREATED], + action_type='someplugin.not_installed_filterset_test', + ) + unavailable_rule.object_types.set([ObjectType.objects.get_for_model(Site)]) + + params = {'action_is_available': True} + qs = self.filterset(params, EventRule.objects.all()).qs + self.assertEqual(qs.count(), 5) + self.assertNotIn(unavailable_rule, qs) + + params = {'action_is_available': False} + qs = self.filterset(params, EventRule.objects.all()).qs + self.assertEqual(qs.count(), 1) + self.assertEqual(qs.first(), unavailable_rule) + + def test_action_type_registered_plugin_style_slug(self): + """A plugin-registered action slug is a valid action_type filter value, not just the core actions.""" + from netbox.event_rules import EventRuleAction, register_event_rule_action + from netbox.registry import registry + + class FilterTestAction(EventRuleAction): + slug = 'test.filterset_registered_action' + label = 'Filterset Test Action' + object_required = False + + register_event_rule_action(FilterTestAction) + self.addCleanup(registry['event_rule_actions'].pop, FilterTestAction.slug, None) + + rule = EventRule.objects.create( + name='Filterset Registered Action Rule', + event_types=[OBJECT_CREATED], + action_type=FilterTestAction.slug, + ) + rule.object_types.set([ObjectType.objects.get_for_model(Site)]) + + params = {'action_type': [FilterTestAction.slug]} + qs = self.filterset(params, EventRule.objects.all()).qs + self.assertEqual(qs.count(), 1) + self.assertEqual(qs.first(), rule) + def test_enabled(self): params = {'enabled': True} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) diff --git a/netbox/extras/tests/test_forms.py b/netbox/extras/tests/test_forms.py index 909d72801..60744c5bb 100644 --- a/netbox/extras/tests/test_forms.py +++ b/netbox/extras/tests/test_forms.py @@ -1,19 +1,25 @@ import tempfile from pathlib import Path -from django.core.exceptions import NON_FIELD_ERRORS +from django.core.exceptions import NON_FIELD_ERRORS, ValidationError from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase from core.choices import ManagedFileRootPathChoices +from core.events import OBJECT_CREATED from core.models import DataSource, ObjectType from dcim.forms import SiteForm from dcim.models import Site -from extras.choices import CustomFieldTypeChoices +from extras.choices import CustomFieldTypeChoices, EventRuleActionChoices from extras.forms import SavedFilterForm, TableConfigBulkEditForm, TableConfigForm -from extras.forms.model_forms import CustomFieldChoiceSetForm +from extras.forms.bulk_import import EventRuleImportForm +from extras.forms.filtersets import EventRuleFilterForm +from extras.forms.model_forms import CustomFieldChoiceSetForm, EventRuleForm from extras.forms.scripts import ScriptFileForm -from extras.models import CustomField, CustomFieldChoiceSet, ScriptModule +from extras.models import CustomField, CustomFieldChoiceSet, EventRule, NotificationGroup, Script, ScriptModule, Webhook +from netbox.event_rules import EventRuleAction, register_event_rule_action +from netbox.registry import registry +from utilities.forms.widgets import HTMXSelect class CustomFieldModelFormTestCase(TestCase): @@ -337,3 +343,353 @@ class TableConfigFormTestCase(TestCase): form = TableConfigBulkEditForm() self.assertIn('changelog_message', form.fields) self.assertIn('changelog_message', form.meta_fields) + + +class EventRuleFormTestCase(TestCase): + """ + EventRuleForm's action_choice field is built dynamically from the EventRuleAction registry, + for both core actions and those registered by a plugin. + """ + + def tearDown(self): + super().tearDown() + registry['event_rule_actions'].pop('test.form_no_object_action', None) + + def test_action_type_widget_is_htmx_select(self): + """ + action_choice refreshes via HTMX when action_type changes. The widget must be set on the + field itself: Meta.widgets applies only to fields the ModelForm generates from the model, + and action_type is declared explicitly. + """ + form = EventRuleForm() + widget = form.fields['action_type'].widget + self.assertIsInstance(widget, HTMXSelect) + self.assertEqual(widget.attrs.get('hx-target'), '#event-rule-action') + + def test_action_choice_field_for_webhook(self): + webhook = Webhook.objects.create(name='Form Test Webhook', payload_url='http://localhost:9000/') + form = EventRuleForm(data={'action_type': EventRuleActionChoices.WEBHOOK}) + self.assertIn('action_choice', form.fields) + self.assertIn(webhook, form.fields['action_choice'].queryset) + + def test_action_choice_field_for_script(self): + form = EventRuleForm(data={'action_type': EventRuleActionChoices.SCRIPT}) + self.assertIn('action_choice', form.fields) + self.assertEqual(form.fields['action_choice'].queryset.model, Script) + + def test_action_choice_field_for_notification(self): + form = EventRuleForm(data={'action_type': EventRuleActionChoices.NOTIFICATION}) + self.assertIn('action_choice', form.fields) + self.assertEqual(form.fields['action_choice'].queryset.model, NotificationGroup) + + def test_action_choice_field_labels(self): + """The object picker is labeled for the object being selected, not for the action itself.""" + for action_type, label in ( + (EventRuleActionChoices.WEBHOOK, 'Webhook'), + (EventRuleActionChoices.SCRIPT, 'Script'), + (EventRuleActionChoices.NOTIFICATION, 'Notification group'), + ): + form = EventRuleForm(data={'action_type': action_type}) + self.assertEqual(form.fields['action_choice'].label, label) + + def test_action_choice_field_honors_object_label(self): + class LabeledObjectAction(EventRuleAction): + slug = 'test.form_labeled_object_action' + label = 'Form Labeled Object Action' + object_model = Webhook + object_label = 'Destination' + + register_event_rule_action(LabeledObjectAction) + self.addCleanup(registry['event_rule_actions'].pop, LabeledObjectAction.slug, None) + + form = EventRuleForm(data={'action_type': LabeledObjectAction.slug}) + self.assertEqual(form.fields['action_choice'].label, 'Destination') + + def test_action_choice_field_omitted_for_registered_no_object_action(self): + class NoObjectAction(EventRuleAction): + slug = 'test.form_no_object_action' + label = 'Form No-Object Action' + object_required = False + + register_event_rule_action(NoObjectAction) + + form = EventRuleForm(data={'action_type': 'test.form_no_object_action'}) + self.assertNotIn('action_choice', form.fields) + + def test_action_choice_field_falls_back_to_initial_for_unregistered_action(self): + """ + get_field_value() falls back to the field's own initial (webhook) for an unregistered + action_type, so init_action_choice() still builds a usable picker. + """ + form = EventRuleForm(data={'action_type': 'not.a.registered.action'}) + self.assertIn('action_choice', form.fields) + self.assertEqual(form.fields['action_choice'].queryset.model, Webhook) + + def test_submit_and_save_with_registered_no_object_action(self): + """A runtime-registered action can be submitted and saved end-to-end through the form.""" + class NoObjectAction(EventRuleAction): + slug = 'test.form_no_object_action' + label = 'Form No-Object Action' + object_required = False + + def enqueue(self, **kwargs): + pass + + register_event_rule_action(NoObjectAction) + + object_type = ObjectType.objects.get_for_model(Site) + form = EventRuleForm(data={ + 'name': 'Form No-Object Rule', + 'object_types': [object_type.pk], + 'event_types': [OBJECT_CREATED], + 'action_type': 'test.form_no_object_action', + }) + self.assertTrue(form.is_valid(), form.errors) + rule = form.save() + self.assertIsNone(rule.action_object_type) + self.assertIsNone(rule.action_object_id) + + def test_submit_and_save_webhook_action(self): + """The generalized form still saves a core Webhook action correctly.""" + webhook = Webhook.objects.create(name='Form Submit Webhook', payload_url='http://localhost:9000/') + object_type = ObjectType.objects.get_for_model(Site) + form = EventRuleForm(data={ + 'name': 'Form Webhook Rule', + 'object_types': [object_type.pk], + 'event_types': [OBJECT_CREATED], + 'action_type': EventRuleActionChoices.WEBHOOK, + 'action_choice': webhook.pk, + }) + self.assertTrue(form.is_valid(), form.errors) + rule = form.save() + self.assertEqual(rule.action_object, webhook) + + def test_switching_to_optional_object_action_clears_stale_action_object(self): + """ + Switching an existing rule to an action which declares object_model but not + object_required, leaving the picker blank, must clear the old action_object. + """ + class OptionalObjectAction(EventRuleAction): + slug = 'test.optional_object_action' + label = 'Optional Object Action' + object_model = Webhook + object_required = False + + register_event_rule_action(OptionalObjectAction) + self.addCleanup(registry['event_rule_actions'].pop, 'test.optional_object_action', None) + + webhook = Webhook.objects.create(name='Stale Webhook', payload_url='http://localhost:9000/') + webhook_type = ObjectType.objects.get_for_model(Webhook) + object_type = ObjectType.objects.get_for_model(Site) + rule = EventRule.objects.create( + name='Stale Object Rule', + event_types=[OBJECT_CREATED], + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=webhook_type, + action_object_id=webhook.pk, + ) + rule.object_types.set([object_type]) + + form = EventRuleForm(data={ + 'name': rule.name, + 'object_types': [object_type.pk], + 'event_types': [OBJECT_CREATED], + 'action_type': 'test.optional_object_action', + # action_choice omitted: the user left the picker blank + }, instance=rule) + self.assertTrue(form.is_valid(), form.errors) + saved = form.save() + self.assertIsNone(saved.action_object_type) + self.assertIsNone(saved.action_object_id) + + +class EventRuleFilterFormTestCase(TestCase): + + def test_action_type_choices_reflect_the_live_registry(self): + """ + The filter form's action_type choices must be read from the registry on access, not frozen + when this module was first imported. + """ + class FilterFormAction(EventRuleAction): + slug = 'test.filter_form_action' + label = 'Filter Form Action' + + register_event_rule_action(FilterFormAction) + self.addCleanup(registry['event_rule_actions'].pop, FilterFormAction.slug, None) + + choices = dict(EventRuleFilterForm().fields['action_type'].choices) + self.assertEqual(choices.get(FilterFormAction.slug), 'Filter Form Action') + self.assertIn(None, choices) # The blank choice is retained + + +class EventRuleImportFormTestCase(TestCase): + """ + EventRuleImportForm resolves action_object via each registered action's resolve_import_object() + hook, and treats it as optional (an action need not operate against a target object). + """ + + def tearDown(self): + super().tearDown() + registry['event_rule_actions'].pop('test.import_no_object_action', None) + + def test_resolves_webhook_by_name(self): + webhook = Webhook.objects.create(name='Import Test Webhook', payload_url='http://localhost:9000/') + form = EventRuleImportForm(data={ + 'name': 'Import Webhook Rule', + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': EventRuleActionChoices.WEBHOOK, + 'action_object': webhook.name, + }) + self.assertTrue(form.is_valid(), form.errors) + self.assertEqual(form.instance.action_object, webhook) + + def test_resolves_notification_group_by_name(self): + """The import form resolves a notification group, not just webhooks and scripts.""" + group = NotificationGroup.objects.create(name='Import Test Group') + form = EventRuleImportForm(data={ + 'name': 'Import Notification Rule', + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': EventRuleActionChoices.NOTIFICATION, + 'action_object': group.name, + }) + self.assertTrue(form.is_valid(), form.errors) + self.assertEqual(form.instance.action_object, group) + + def test_unresolvable_webhook_name_is_rejected(self): + form = EventRuleImportForm(data={ + 'name': 'Import Bad Webhook Rule', + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': EventRuleActionChoices.WEBHOOK, + 'action_object': 'Does Not Exist', + }) + self.assertFalse(form.is_valid()) + self.assertIn('action_object', form.errors) + + def test_unregistered_action_type_is_rejected(self): + form = EventRuleImportForm(data={ + 'name': 'Import Bad Type Rule', + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': 'not.a.registered.action', + 'action_object': 'whatever', + }) + self.assertFalse(form.is_valid()) + self.assertIn('action_type', form.errors) + + def test_submit_no_object_action_with_blank_action_object_succeeds(self): + """A blank action_object must be accepted for bulk-importing a no-object action.""" + class NoObjectAction(EventRuleAction): + slug = 'test.import_no_object_action' + label = 'Import No-Object Action' + object_required = False + + register_event_rule_action(NoObjectAction) + + form = EventRuleImportForm(data={ + 'name': 'Import No-Object Rule', + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': 'test.import_no_object_action', + 'action_object': '', + }) + self.assertTrue(form.is_valid(), form.errors) + rule = form.save() + self.assertIsNone(rule.action_object_type) + self.assertIsNone(rule.action_object_id) + + def test_blank_action_object_rejected_for_object_required_action(self): + """A blank action_object must be rejected cleanly (not raise) for an action which requires one.""" + form = EventRuleImportForm(data={ + 'name': 'Import Webhook No Object', + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': EventRuleActionChoices.WEBHOOK, + 'action_object': '', + }) + self.assertFalse(form.is_valid()) + self.assertIn('action_object', form.errors) + + def test_action_object_rejected_for_action_without_object_model(self): + """ + An action declaring no object_model rejects a supplied action_object as inapplicable, + rather than reporting it as an unsupported bulk import. + """ + class NoObjectAction(EventRuleAction): + slug = 'test.import_no_object_action' + label = 'Import No-Object Action' + object_required = False + + register_event_rule_action(NoObjectAction) + + form = EventRuleImportForm(data={ + 'name': 'Import No-Object Rule With Object', + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': 'test.import_no_object_action', + 'action_object': 'Some Object', + }) + self.assertFalse(form.is_valid()) + self.assertIn('does not operate against a target object', str(form.errors['action_object'])) + + def test_csv_update_to_optional_object_action_clears_stale_action_object(self): + """ + A CSV row updating an existing rule to an action_type which declares object_model but not + object_required, with action_object left blank, must clear the previous action_object. + """ + class OptionalObjectAction(EventRuleAction): + slug = 'test.import_optional_object_action' + label = 'Import Optional Object Action' + object_model = Webhook + object_required = False + + register_event_rule_action(OptionalObjectAction) + self.addCleanup(registry['event_rule_actions'].pop, 'test.import_optional_object_action', None) + + webhook = Webhook.objects.create(name='CSV Stale Webhook', payload_url='http://localhost:9000/') + webhook_type = ObjectType.objects.get_for_model(Webhook) + rule = EventRule.objects.create( + name='CSV Stale Object Rule', + event_types=[OBJECT_CREATED], + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=webhook_type, + action_object_id=webhook.pk, + ) + + form = EventRuleImportForm(data={ + 'name': rule.name, + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': 'test.import_optional_object_action', + 'action_object': '', + }, instance=rule) + self.assertTrue(form.is_valid(), form.errors) + saved = form.save() + self.assertIsNone(saved.action_object_type) + self.assertIsNone(saved.action_object_id) + + def test_action_validate_error_on_unexposed_field_becomes_non_field_error(self): + """A validate() error keyed by a field this form doesn't expose (e.g. action_data) must not raise.""" + class ActionDataValidatingAction(EventRuleAction): + slug = 'test.import_action_data_validating' + label = 'Import Action Data Validating' + object_required = False + + def validate(self, *, action_object, action_data): + raise ValidationError({'action_data': 'Bad action_data for test'}) + + register_event_rule_action(ActionDataValidatingAction) + self.addCleanup(registry['event_rule_actions'].pop, 'test.import_action_data_validating', None) + + form = EventRuleImportForm(data={ + 'name': 'Import Bad Action Data Rule', + 'object_types': 'dcim.site', + 'event_types': 'object_created', + 'action_type': 'test.import_action_data_validating', + 'action_object': '', + }) + self.assertFalse(form.is_valid()) + self.assertIn(NON_FIELD_ERRORS, form.errors) + self.assertIn('Bad action_data for test', form.errors[NON_FIELD_ERRORS]) diff --git a/netbox/extras/tests/test_graphql.py b/netbox/extras/tests/test_graphql.py new file mode 100644 index 000000000..2481ea2e8 --- /dev/null +++ b/netbox/extras/tests/test_graphql.py @@ -0,0 +1,56 @@ +import json + +from django.urls import reverse +from rest_framework import status + +from core.events import OBJECT_CREATED +from core.models import ObjectType +from dcim.models import Site +from extras.choices import EventRuleActionChoices +from extras.graphql.enums import EventRuleActionEnum +from extras.models import EventRule, Webhook +from utilities.testing import APITestCase + + +class EventRuleActionEnumTestCase(APITestCase): + """EventRuleActionEnum must reflect the live action registry, and the filter must use it.""" + + def test_enum_contains_core_actions(self): + # A subset check, since an installed plugin may register actions of its own + values = {member.value for member in EventRuleActionEnum} + core_slugs = { + EventRuleActionChoices.WEBHOOK, EventRuleActionChoices.SCRIPT, EventRuleActionChoices.NOTIFICATION, + } + self.assertLessEqual(core_slugs, values) + + def test_filter_event_rules_by_action_type(self): + webhook = Webhook.objects.create(name='GraphQL Enum Test Webhook', payload_url='http://localhost:9000/') + webhook_type = ObjectType.objects.get_for_model(Webhook) + site_type = ObjectType.objects.get_for_model(Site) + + webhook_rule = EventRule.objects.create( + name='GraphQL Enum Webhook Rule', + event_types=[OBJECT_CREATED], + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=webhook_type, + action_object_id=webhook.pk, + ) + webhook_rule.object_types.set([site_type]) + + script_rule = EventRule.objects.create( + name='GraphQL Enum Script Rule', + event_types=[OBJECT_CREATED], + action_type=EventRuleActionChoices.SCRIPT, + ) + script_rule.object_types.set([site_type]) + + self.add_permissions('extras.view_eventrule') + url = reverse('graphql') + query = '{event_rule_list(filters: {action_type: {exact: WEBHOOK}}) {name action_type}}' + response = self.client.post(url, data={'query': query}, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + data = json.loads(response.content) + self.assertNotIn('errors', data) + names = {rule['name'] for rule in data['data']['event_rule_list']} + self.assertEqual(names, {'GraphQL Enum Webhook Rule'}) diff --git a/netbox/extras/tests/test_models.py b/netbox/extras/tests/test_models.py index 6f614654c..39c2abe22 100644 --- a/netbox/extras/tests/test_models.py +++ b/netbox/extras/tests/test_models.py @@ -32,6 +32,7 @@ from extras.models import ( TableConfig, Tag, TaggedItem, + Webhook, ) from extras.models.mixins import RenderTemplateMixin from tenancy.models import Tenant, TenantGroup @@ -1586,8 +1587,16 @@ class EventRuleTestCase(TestCase): """ clean() should accept a JSON object (or null) as action_data. """ + webhook = Webhook.objects.create(name='Action Data Test Webhook', payload_url='http://localhost:9000/') + webhook_type = ObjectType.objects.get_for_model(Webhook) for value in ({'key': 'value'}, None): - rule = EventRule(name='test', event_types=[OBJECT_CREATED], action_data=value) + rule = EventRule( + name='test', + event_types=[OBJECT_CREATED], + action_data=value, + action_object_type=webhook_type, + action_object_id=webhook.pk, + ) rule.clean() def test_action_data_clean_rejects_non_dict(self): diff --git a/netbox/extras/tests/test_tables.py b/netbox/extras/tests/test_tables.py index 1fbd1fe2b..1afdbf87d 100644 --- a/netbox/extras/tests/test_tables.py +++ b/netbox/extras/tests/test_tables.py @@ -1,4 +1,9 @@ -from extras.models import Bookmark, Notification, Subscription +from django.test import TestCase + +from core.events import OBJECT_CREATED +from core.models import ObjectType +from dcim.models import Site +from extras.models import Bookmark, EventRule, Notification, Subscription from extras.tables import * from utilities.testing import TableTestCases @@ -69,6 +74,39 @@ class EventRuleTableTestCase(TableTestCases.StandardTableTestCase): table = EventRuleTable +class EventRuleTableActionTypeRenderingTestCase(TestCase): + """ + render_action_type() badges an unregistered action as unavailable; value_action_type() carries + the same label for non-HTML output (e.g. CSV export), without the markup. + """ + + def test_render_action_type_for_registered_action(self): + rule = EventRule.objects.create(name='Render Test Rule', event_types=[OBJECT_CREATED], action_type='webhook') + rule.object_types.set([ObjectType.objects.get_for_model(Site)]) + + table = EventRuleTable(EventRule.objects.filter(pk=rule.pk)) + self.assertEqual(table.render_action_type(rule), 'Webhook') + self.assertEqual(table.value_action_type(rule), 'Webhook') + + def test_render_action_type_for_unregistered_action(self): + rule = EventRule.objects.create( + name='Render Test Unavailable Rule', + event_types=[OBJECT_CREATED], + action_type='someplugin.not_installed_render_test', + ) + rule.object_types.set([ObjectType.objects.get_for_model(Site)]) + + table = EventRuleTable(EventRule.objects.filter(pk=rule.pk)) + rendered = table.render_action_type(rule) + self.assertIn('someplugin.not_installed_render_test (unavailable)', rendered) + self.assertIn('badge text-bg-red', rendered) + + # The same label, without markup + value = table.value_action_type(rule) + self.assertEqual(value, 'someplugin.not_installed_render_test (unavailable)') + self.assertNotIn('" + + def get_object_queryset(self): + """ + Return the queryset of objects eligible for selection as this action's action_object, or + None if object_model is not set. + """ + if self.object_model is None: + return None + return self.object_model.objects.all() + + def get_object_label(self): + """ + Return the label for this action's object selection field, or None if object_model is not + set. Defaults to object_model's verbose name, unless object_label has been set explicitly. + """ + if self.object_label: + return self.object_label + if self.object_model is None: + return None + return capfirst(self.object_model._meta.verbose_name) + + def resolve_import_object(self, value): + """ + Optional hook: resolve a CSV/bulk-import "action object" string to a model instance. Raise + django.core.exceptions.ObjectDoesNotExist (or a subclass) if the value doesn't resolve. + Return None (the default) if this action doesn't support bulk import. + """ + return + + def _validate(self, *, action_object, action_data): + """ + Entry point called from EventRule.clean(). Enforces the base object_required/object_model + checks, then delegates to validate() for any action-specific validation. + """ + if self.object_required and action_object is None: + raise ValidationError({ + 'action_object_id': _("This action requires a target object to be selected."), + }) + if action_object is not None: + if self.object_model is None: + raise ValidationError({ + 'action_object_id': _("This action does not operate against a target object."), + }) + if not isinstance(action_object, self.object_model): + raise ValidationError({ + 'action_object_id': _("Selected object is not a valid {model}.").format( + model=self.object_model._meta.verbose_name + ), + }) + self.validate(action_object=action_object, action_data=action_data) + + def validate(self, *, action_object, action_data): + """ + Optional hook: add custom validation, raising ValidationError on failure. No-op by + default; no need to call super() -- _validate() above runs the base checks regardless. + """ + pass + + def enqueue(self, *, event_rule, event_context, action_object, action_data): + """ + Perform (or schedule) this action in response to a queued event. Implementations should + not raise for conditions that are the fault of this EventRule's own configuration alone; + log and return instead, so that other EventRules processed in the same batch are + unaffected. + """ + raise NotImplementedError(f"{self.__class__.__name__} must implement enqueue()") + + +def register_event_rule_action(cls, *, is_plugin_provided=True): + """ + Register an EventRuleAction subclass. Can be used as a decorator, or called directly (e.g. when + iterating a plugin's declared event_rule_actions): + + @register_event_rule_action + class MyAction(EventRuleAction): + slug = 'myplugin.my_action' + ... + + Raises ImproperlyConfigured if slug/label are missing, the slug is malformed, already + registered, collides via enum_key() with another registered slug once both feed the GraphQL + EventRuleActionEnum (see extras.graphql.enums), or object_required is set without an + object_model to validate the object against. Checking slug/label here rather than at class + definition means an intermediate base class shared by several concrete actions in a plugin can + leave them unset. + + is_plugin_provided determines whether a dispatch-time exception from this action is isolated + or propagates (see process_event_rules() in extras.events); defaults to True. NetBox's own + core registrations (extras.apps.ExtrasConfig) pass False explicitly. + """ + instance = cls() + if not instance.slug: + raise ImproperlyConfigured(f"{cls.__name__} must define a non-empty 'slug' attribute.") + if not instance.label: + raise ImproperlyConfigured(f"{cls.__name__} must define a 'label' attribute.") + if instance.object_required and instance.object_model is None: + # Unsatisfiable: an object is mandatory, yet any object supplied is rejected by _validate(). + raise ImproperlyConfigured( + f"{cls.__name__} sets object_required but no object_model; a target object cannot be " + f"required for an action which does not operate against one." + ) + if not SLUG_RE.fullmatch(instance.slug): + raise ImproperlyConfigured( + f"Invalid event rule action slug {instance.slug!r}: must be lowercase, start with a " + f"letter, and use only letters, digits, underscores, and dot-separated segments." + ) + if instance.slug in registry['event_rule_actions']: + raise ImproperlyConfigured(f"An event rule action named {instance.slug} has already been registered!") + new_key = enum_key(instance.slug) + for existing in registry['event_rule_actions'].values(): + if enum_key(existing.slug) == new_key: + raise ImproperlyConfigured( + f"Event rule action slug {instance.slug!r} collides with the already-registered " + f"{existing.slug!r} once both are sanitized into a GraphQL enum member name." + ) + instance.is_plugin_provided = is_plugin_provided + registry['event_rule_actions'][instance.slug] = instance + return cls + + +def get_event_rule_action(slug): + return registry['event_rule_actions'].get(slug) + + +def get_event_rule_action_choices(): + return [ + Choice(action.slug, action.label, description=action.description) + for action in registry['event_rule_actions'].values() + ] + + +def get_event_rule_action_slugs(): + return list(registry['event_rule_actions'].keys()) diff --git a/netbox/netbox/plugins/__init__.py b/netbox/netbox/plugins/__init__.py index 62cf47c16..7c45c6b48 100644 --- a/netbox/netbox/plugins/__init__.py +++ b/netbox/netbox/plugins/__init__.py @@ -7,6 +7,7 @@ from django.utils.module_loading import import_string from packaging import version from core.exceptions import IncompatiblePluginError +from netbox.event_rules import register_event_rule_action from netbox.registry import registry from netbox.search import register_search from netbox.utils import register_data_backend @@ -35,6 +36,7 @@ registry['plugins'].update({ DEFAULT_RESOURCE_PATHS = { 'search_indexes': 'search.indexes', 'data_backends': 'data_backends.backends', + 'event_rule_actions': 'event_rules.event_rule_actions', 'graphql_schema': 'graphql.schema', 'jinja_filters': 'jinja_env.filters', 'graphql_type_extensions': 'graphql.type_extensions', @@ -86,6 +88,7 @@ class PluginConfig(AppConfig): # Optional plugin resources search_indexes = None data_backends = None + event_rule_actions = None graphql_schema = None jinja_filters = None graphql_type_extensions = None @@ -142,6 +145,11 @@ class PluginConfig(AppConfig): for backend in data_backends: register_data_backend()(backend) + # Register event rule actions (if defined) + event_rule_actions = self._load_resource('event_rule_actions') or [] + for action in event_rule_actions: + register_event_rule_action(action) + # Register Jinja filters (if defined) if jinja_filters := self._load_resource('jinja_filters'): register_jinja_filters(jinja_filters) diff --git a/netbox/netbox/registry.py b/netbox/netbox/registry.py index 35ce3c976..caa67169d 100644 --- a/netbox/netbox/registry.py +++ b/netbox/netbox/registry.py @@ -25,6 +25,7 @@ class Registry(dict): registry = Registry({ 'counter_fields': collections.defaultdict(dict), 'data_backends': dict(), + 'event_rule_actions': dict(), 'event_types': dict(), 'filtersets': dict(), 'model_actions': collections.defaultdict(set), diff --git a/netbox/netbox/tests/dummy_plugin/event_rules.py b/netbox/netbox/tests/dummy_plugin/event_rules.py new file mode 100644 index 000000000..fa3d5af29 --- /dev/null +++ b/netbox/netbox/tests/dummy_plugin/event_rules.py @@ -0,0 +1,15 @@ +from netbox.event_rules import EventRuleAction + +__all__ = ( + 'DummyRaisingAction', +) + + +class DummyRaisingAction(EventRuleAction): + """A plugin action for testing process_event_rules() exception handling. Registered per-test, not on load.""" + slug = 'dummy_plugin.raising_action' + label = 'Dummy Raising Action' + object_required = False + + def enqueue(self, **kwargs): + raise RuntimeError("intentional failure for test")