NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-08-08 08:38:16 -06:00
Code Issues Actions 15 Packages Projects Releases Wiki Activity

Initial work on #16886

Browse Source
This commit is contained in:
Jeremy Stretch 2024-07-17 21:02:12 -04:00
parent b2c5a4639c
commit 8e2f459d10
23 changed files with 238 additions and 245 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -0,0 +1,3 @@
# Events
TODO

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

@ -12,16 +12,18 @@ __all__ = (
'OBJECT_UPDATED',
)
NAMESPACE = 'netbox'
# Object events
OBJECT_CREATED = 'object_created'
OBJECT_UPDATED = 'object_updated'
OBJECT_DELETED = 'object_deleted'
OBJECT_CREATED = f'{NAMESPACE}.object_created'
OBJECT_UPDATED = f'{NAMESPACE}.object_updated'
OBJECT_DELETED = f'{NAMESPACE}.object_deleted'
# Job events
JOB_STARTED = 'job_started'
JOB_COMPLETED = 'job_completed'
JOB_FAILED = 'job_failed'
JOB_ERRORED = 'job_errored'
JOB_STARTED = f'{NAMESPACE}.job_started'
JOB_COMPLETED = f'{NAMESPACE}.job_completed'
JOB_FAILED = f'{NAMESPACE}.job_failed'
JOB_ERRORED = f'{NAMESPACE}.job_errored'
# Register core events
Event(name=OBJECT_CREATED, text=_('Object created')).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')

2
netbox/extras/events.py
View File

@ -176,7 +176,7 @@ def process_event_queue(events):
# 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},
event_types__contains=[event['event_type']],
object_types=object_type,
enabled=True
)

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(

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

@ -1,4 +1,5 @@
from django import forms
from django.contrib.postgres.forms import SimpleArrayField
from django.utils.translation import gettext_lazy as _
from extras.choices import *
@ -248,33 +249,18 @@ class EventRuleBulkEditForm(NetBoxModelBulkEditForm):
required=False,
widget=BulkEditNullBooleanSelect()
)
type_create = forms.NullBooleanField(
label=_('On create'),
required=False,
widget=BulkEditNullBooleanSelect()
event_types = SimpleArrayField(
label=_('Event types'),
base_field=forms.CharField(),
required=False
)
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):

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

@ -196,8 +196,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):

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

@ -2,6 +2,7 @@ import json
import re
from django import forms
from django.contrib.postgres.forms import SimpleArrayField
from django.utils.safestring import mark_safe
from django.utils.translation import gettext_lazy as _
@ -303,6 +304,10 @@ class EventRuleForm(NetBoxModelForm):
label=_('Object types'),
queryset=ObjectType.objects.with_feature('event_rules'),
)
event_types = SimpleArrayField(
label=_('Event types'),
base_field=forms.CharField()
)
action_choice = forms.ChoiceField(
label=_('Action choice'),
choices=[]
@ -319,7 +324,7 @@ 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('event_types', name=_('Event Types')),
FieldSet('conditions', name=_('Conditions')),
FieldSet('action_type', 'action_choice', 'action_data', name=_('Action')),
)
@ -327,17 +332,9 @@ class EventRuleForm(NetBoxModelForm):
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(),

10
netbox/extras/graphql/filters.py
View File

@ -9,7 +9,7 @@ __all__ = (
'CustomFieldFilter',
'CustomFieldChoiceSetFilter',
'CustomLinkFilter',
'EventRuleFilter',
# 'EventRuleFilter',
'ExportTemplateFilter',
'ImageAttachmentFilter',
'JournalEntryFilter',
@ -92,7 +92,7 @@ class WebhookFilter(BaseFilterMixin):
pass
@strawberry_django.filter(models.EventRule, lookups=True)
@autotype_decorator(filtersets.EventRuleFilterSet)
class EventRuleFilter(BaseFilterMixin):
pass
# @strawberry_django.filter(models.EventRule, lookups=True)
# @autotype_decorator(filtersets.EventRuleFilterSet)
# class EventRuleFilter(BaseFilterMixin):
# pass

3
netbox/extras/graphql/types.py
View File

@ -182,7 +182,8 @@ class WebhookType(OrganizationalObjectType):
@strawberry_django.type(
models.EventRule,
exclude=['content_types',],
filters=EventRuleFilter
# TODO: Fix GraphQL filter
# filters=EventRuleFilter
)
class EventRuleType(OrganizationalObjectType):
action_object_type: Annotated["ContentTypeType", strawberry.lazy('netbox.graphql.types')] | None

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',
),
]

32
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
@ -60,30 +61,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),
help_text=_("The types of event which will trigger this rule.")
)
enabled = models.BooleanField(
verbose_name=_('enabled'),
@ -145,9 +125,7 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, ChangeLogged
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
]):
if not self.event_types:
raise ValidationError(
_("At least one event type must be selected: create, update, delete, job start, and/or job end.")
)

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': ','.join([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}",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):

21
netbox/netbox/events.py
View File

@ -13,9 +13,28 @@ __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:
name: str
@ -26,6 +45,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>