NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-13 16:47:34 -06:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Closes #16886: Dynamic event types (#16938)

Browse Source 
* Initial work on #16886

* Restore GraphQL filter

* Remove namespace

* Add Event documentation

* Use MultipleChoiceField for event_types

* Fix event_types field class on EventRuleImportForm

* Fix tests

* Simplify event queue handling logic

* Misc cleanup
This commit is contained in:
Jeremy Stretch 2024-07-22 08:12:04 -04:00 committed by GitHub
parent 4a53a96541
commit 44a9350986
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
22 changed files with 269 additions and 268 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
docs/plugins/development/events.md Normal file
View File

@ -0,0 +1,16 @@
# Events
Plugins can register their own custom event types for use with NetBox [event rules](../../models/extras/eventrule.md). This is accomplished by calling the `register()` method on an instance of the `Event` class. This can be done anywhere within the plugin. An example is provided below.
```python
from django.utils.translation import gettext_lazy as _
from netbox.events import Event, EVENT_TYPE_SUCCESS
Event(
name='ticket_opened',
text=_('Ticket opened'),
type=EVENT_TYPE_SUCCESS
).register()
```
::: netbox.events.Event

1
mkdocs.yml
View File

@ -142,6 +142,7 @@ nav:
- Forms: 'plugins/development/forms.md'
- Filters & Filter Sets: 'plugins/development/filtersets.md'
- Search: 'plugins/development/search.md'
- Events: 'plugins/development/events.md'
- Data Backends: 'plugins/development/data-backends.md'
- REST API: 'plugins/development/rest-api.md'
- GraphQL API: 'plugins/development/graphql-api.md'

16
netbox/core/events.py
View File

@ -1,6 +1,6 @@
from django.utils.translation import gettext as _
from netbox.events import *
from netbox.events import Event, EVENT_TYPE_DANGER, EVENT_TYPE_SUCCESS, EVENT_TYPE_WARNING
__all__ = (
'JOB_COMPLETED',
@ -24,10 +24,10 @@ JOB_FAILED = 'job_failed'
JOB_ERRORED = 'job_errored'
# Register core events
Event(name=OBJECT_CREATED, text=_('Object created')).register()
Event(name=OBJECT_UPDATED, text=_('Object updated')).register()
Event(name=OBJECT_DELETED, text=_('Object deleted')).register()
Event(name=JOB_STARTED, text=_('Job started')).register()
Event(name=JOB_COMPLETED, text=_('Job completed'), type=EVENT_TYPE_SUCCESS).register()
Event(name=JOB_FAILED, text=_('Job failed'), type=EVENT_TYPE_WARNING).register()
Event(name=JOB_ERRORED, text=_('Job errored'), type=EVENT_TYPE_DANGER).register()
Event(OBJECT_CREATED, _('Object created')).register()
Event(OBJECT_UPDATED, _('Object updated')).register()
Event(OBJECT_DELETED, _('Object deleted')).register()
Event(JOB_STARTED, _('Job started')).register()
Event(JOB_COMPLETED, _('Job completed'), type=EVENT_TYPE_SUCCESS).register()
Event(JOB_FAILED, _('Job failed'), type=EVENT_TYPE_WARNING).register()
Event(JOB_ERRORED, _('Job errored'), type=EVENT_TYPE_DANGER).register()

6
netbox/extras/api/serializers_/events.py
View File

@ -34,9 +34,9 @@ class EventRuleSerializer(NetBoxModelSerializer):
class Meta:
model = EventRule
fields = [
'id', 'url', 'display_url', 'display', 'object_types', 'name', 'type_create', 'type_update', 'type_delete',
'type_job_start', 'type_job_end', 'enabled', 'conditions', 'action_type', 'action_object_type',
'action_object_id', 'action_object', 'description', 'custom_fields', 'tags', 'created', 'last_updated',
'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',
'tags', 'created', 'last_updated',
]
brief_fields = ('id', 'url', 'display', 'name', 'description')

28
netbox/extras/events.py
View File

@ -1,3 +1,4 @@
from collections import defaultdict
import logging
from django.conf import settings
@ -152,35 +153,20 @@ def process_event_queue(events):
"""
Flush a list of object representation to RQ for EventRule processing.
"""
events_cache = {
'type_create': {},
'type_update': {},
'type_delete': {},
}
event_actions = {
# TODO: Add EventRule support for dynamically registered event types
OBJECT_CREATED: 'type_create',
OBJECT_UPDATED: 'type_update',
OBJECT_DELETED: 'type_delete',
JOB_STARTED: 'type_job_start',
JOB_COMPLETED: 'type_job_end',
# Map failed & errored jobs to type_job_end
JOB_FAILED: 'type_job_end',
JOB_ERRORED: 'type_job_end',
}
events_cache = defaultdict(dict)
for event in events:
action_flag = event_actions[event['event_type']]
event_type = event['event_type']
object_type = event['object_type']
# Cache applicable Event Rules
if object_type not in events_cache[action_flag]:
events_cache[action_flag][object_type] = EventRule.objects.filter(
**{action_flag: True},
if object_type not in events_cache[event_type]:
events_cache[event_type][object_type] = EventRule.objects.filter(
event_types__contains=[event['event_type']],
object_types=object_type,
enabled=True
)
event_rules = events_cache[action_flag][object_type]
event_rules = events_cache[event_type][object_type]
process_event_rules(
event_rules=event_rules,

9
netbox/extras/filtersets.py
View File

@ -99,6 +99,9 @@ class EventRuleFilterSet(NetBoxModelFilterSet):
object_type = ContentTypeFilter(
field_name='object_types'
)
event_type = MultiValueCharFilter(
method='filter_event_type'
)
action_type = django_filters.MultipleChoiceFilter(
choices=EventRuleActionChoices
)
@ -108,8 +111,7 @@ class EventRuleFilterSet(NetBoxModelFilterSet):
class Meta:
model = EventRule
fields = (
'id', 'name', 'type_create', 'type_update', 'type_delete', 'type_job_start', 'type_job_end', 'enabled',
'action_type', 'description',
'id', 'name', 'enabled', 'action_type', 'description',
)
def search(self, queryset, name, value):
@ -121,6 +123,9 @@ class EventRuleFilterSet(NetBoxModelFilterSet):
Q(comments__icontains=value)
)
def filter_event_type(self, queryset, name, value):
return queryset.filter(event_types__overlap=value)
class CustomFieldFilterSet(ChangeLoggedModelFilterSet):
q = django_filters.CharFilter(

32
netbox/extras/forms/bulk_edit.py
View File

@ -3,6 +3,7 @@ from django.utils.translation import gettext_lazy as _
from extras.choices import *
from extras.models import *
from netbox.events import get_event_type_choices
from netbox.forms import NetBoxModelBulkEditForm
from utilities.forms import BulkEditForm, add_blank_choice
from utilities.forms.fields import ColorField, CommentField, DynamicModelChoiceField
@ -248,33 +249,18 @@ class EventRuleBulkEditForm(NetBoxModelBulkEditForm):
required=False,
widget=BulkEditNullBooleanSelect()
)
type_create = forms.NullBooleanField(
label=_('On create'),
event_types = forms.MultipleChoiceField(
choices=get_event_type_choices(),
required=False,
widget=BulkEditNullBooleanSelect()
label=_('Event types')
)
type_update = forms.NullBooleanField(
label=_('On update'),
required=False,
widget=BulkEditNullBooleanSelect()
)
type_delete = forms.NullBooleanField(
label=_('On delete'),
required=False,
widget=BulkEditNullBooleanSelect()
)
type_job_start = forms.NullBooleanField(
label=_('On job start'),
required=False,
widget=BulkEditNullBooleanSelect()
)
type_job_end = forms.NullBooleanField(
label=_('On job end'),
required=False,
widget=BulkEditNullBooleanSelect()
description = forms.CharField(
label=_('Description'),
max_length=200,
required=False
)
nullable_fields = ('description', 'conditions',)
nullable_fields = ('description', 'conditions')
class TagBulkEditForm(BulkEditForm):

14
netbox/extras/forms/bulk_import.py
View File

@ -8,12 +8,13 @@ from django.utils.translation import gettext_lazy as _
from core.models import ObjectType
from extras.choices import *
from extras.models import *
from netbox.events import get_event_type_choices
from netbox.forms import NetBoxModelImportForm
from users.models import Group, User
from utilities.forms import CSVModelForm
from utilities.forms.fields import (
CSVChoiceField, CSVContentTypeField, CSVModelChoiceField, CSVModelMultipleChoiceField, CSVMultipleContentTypeField,
SlugField,
CSVChoiceField, CSVContentTypeField, CSVModelChoiceField, CSVModelMultipleChoiceField, CSVMultipleChoiceField,
CSVMultipleContentTypeField, SlugField,
)
__all__ = (
@ -187,6 +188,11 @@ class EventRuleImportForm(NetBoxModelImportForm):
queryset=ObjectType.objects.with_feature('event_rules'),
help_text=_("One or more assigned object types")
)
event_types = CSVMultipleChoiceField(
choices=get_event_type_choices(),
label=_('Event types'),
help_text=_('The event type(s) which will trigger this rule')
)
action_object = forms.CharField(
label=_('Action object'),
required=True,
@ -196,8 +202,8 @@ class EventRuleImportForm(NetBoxModelImportForm):
class Meta:
model = EventRule
fields = (
'name', 'description', 'enabled', 'conditions', 'object_types', 'type_create', 'type_update',
'type_delete', 'type_job_start', 'type_job_end', 'action_type', 'action_object', 'comments', 'tags'
'name', 'description', 'enabled', 'conditions', 'object_types', 'event_types', 'action_type',
'action_object', 'comments', 'tags'
)
def clean(self):

44
netbox/extras/forms/filtersets.py
View File

@ -6,6 +6,7 @@ from core.models import ObjectType, DataFile, DataSource
from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site, SiteGroup
from extras.choices import *
from extras.models import *
from netbox.events import get_event_type_choices
from netbox.forms.base import NetBoxModelFilterSetForm
from netbox.forms.mixins import SavedFiltersMixin
from tenancy.models import Tenant, TenantGroup
@ -274,14 +275,18 @@ class EventRuleFilterForm(NetBoxModelFilterSetForm):
fieldsets = (
FieldSet('q', 'filter_id', 'tag'),
FieldSet('object_type_id', 'action_type', 'enabled', name=_('Attributes')),
FieldSet('type_create', 'type_update', 'type_delete', 'type_job_start', 'type_job_end', name=_('Events')),
FieldSet('object_type_id', 'event_type', 'action_type', 'enabled', name=_('Attributes')),
)
object_type_id = ContentTypeMultipleChoiceField(
queryset=ObjectType.objects.with_feature('event_rules'),
required=False,
label=_('Object type')
)
event_type = forms.MultipleChoiceField(
choices=get_event_type_choices,
required=False,
label=_('Event type')
)
action_type = forms.ChoiceField(
choices=add_blank_choice(EventRuleActionChoices),
required=False,
@ -294,41 +299,6 @@ class EventRuleFilterForm(NetBoxModelFilterSetForm):
choices=BOOLEAN_WITH_BLANK_CHOICES
)
)
type_create = forms.NullBooleanField(
required=False,
widget=forms.Select(
choices=BOOLEAN_WITH_BLANK_CHOICES
),
label=_('Object creations')
)
type_update = forms.NullBooleanField(
required=False,
widget=forms.Select(
choices=BOOLEAN_WITH_BLANK_CHOICES
),
label=_('Object updates')
)
type_delete = forms.NullBooleanField(
required=False,
widget=forms.Select(
choices=BOOLEAN_WITH_BLANK_CHOICES
),
label=_('Object deletions')
)
type_job_start = forms.NullBooleanField(
required=False,
widget=forms.Select(
choices=BOOLEAN_WITH_BLANK_CHOICES
),
label=_('Job starts')
)
type_job_end = forms.NullBooleanField(
required=False,
widget=forms.Select(
choices=BOOLEAN_WITH_BLANK_CHOICES
),
label=_('Job terminations')
)
class TagFilterForm(SavedFiltersMixin, FilterForm):

20
netbox/extras/forms/model_forms.py
View File

@ -10,6 +10,7 @@ from core.models import ObjectType
from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site, SiteGroup
from extras.choices import *
from extras.models import *
from netbox.events import get_event_type_choices
from netbox.forms import NetBoxModelForm
from tenancy.models import Tenant, TenantGroup
from users.models import Group, User
@ -303,6 +304,10 @@ class EventRuleForm(NetBoxModelForm):
label=_('Object types'),
queryset=ObjectType.objects.with_feature('event_rules'),
)
event_types = forms.MultipleChoiceField(
choices=get_event_type_choices(),
label=_('Event types')
)
action_choice = forms.ChoiceField(
label=_('Action choice'),
choices=[]
@ -319,25 +324,16 @@ class EventRuleForm(NetBoxModelForm):
fieldsets = (
FieldSet('name', 'description', 'object_types', 'enabled', 'tags', name=_('Event Rule')),
FieldSet('type_create', 'type_update', 'type_delete', 'type_job_start', 'type_job_end', name=_('Events')),
FieldSet('conditions', name=_('Conditions')),
FieldSet('event_types', 'conditions', name=_('Triggers')),
FieldSet('action_type', 'action_choice', 'action_data', name=_('Action')),
)
class Meta:
model = EventRule
fields = (
'object_types', 'name', 'description', 'type_create', 'type_update', 'type_delete', 'type_job_start',
'type_job_end', 'enabled', 'conditions', 'action_type', 'action_object_type', 'action_object_id',
'action_data', 'comments', 'tags'
'object_types', 'name', 'description', 'enabled', 'event_types', 'conditions', 'action_type',
'action_object_type', 'action_object_id', 'action_data', 'comments', 'tags'
)
labels = {
'type_create': _('Creations'),
'type_update': _('Updates'),
'type_delete': _('Deletions'),
'type_job_start': _('Job executions'),
'type_job_end': _('Job terminations'),
}
widgets = {
'conditions': forms.Textarea(attrs={'class': 'font-monospace'}),
'action_type': HTMXSelect(),

75
netbox/extras/migrations/0119_eventrule_event_types.py Normal file
View File

@ -0,0 +1,75 @@
import django.contrib.postgres.fields
from django.db import migrations, models
from core.events import *
def set_event_types(apps, schema_editor):
EventRule = apps.get_model('extras', 'EventRule')
event_rules = EventRule.objects.all()
for event_rule in event_rules:
event_rule.event_types = []
if event_rule.type_create:
event_rule.event_types.append(OBJECT_CREATED)
if event_rule.type_update:
event_rule.event_types.append(OBJECT_UPDATED)
if event_rule.type_delete:
event_rule.event_types.append(OBJECT_DELETED)
if event_rule.type_job_start:
event_rule.event_types.append(JOB_STARTED)
if event_rule.type_job_end:
# Map type_job_end to all job termination events
event_rule.event_types.extend([JOB_COMPLETED, JOB_ERRORED, JOB_FAILED])
EventRule.objects.bulk_update(event_rules, ['event_types'])
class Migration(migrations.Migration):
dependencies = [
('extras', '0118_notifications'),
]
operations = [
migrations.AddField(
model_name='eventrule',
name='event_types',
field=django.contrib.postgres.fields.ArrayField(
base_field=models.CharField(max_length=50),
blank=True,
null=True,
size=None
),
),
migrations.RunPython(
code=set_event_types,
reverse_code=migrations.RunPython.noop
),
migrations.AlterField(
model_name='eventrule',
name='event_types',
field=django.contrib.postgres.fields.ArrayField(base_field=models.CharField(max_length=50), size=None),
preserve_default=False,
),
migrations.RemoveField(
model_name='eventrule',
name='type_create',
),
migrations.RemoveField(
model_name='eventrule',
name='type_delete',
),
migrations.RemoveField(
model_name='eventrule',
name='type_job_end',
),
migrations.RemoveField(
model_name='eventrule',
name='type_job_start',
),
migrations.RemoveField(
model_name='eventrule',
name='type_update',
),
]

37
netbox/extras/models/models.py
View File

@ -3,6 +3,7 @@ import urllib.parse
from django.conf import settings
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.postgres.fields import ArrayField
from django.core.validators import ValidationError
from django.db import models
from django.http import HttpResponse
@ -17,6 +18,7 @@ from extras.conditions import ConditionSet
from extras.constants import *
from extras.utils import image_upload
from netbox.config import get_config
from netbox.events import get_event_type_choices
from netbox.models import ChangeLoggedModel
from netbox.models.features import (
CloningMixin, CustomFieldsMixin, CustomLinksMixin, ExportTemplatesMixin, SyncedDataMixin, TagsMixin,
@ -60,30 +62,9 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, ChangeLogged
max_length=200,
blank=True
)
type_create = models.BooleanField(
verbose_name=_('on create'),
default=False,
help_text=_("Triggers when a matching object is created.")
)
type_update = models.BooleanField(
verbose_name=_('on update'),
default=False,
help_text=_("Triggers when a matching object is updated.")
)
type_delete = models.BooleanField(
verbose_name=_('on delete'),
default=False,
help_text=_("Triggers when a matching object is deleted.")
)
type_job_start = models.BooleanField(
verbose_name=_('on job start'),
default=False,
help_text=_("Triggers when a job for a matching object is started.")
)
type_job_end = models.BooleanField(
verbose_name=_('on job end'),
default=False,
help_text=_("Triggers when a job for a matching object terminates.")
event_types = ArrayField(
base_field=models.CharField(max_length=50, choices=get_event_type_choices),
help_text=_("The types of event which will trigger this rule.")
)
enabled = models.BooleanField(
verbose_name=_('enabled'),
@ -144,14 +125,6 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, ChangeLogged
def clean(self):
super().clean()
# At least one action type must be selected
if not any([
self.type_create, self.type_update, self.type_delete, self.type_job_start, self.type_job_end
]):
raise ValidationError(
_("At least one event type must be selected: create, update, delete, job start, and/or job end.")
)
# Validate that any conditions are in the correct format
if self.conditions:
try:

25
netbox/extras/tables/tables.py
View File

@ -6,6 +6,7 @@ from django.utils.translation import gettext_lazy as _
from extras.models import *
from netbox.constants import EMPTY_TABLE_TEXT
from netbox.events import get_event_text
from netbox.tables import BaseTable, NetBoxTable, columns
from .columns import NotificationActionsColumn
@ -399,20 +400,10 @@ class EventRuleTable(NetBoxTable):
enabled = columns.BooleanColumn(
verbose_name=_('Enabled'),
)
type_create = columns.BooleanColumn(
verbose_name=_('Create')
)
type_update = columns.BooleanColumn(
verbose_name=_('Update')
)
type_delete = columns.BooleanColumn(
verbose_name=_('Delete')
)
type_job_start = columns.BooleanColumn(
verbose_name=_('Job Start')
)
type_job_end = columns.BooleanColumn(
verbose_name=_('Job End')
event_types = columns.ArrayColumn(
verbose_name=_('Event Types'),
func=get_event_text,
orderable=False
)
tags = columns.TagColumn(
url_name='extras:webhook_list'
@ -422,12 +413,10 @@ class EventRuleTable(NetBoxTable):
model = EventRule
fields = (
'pk', 'id', 'name', 'enabled', 'description', 'action_type', 'action_object', 'object_types',
'type_create', 'type_update', 'type_delete', 'type_job_start', 'type_job_end', 'tags', 'created',
'last_updated',
'event_types', 'tags', 'created', 'last_updated',
)
default_columns = (
'pk', 'name', 'enabled', 'action_type', 'action_object', 'object_types', 'type_create', 'type_update',
'type_delete', 'type_job_start', 'type_job_end',
'pk', 'name', 'enabled', 'action_type', 'action_object', 'object_types', 'event_types',
)

14
netbox/extras/tests/test_api.py
View File

@ -1,6 +1,5 @@
import datetime
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.utils.timezone import make_aware
@ -13,6 +12,7 @@ from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Rack, Loca
from extras.choices import *
from extras.models import *
from extras.scripts import BooleanVar, IntegerVar, Script as PythonClass, StringVar
from netbox.events import *
from users.models import Group, User
from utilities.testing import APITestCase, APIViewTestCases
@ -113,9 +113,9 @@ class EventRuleTest(APIViewTestCases.APIViewTestCase):
Webhook.objects.bulk_create(webhooks)
event_rules = (
EventRule(name='EventRule 1', type_create=True, action_object=webhooks[0]),
EventRule(name='EventRule 2', type_create=True, action_object=webhooks[1]),
EventRule(name='EventRule 3', type_create=True, action_object=webhooks[2]),
EventRule(name='EventRule 1', event_types=[OBJECT_CREATED], action_object=webhooks[0]),
EventRule(name='EventRule 2', event_types=[OBJECT_CREATED], action_object=webhooks[1]),
EventRule(name='EventRule 3', event_types=[OBJECT_CREATED], action_object=webhooks[2]),
)
EventRule.objects.bulk_create(event_rules)
@ -123,7 +123,7 @@ class EventRuleTest(APIViewTestCases.APIViewTestCase):
{
'name': 'EventRule 4',
'object_types': ['dcim.device', 'dcim.devicetype'],
'type_create': True,
'event_types': [OBJECT_CREATED],
'action_type': EventRuleActionChoices.WEBHOOK,
'action_object_type': 'extras.webhook',
'action_object_id': webhooks[3].pk,
@ -131,7 +131,7 @@ class EventRuleTest(APIViewTestCases.APIViewTestCase):
{
'name': 'EventRule 5',
'object_types': ['dcim.device', 'dcim.devicetype'],
'type_create': True,
'event_types': [OBJECT_CREATED],
'action_type': EventRuleActionChoices.WEBHOOK,
'action_object_type': 'extras.webhook',
'action_object_id': webhooks[4].pk,
@ -139,7 +139,7 @@ class EventRuleTest(APIViewTestCases.APIViewTestCase):
{
'name': 'EventRule 6',
'object_types': ['dcim.device', 'dcim.devicetype'],
'type_create': True,
'event_types': [OBJECT_CREATED],
'action_type': EventRuleActionChoices.WEBHOOK,
'action_object_type': 'extras.webhook',
'action_object_id': webhooks[5].pk,

13
netbox/extras/tests/test_conditions.py
View File

@ -1,6 +1,7 @@
from django.contrib.contenttypes.models import ContentType
from django.test import TestCase
from core.events import *
from dcim.choices import SiteStatusChoices
from dcim.models import Site
from extras.conditions import Condition, ConditionSet
@ -230,8 +231,7 @@ class ConditionSetTest(TestCase):
"""
event_rule = EventRule(
name='Event Rule 1',
type_create=True,
type_update=True,
event_types=[OBJECT_CREATED, OBJECT_UPDATED],
conditions={
'attr': 'status.value',
'value': 'active',
@ -251,8 +251,7 @@ class ConditionSetTest(TestCase):
"""
event_rule = EventRule(
name='Event Rule 1',
type_create=True,
type_update=True,
event_types=[OBJECT_CREATED, OBJECT_UPDATED],
conditions={
"attr": "status.value",
"value": ["planned", "staging"],
@ -273,8 +272,7 @@ class ConditionSetTest(TestCase):
"""
event_rule = EventRule(
name='Event Rule 1',
type_create=True,
type_update=True,
event_types=[OBJECT_CREATED, OBJECT_UPDATED],
conditions={
"attr": "status.value",
"value": ["planned", "staging"],
@ -300,8 +298,7 @@ class ConditionSetTest(TestCase):
webhook = Webhook.objects.create(name='Webhook 100', payload_url='http://example.com/?1', http_method='POST')
form = EventRuleForm({
"name": "Event Rule 1",
"type_create": True,
"type_update": True,
"event_types": [OBJECT_CREATED, OBJECT_UPDATED],
"action_object_type": ct.pk,
"action_type": "webhook",
"action_choice": webhook.pk,

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

@ -46,22 +46,22 @@ class EventRuleTest(APITestCase):
webhook_type = ObjectType.objects.get(app_label='extras', model='webhook')
event_rules = EventRule.objects.bulk_create((
EventRule(
name='Webhook Event 1',
type_create=True,
name='Event Rule 1',
event_types=[OBJECT_CREATED],
action_type=EventRuleActionChoices.WEBHOOK,
action_object_type=webhook_type,
action_object_id=webhooks[0].id
),
EventRule(
name='Webhook Event 2',
type_update=True,
name='Event Rule 2',
event_types=[OBJECT_UPDATED],
action_type=EventRuleActionChoices.WEBHOOK,
action_object_type=webhook_type,
action_object_id=webhooks[0].id
),
EventRule(
name='Webhook Event 3',
type_delete=True,
name='Event Rule 3',
event_types=[OBJECT_DELETED],
action_type=EventRuleActionChoices.WEBHOOK,
action_object_type=webhook_type,
action_object_id=webhooks[0].id
@ -82,8 +82,7 @@ class EventRuleTest(APITestCase):
"""
event_rule = EventRule(
name='Event Rule 1',
type_create=True,
type_update=True,
event_types=[OBJECT_CREATED, OBJECT_UPDATED],
conditions={
'and': [
{
@ -131,7 +130,7 @@ class EventRuleTest(APITestCase):
# Verify that a background task was queued for the new object
self.assertEqual(self.queue.count, 1)
job = self.queue.jobs[0]
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_create=True))
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 1'))
self.assertEqual(job.kwargs['event_type'], OBJECT_CREATED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['data']['id'], response.data['id'])
@ -181,7 +180,7 @@ class EventRuleTest(APITestCase):
# Verify that a background task was queued for each new object
self.assertEqual(self.queue.count, 3)
for i, job in enumerate(self.queue.jobs):
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_create=True))
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 1'))
self.assertEqual(job.kwargs['event_type'], OBJECT_CREATED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['data']['id'], response.data[i]['id'])
@ -212,7 +211,7 @@ class EventRuleTest(APITestCase):
# Verify that a background task was queued for the updated object
self.assertEqual(self.queue.count, 1)
job = self.queue.jobs[0]
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_update=True))
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 2'))
self.assertEqual(job.kwargs['event_type'], OBJECT_UPDATED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['data']['id'], site.pk)
@ -268,7 +267,7 @@ class EventRuleTest(APITestCase):
# Verify that a background task was queued for each updated object
self.assertEqual(self.queue.count, 3)
for i, job in enumerate(self.queue.jobs):
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_update=True))
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 2'))
self.assertEqual(job.kwargs['event_type'], OBJECT_UPDATED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['data']['id'], data[i]['id'])
@ -294,7 +293,7 @@ class EventRuleTest(APITestCase):
# Verify that a task was queued for the deleted object
self.assertEqual(self.queue.count, 1)
job = self.queue.jobs[0]
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_delete=True))
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 3'))
self.assertEqual(job.kwargs['event_type'], OBJECT_DELETED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['data']['id'], site.pk)
@ -327,7 +326,7 @@ class EventRuleTest(APITestCase):
# Verify that a background task was queued for each deleted object
self.assertEqual(self.queue.count, 3)
for i, job in enumerate(self.queue.jobs):
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_delete=True))
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 3'))
self.assertEqual(job.kwargs['event_type'], OBJECT_DELETED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['data']['id'], sites[i].pk)
@ -342,7 +341,7 @@ class EventRuleTest(APITestCase):
A dummy implementation of Session.send() to be used for testing.
Always returns a 200 HTTP response.
"""
event = EventRule.objects.get(type_create=True)
event = EventRule.objects.get(name='Event Rule 1')
webhook = event.action_object
signature = generate_signature(request.body, webhook.secret)

55
netbox/extras/tests/test_filtersets.py
View File

@ -6,6 +6,7 @@ from django.test import TestCase
from circuits.models import Provider
from core.choices import ManagedFileRootPathChoices, ObjectChangeActionChoices
from core.events import *
from core.models import ObjectChange, ObjectType
from dcim.filtersets import SiteFilterSet
from dcim.models import DeviceRole, DeviceType, Manufacturer, Platform, Rack, Region, Site, SiteGroup
@ -251,7 +252,7 @@ class WebhookTestCase(TestCase, BaseFilterSetTests):
class EventRuleTestCase(TestCase, BaseFilterSetTests):
queryset = EventRule.objects.all()
filterset = EventRuleFilterSet
ignore_fields = ('action_data', 'conditions')
ignore_fields = ('action_data', 'conditions', 'event_types')
@classmethod
def setUpTestData(cls):
@ -292,11 +293,7 @@ class EventRuleTestCase(TestCase, BaseFilterSetTests):
name='Event Rule 1',
action_object=webhooks[0],
enabled=True,
type_create=True,
type_update=False,
type_delete=False,
type_job_start=False,
type_job_end=False,
event_types=[OBJECT_CREATED],
action_type=EventRuleActionChoices.WEBHOOK,
description='foobar1'
),
@ -304,11 +301,7 @@ class EventRuleTestCase(TestCase, BaseFilterSetTests):
name='Event Rule 2',
action_object=webhooks[1],
enabled=True,
type_create=False,
type_update=True,
type_delete=False,
type_job_start=False,
type_job_end=False,
event_types=[OBJECT_UPDATED],
action_type=EventRuleActionChoices.WEBHOOK,
description='foobar2'
),
@ -316,11 +309,7 @@ class EventRuleTestCase(TestCase, BaseFilterSetTests):
name='Event Rule 3',
action_object=webhooks[2],
enabled=False,
type_create=False,
type_update=False,
type_delete=True,
type_job_start=False,
type_job_end=False,
event_types=[OBJECT_DELETED],
action_type=EventRuleActionChoices.WEBHOOK,
description='foobar3'
),
@ -328,22 +317,14 @@ class EventRuleTestCase(TestCase, BaseFilterSetTests):
name='Event Rule 4',
action_object=scripts[0],
enabled=False,
type_create=False,
type_update=False,
type_delete=False,
type_job_start=True,
type_job_end=False,
event_types=[JOB_STARTED],
action_type=EventRuleActionChoices.SCRIPT,
),
EventRule(
name='Event Rule 5',
action_object=scripts[1],
enabled=False,
type_create=False,
type_update=False,
type_delete=False,
type_job_start=False,
type_job_end=True,
event_types=[JOB_COMPLETED],
action_type=EventRuleActionChoices.SCRIPT,
),
)
@ -384,25 +365,9 @@ class EventRuleTestCase(TestCase, BaseFilterSetTests):
params = {'enabled': False}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
def test_type_create(self):
params = {'type_create': True}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_type_update(self):
params = {'type_update': True}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_type_delete(self):
params = {'type_delete': True}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_type_job_start(self):
params = {'type_job_start': True}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_type_job_end(self):
params = {'type_job_end': True}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_event_type(self):
params = {'event_type': [OBJECT_CREATED, OBJECT_UPDATED]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
class CustomLinkTestCase(TestCase, ChangeLoggedFilterSetTests):

17
netbox/extras/tests/test_views.py
View File

@ -1,6 +1,7 @@
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from core.events import *
from core.models import ObjectType
from dcim.models import DeviceType, Manufacturer, Site
from extras.choices import *
@ -394,9 +395,9 @@ class EventRulesTestCase(ViewTestCases.PrimaryObjectViewTestCase):
site_type = ObjectType.objects.get_for_model(Site)
event_rules = (
EventRule(name='EventRule 1', type_create=True, action_object=webhooks[0]),
EventRule(name='EventRule 2', type_create=True, action_object=webhooks[1]),
EventRule(name='EventRule 3', type_create=True, action_object=webhooks[2]),
EventRule(name='EventRule 1', event_types=[OBJECT_CREATED], action_object=webhooks[0]),
EventRule(name='EventRule 2', event_types=[OBJECT_CREATED], action_object=webhooks[1]),
EventRule(name='EventRule 3', event_types=[OBJECT_CREATED], action_object=webhooks[2]),
)
for event in event_rules:
event.save()
@ -406,9 +407,7 @@ class EventRulesTestCase(ViewTestCases.PrimaryObjectViewTestCase):
cls.form_data = {
'name': 'Event X',
'object_types': [site_type.pk],
'type_create': False,
'type_update': True,
'type_delete': True,
'event_types': [OBJECT_UPDATED, OBJECT_DELETED],
'conditions': None,
'action_type': 'webhook',
'action_object_type': webhook_ct.pk,
@ -418,8 +417,8 @@ class EventRulesTestCase(ViewTestCases.PrimaryObjectViewTestCase):
}
cls.csv_data = (
"name,object_types,type_create,action_type,action_object",
"Webhook 4,dcim.site,True,webhook,Webhook 1",
f'name,object_types,event_types,action_type,action_object',
f'Webhook 4,dcim.site,"{OBJECT_CREATED},{OBJECT_UPDATED}",webhook,Webhook 1',
)
cls.csv_update_data = (
@ -430,7 +429,7 @@ class EventRulesTestCase(ViewTestCases.PrimaryObjectViewTestCase):
)
cls.bulk_edit_data = {
'type_update': True,
'description': 'New description',
}

9
netbox/extras/views.py
View File

@ -19,6 +19,7 @@ from extras.choices import LogLevelChoices
from extras.dashboard.forms import DashboardWidgetAddForm, DashboardWidgetForm
from extras.dashboard.utils import get_widget_class
from netbox.constants import DEFAULT_ACTION_PERMISSIONS
from netbox.registry import registry
from netbox.views import generic
from netbox.views.generic.mixins import TableMixin
from utilities.forms import ConfirmationForm, get_field_value
@ -550,6 +551,14 @@ class EventRuleListView(generic.ObjectListView):
class EventRuleView(generic.ObjectView):
queryset = EventRule.objects.all()
def get_extra_context(self, request, instance):
return {
'event_types': [
event for name, event in registry['events'].items()
if name in instance.event_types
]
}
@register_model_view(EventRule, 'edit')
class EventRuleEditView(generic.ObjectEditView):

30
netbox/netbox/events.py
View File

@ -13,11 +13,39 @@ __all__ = (
'EVENT_TYPE_SUCCESS',
'EVENT_TYPE_WARNING',
'Event',
'get_event',
'get_event_type_choices',
'get_event_text',
)
def get_event(name):
return registry['events'].get(name)
def get_event_text(name):
if event := registry['events'].get(name):
return event.text
return ''
def get_event_type_choices():
return [
(event.name, event.text) for event in registry['events'].values()
]
@dataclass
class Event:
"""
A type of event which can occur in NetBox. Event rules can be defined to automatically
perform some action in response to an event.
Args:
name: The unique name under which the event is registered.
text: The human-friendly event name. This should support translation.
type: The event's classification (info, success, warning, or danger). The default type is info.
"""
name: str
text: str
type: str = EVENT_TYPE_INFO
@ -26,6 +54,8 @@ class Event:
return self.text
def register(self):
if self.name in registry['events']:
raise Exception(f"An event named {self.name} has already been registered!")
registry['events'][self.name] = self
def color(self):

42
netbox/templates/extras/eventrule.html
View File

@ -34,29 +34,25 @@
</table>
</div>
<div class="card">
<h5 class="card-header">{% trans "Events" %}</h5>
<table class="table table-hover attr-table">
<tr>
<th scope="row">{% trans "Create" %}</th>
<td>{% checkmark object.type_create %}</td>
</tr>
<tr>
<th scope="row">{% trans "Update" %}</th>
<td>{% checkmark object.type_update %}</td>
</tr>
<tr>
<th scope="row">{% trans "Delete" %}</th>
<td>{% checkmark object.type_delete %}</td>
</tr>
<tr>
<th scope="row">{% trans "Job start" %}</th>
<td>{% checkmark object.type_job_start %}</td>
</tr>
<tr>
<th scope="row">{% trans "Job end" %}</th>
<td>{% checkmark object.type_job_end %}</td>
</tr>
</table>
<h5 class="card-header">{% trans "Event Types" %}</h5>
<ul class="list-group list-group-flush">
{% for name, event in registry.events.items %}
<li class="list-group-item">
<div class="row align-items-center">
<div class="col-auto">
{% if name in object.event_types %}
{% checkmark True %}
{% else %}
{{ ''|placeholder }}
{% endif %}
</div>
<div class="col">
{{ event }}
</div>
</div>
</li>
{% endfor %}
</ul>
</div>
{% plugin_left_page object %}
</div>

5
netbox/utilities/testing/base.py
View File

@ -137,7 +137,10 @@ class ModelTestCase(TestCase):
# Convert ArrayFields to CSV strings
if type(field) is ArrayField:
if type(field.base_field) is ArrayField:
if getattr(field.base_field, 'choices', None):
# Values for fields with pre-defined choices can be returned as lists
model_dict[key] = value
elif type(field.base_field) is ArrayField:
# Handle nested arrays (e.g. choice sets)
model_dict[key] = '\n'.join([f'{k},{v}' for k, v in value])
elif issubclass(type(field.base_field), RangeField):