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

Merge branch 'feature' into 15692-background-jobs

Browse Source
This commit is contained in:
Jeremy Stretch 2024-07-24 14:39:48 -04:00
commit 1264302877
54 changed files with 1419 additions and 294 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
docs/models/circuits/circuitgroup.md Normal file
View File

@ -0,0 +1,13 @@
# Circuit Groups
[Circuits](./circuit.md) can be arranged into administrative groups for organization. The assignment of a circuit to a group is optional.
## Fields
### Name
A unique human-friendly name.
### Slug
A unique URL-friendly identifier. (This value can be used for filtering.)

25
docs/models/circuits/circuitgroupassignment.md Normal file
View File

@ -0,0 +1,25 @@
# Circuit Group Assignments
Circuits can be assigned to [circuit groups](./circuitgroup.md) for correlation purposes. For instance, three circuits, each belonging to a different provider, may each be assigned to the same circuit group. Each assignment may optionally include a priority designation.
## Fields
### Group
The [circuit group](./circuitgroup.md) being assigned.
### Circuit
The [circuit](./circuit.md) that is being assigned to the group.
### Priority
The circuit's operation priority relative to its peers within the group. The assignment of a priority is optional. Choices include:
* Primary
* Secondary
* Tertiary
* Inactive
!!! tip
Additional priority choices may be defined by setting `CircuitGroupAssignment.priority` under the [`FIELD_CHOICES`](../../configuration/data-validation.md#field_choices) configuration parameter.

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

3
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'
@ -163,6 +164,8 @@ nav:
- Data Model:
- Circuits:
- Circuit: 'models/circuits/circuit.md'
- CircuitGroup: 'models/circuits/circuitgroup.md'
- CircuitGroupAssignment: 'models/circuits/circuitgroupassignment.md'
- Circuit Termination: 'models/circuits/circuittermination.md'
- Circuit Type: 'models/circuits/circuittype.md'
- Provider: 'models/circuits/provider.md'

48
netbox/circuits/api/serializers_/circuits.py
View File

@ -1,7 +1,7 @@
from rest_framework import serializers
from circuits.choices import CircuitStatusChoices
from circuits.models import Circuit, CircuitTermination, CircuitType
from circuits.choices import CircuitPriorityChoices, CircuitStatusChoices
from circuits.models import Circuit, CircuitGroup, CircuitGroupAssignment, CircuitTermination, CircuitType
from dcim.api.serializers_.cables import CabledObjectSerializer
from dcim.api.serializers_.sites import SiteSerializer
from netbox.api.fields import ChoiceField, RelatedObjectCountField
@ -12,6 +12,8 @@ from .providers import ProviderAccountSerializer, ProviderNetworkSerializer, Pro
__all__ = (
'CircuitSerializer',
'CircuitGroupAssignmentSerializer',
'CircuitGroupSerializer',
'CircuitTerminationSerializer',
'CircuitTypeSerializer',
)
@ -43,6 +45,34 @@ class CircuitCircuitTerminationSerializer(WritableNestedSerializer):
]
class CircuitGroupSerializer(NetBoxModelSerializer):
tenant = TenantSerializer(nested=True, required=False, allow_null=True)
circuit_count = RelatedObjectCountField('assignments')
class Meta:
model = CircuitGroup
fields = [
'id', 'url', 'display_url', 'display', 'name', 'slug', 'description', 'tenant',
'tags', 'custom_fields', 'created', 'last_updated', 'circuit_count'
]
brief_fields = ('id', 'url', 'display', 'name')
class CircuitGroupAssignmentSerializer_(NetBoxModelSerializer):
"""
Base serializer for group assignments under CircuitSerializer.
"""
group = CircuitGroupSerializer(nested=True)
priority = ChoiceField(choices=CircuitPriorityChoices, allow_blank=True, required=False)
class Meta:
model = CircuitGroupAssignment
fields = [
'id', 'url', 'display_url', 'display', 'group', 'priority', 'tags', 'created', 'last_updated',
]
brief_fields = ('id', 'url', 'display', 'group', 'priority')
class CircuitSerializer(NetBoxModelSerializer):
provider = ProviderSerializer(nested=True)
provider_account = ProviderAccountSerializer(nested=True, required=False, allow_null=True, default=None)
@ -51,13 +81,14 @@ class CircuitSerializer(NetBoxModelSerializer):
tenant = TenantSerializer(nested=True, required=False, allow_null=True)
termination_a = CircuitCircuitTerminationSerializer(read_only=True, allow_null=True)
termination_z = CircuitCircuitTerminationSerializer(read_only=True, allow_null=True)
assignments = CircuitGroupAssignmentSerializer_(nested=True, many=True, required=False)
class Meta:
model = Circuit
fields = [
'id', 'url', 'display_url', 'display', 'cid', 'provider', 'provider_account', 'type', 'status', 'tenant',
'install_date', 'termination_date', 'commit_rate', 'description', 'termination_a', 'termination_z',
'comments', 'tags', 'custom_fields', 'created', 'last_updated',
'comments', 'tags', 'custom_fields', 'created', 'last_updated', 'assignments',
]
brief_fields = ('id', 'url', 'display', 'cid', 'description')
@ -75,3 +106,14 @@ class CircuitTerminationSerializer(NetBoxModelSerializer, CabledObjectSerializer
'link_peers', 'link_peers_type', 'tags', 'custom_fields', 'created', 'last_updated', '_occupied',
]
brief_fields = ('id', 'url', 'display', 'circuit', 'term_side', 'description', 'cable', '_occupied')
class CircuitGroupAssignmentSerializer(CircuitGroupAssignmentSerializer_):
circuit = CircuitSerializer(nested=True)
class Meta:
model = CircuitGroupAssignment
fields = [
'id', 'url', 'display_url', 'display', 'group', 'circuit', 'priority', 'tags', 'created', 'last_updated',
]
brief_fields = ('id', 'url', 'display', 'group', 'circuit', 'priority')

2
netbox/circuits/api/urls.py
View File

@ -14,6 +14,8 @@ router.register('provider-networks', views.ProviderNetworkViewSet)
router.register('circuit-types', views.CircuitTypeViewSet)
router.register('circuits', views.CircuitViewSet)
router.register('circuit-terminations', views.CircuitTerminationViewSet)
router.register('circuit-groups', views.CircuitGroupViewSet)
router.register('circuit-group-assignments', views.CircuitGroupAssignmentViewSet)
app_name = 'circuits-api'
urlpatterns = router.urls

20
netbox/circuits/api/views.py
View File

@ -55,6 +55,26 @@ class CircuitTerminationViewSet(PassThroughPortMixin, NetBoxModelViewSet):
filterset_class = filtersets.CircuitTerminationFilterSet
#
# Circuit Groups
#
class CircuitGroupViewSet(NetBoxModelViewSet):
queryset = CircuitGroup.objects.all()
serializer_class = serializers.CircuitGroupSerializer
filterset_class = filtersets.CircuitGroupFilterSet
#
# Circuit Group Assignments
#
class CircuitGroupAssignmentViewSet(NetBoxModelViewSet):
queryset = CircuitGroupAssignment.objects.all()
serializer_class = serializers.CircuitGroupAssignmentSerializer
filterset_class = filtersets.CircuitGroupAssignmentFilterSet
#
# Provider accounts
#

16
netbox/circuits/choices.py
View File

@ -76,3 +76,19 @@ class CircuitTerminationPortSpeedChoices(ChoiceSet):
(1544, 'T1 (1.544 Mbps)'),
(2048, 'E1 (2.048 Mbps)'),
]
class CircuitPriorityChoices(ChoiceSet):
key = 'CircuitGroupAssignment.priority'
PRIORITY_PRIMARY = 'primary'
PRIORITY_SECONDARY = 'secondary'
PRIORITY_TERTIARY = 'tertiary'
PRIORITY_INACTIVE = 'inactive'
CHOICES = [
(PRIORITY_PRIMARY, _('Primary')),
(PRIORITY_SECONDARY, _('Secondary')),
(PRIORITY_TERTIARY, _('Tertiary')),
(PRIORITY_INACTIVE, _('Inactive')),
]

42
netbox/circuits/filtersets.py
View File

@ -13,6 +13,8 @@ from .models import *
__all__ = (
'CircuitFilterSet',
'CircuitGroupAssignmentFilterSet',
'CircuitGroupFilterSet',
'CircuitTerminationFilterSet',
'CircuitTypeFilterSet',
'ProviderNetworkFilterSet',
@ -303,3 +305,43 @@ class CircuitTerminationFilterSet(NetBoxModelFilterSet, CabledObjectFilterSet):
Q(pp_info__icontains=value) |
Q(description__icontains=value)
).distinct()
class CircuitGroupFilterSet(OrganizationalModelFilterSet, TenancyFilterSet):
class Meta:
model = CircuitGroup
fields = ('id', 'name', 'slug', 'description')
class CircuitGroupAssignmentFilterSet(NetBoxModelFilterSet):
q = django_filters.CharFilter(
method='search',
label=_('Search'),
)
circuit_id = django_filters.ModelMultipleChoiceFilter(
queryset=Circuit.objects.all(),
label=_('Circuit'),
)
group_id = django_filters.ModelMultipleChoiceFilter(
queryset=CircuitGroup.objects.all(),
label=_('Circuit group (ID)'),
)
group = django_filters.ModelMultipleChoiceFilter(
field_name='group__slug',
queryset=CircuitGroup.objects.all(),
to_field_name='slug',
label=_('Circuit group (slug)'),
)
class Meta:
model = CircuitGroupAssignment
fields = ('id', 'circuit', 'group', 'priority')
def search(self, queryset, name, value):
if not value.strip():
return queryset
return queryset.filter(
Q(circuit__cid__icontains=value) |
Q(group__name__icontains=value)
)

41
netbox/circuits/forms/bulk_edit.py
View File

@ -1,7 +1,7 @@
from django import forms
from django.utils.translation import gettext_lazy as _
from circuits.choices import CircuitCommitRateChoices, CircuitStatusChoices
from circuits.choices import CircuitCommitRateChoices, CircuitPriorityChoices, CircuitStatusChoices
from circuits.models import *
from dcim.models import Site
from ipam.models import ASN
@ -14,6 +14,8 @@ from utilities.forms.widgets import BulkEditNullBooleanSelect, DatePicker, Numbe
__all__ = (
'CircuitBulkEditForm',
'CircuitGroupAssignmentBulkEditForm',
'CircuitGroupBulkEditForm',
'CircuitTerminationBulkEditForm',
'CircuitTypeBulkEditForm',
'ProviderBulkEditForm',
@ -219,3 +221,40 @@ class CircuitTerminationBulkEditForm(NetBoxModelBulkEditForm):
FieldSet('port_speed', 'upstream_speed', name=_('Termination Details')),
)
nullable_fields = ('description')
class CircuitGroupBulkEditForm(NetBoxModelBulkEditForm):
description = forms.CharField(
label=_('Description'),
max_length=200,
required=False
)
tenant = DynamicModelChoiceField(
label=_('Tenant'),
queryset=Tenant.objects.all(),
required=False
)
model = CircuitGroup
nullable_fields = (
'description', 'tenant',
)
class CircuitGroupAssignmentBulkEditForm(NetBoxModelBulkEditForm):
circuit = DynamicModelChoiceField(
label=_('Circuit'),
queryset=Circuit.objects.all(),
required=False
)
priority = forms.ChoiceField(
label=_('Priority'),
choices=add_blank_choice(CircuitPriorityChoices),
required=False
)
model = CircuitGroupAssignment
fieldsets = (
FieldSet('circuit', 'priority'),
)
nullable_fields = ('priority',)

23
netbox/circuits/forms/bulk_import.py
View File

@ -11,6 +11,8 @@ from utilities.forms.fields import CSVChoiceField, CSVModelChoiceField, SlugFiel
__all__ = (
'CircuitImportForm',
'CircuitGroupAssignmentImportForm',
'CircuitGroupImportForm',
'CircuitTerminationImportForm',
'CircuitTerminationImportRelatedForm',
'CircuitTypeImportForm',
@ -150,3 +152,24 @@ class CircuitTerminationImportForm(NetBoxModelImportForm, BaseCircuitTermination
'circuit', 'term_side', 'site', 'provider_network', 'port_speed', 'upstream_speed', 'xconnect_id',
'pp_info', 'description', 'tags'
]
class CircuitGroupImportForm(NetBoxModelImportForm):
tenant = CSVModelChoiceField(
label=_('Tenant'),
queryset=Tenant.objects.all(),
required=False,
to_field_name='name',
help_text=_('Assigned tenant')
)
class Meta:
model = CircuitGroup
fields = ('name', 'slug', 'description', 'tenant', 'tags')
class CircuitGroupAssignmentImportForm(NetBoxModelImportForm):
class Meta:
model = CircuitGroupAssignment
fields = ('circuit', 'group', 'priority')

37
netbox/circuits/forms/filtersets.py
View File

@ -1,7 +1,7 @@
from django import forms
from django.utils.translation import gettext as _
from circuits.choices import CircuitCommitRateChoices, CircuitStatusChoices, CircuitTerminationSideChoices
from circuits.choices import CircuitCommitRateChoices, CircuitPriorityChoices, CircuitStatusChoices, CircuitTerminationSideChoices
from circuits.models import *
from dcim.models import Region, Site, SiteGroup
from ipam.models import ASN
@ -13,6 +13,8 @@ from utilities.forms.widgets import DatePicker, NumberWithOptions
__all__ = (
'CircuitFilterForm',
'CircuitGroupAssignmentFilterForm',
'CircuitGroupFilterForm',
'CircuitTerminationFilterForm',
'CircuitTypeFilterForm',
'ProviderFilterForm',
@ -230,3 +232,36 @@ class CircuitTerminationFilterForm(NetBoxModelFilterSetForm):
label=_('Provider')
)
tag = TagFilterField(model)
class CircuitGroupFilterForm(TenancyFilterForm, NetBoxModelFilterSetForm):
model = CircuitGroup
fieldsets = (
FieldSet('q', 'filter_id', 'tag'),
FieldSet('tenant_group_id', 'tenant_id', name=_('Tenant')),
)
tag = TagFilterField(model)
class CircuitGroupAssignmentFilterForm(NetBoxModelFilterSetForm):
model = CircuitGroupAssignment
fieldsets = (
FieldSet('q', 'filter_id', 'tag'),
FieldSet('circuit_id', 'group_id', 'priority', name=_('Assignment')),
)
circuit_id = DynamicModelMultipleChoiceField(
queryset=Circuit.objects.all(),
required=False,
label=_('Circuit')
)
group_id = DynamicModelMultipleChoiceField(
queryset=CircuitGroup.objects.all(),
required=False,
label=_('Group')
)
priority = forms.MultipleChoiceField(
label=_('Priority'),
choices=CircuitPriorityChoices,
required=False
)
tag = TagFilterField(model)

34
netbox/circuits/forms/model_forms.py
View File

@ -12,6 +12,8 @@ from utilities.forms.widgets import DatePicker, NumberWithOptions
__all__ = (
'CircuitForm',
'CircuitGroupAssignmentForm',
'CircuitGroupForm',
'CircuitTerminationForm',
'CircuitTypeForm',
'ProviderForm',
@ -171,3 +173,35 @@ class CircuitTerminationForm(NetBoxModelForm):
options=CircuitTerminationPortSpeedChoices
),
}
class CircuitGroupForm(TenancyForm, NetBoxModelForm):
slug = SlugField()
fieldsets = (
FieldSet('name', 'slug', 'description', 'tags', name=_('Circuit Group')),
FieldSet('tenant_group', 'tenant', name=_('Tenancy')),
)
class Meta:
model = CircuitGroup
fields = [
'name', 'slug', 'description', 'tenant_group', 'tenant', 'tags',
]
class CircuitGroupAssignmentForm(NetBoxModelForm):
group = DynamicModelChoiceField(
label=_('Group'),
queryset=CircuitGroup.objects.all(),
)
circuit = DynamicModelChoiceField(
label=_('Circuit'),
queryset=Circuit.objects.all(),
)
class Meta:
model = CircuitGroupAssignment
fields = [
'group', 'circuit', 'priority', 'tags',
]

14
netbox/circuits/graphql/filters.py
View File

@ -7,6 +7,8 @@ from netbox.graphql.filter_mixins import autotype_decorator, BaseFilterMixin
__all__ = (
'CircuitTerminationFilter',
'CircuitFilter',
'CircuitGroupAssignmentFilter',
'CircuitGroupFilter',
'CircuitTypeFilter',
'ProviderFilter',
'ProviderAccountFilter',
@ -32,6 +34,18 @@ class CircuitTypeFilter(BaseFilterMixin):
pass
@strawberry_django.filter(models.CircuitGroup, lookups=True)
@autotype_decorator(filtersets.CircuitGroupFilterSet)
class CircuitGroupFilter(BaseFilterMixin):
pass
@strawberry_django.filter(models.CircuitGroupAssignment, lookups=True)
@autotype_decorator(filtersets.CircuitGroupAssignmentFilterSet)
class CircuitGroupAssignmentFilter(BaseFilterMixin):
pass
@strawberry_django.filter(models.Provider, lookups=True)
@autotype_decorator(filtersets.ProviderFilterSet)
class ProviderFilter(BaseFilterMixin):

10
netbox/circuits/graphql/schema.py
View File

@ -24,6 +24,16 @@ class CircuitsQuery:
return models.CircuitType.objects.get(pk=id)
circuit_type_list: List[CircuitTypeType] = strawberry_django.field()
@strawberry.field
def circuit_group(self, id: int) -> CircuitGroupType:
return models.CircuitGroup.objects.get(pk=id)
circuit_group_list: List[CircuitGroupType] = strawberry_django.field()
@strawberry.field
def circuit_group_assignment(self, id: int) -> CircuitGroupAssignmentType:
return models.CircuitGroupAssignment.objects.get(pk=id)
circuit_group_assignment_list: List[CircuitGroupAssignmentType] = strawberry_django.field()
@strawberry.field
def provider(self, id: int) -> ProviderType:
return models.Provider.objects.get(pk=id)

23
netbox/circuits/graphql/types.py
View File

@ -6,13 +6,15 @@ import strawberry_django
from circuits import models
from dcim.graphql.mixins import CabledObjectMixin
from extras.graphql.mixins import ContactsMixin, CustomFieldsMixin, TagsMixin
from netbox.graphql.types import NetBoxObjectType, ObjectType, OrganizationalObjectType
from netbox.graphql.types import BaseObjectType, NetBoxObjectType, ObjectType, OrganizationalObjectType
from tenancy.graphql.types import TenantType
from .filters import *
__all__ = (
'CircuitTerminationType',
'CircuitType',
'CircuitGroupAssignmentType',
'CircuitGroupType',
'CircuitTypeType',
'ProviderType',
'ProviderAccountType',
@ -91,3 +93,22 @@ class CircuitType(NetBoxObjectType, ContactsMixin):
tenant: TenantType | None
terminations: List[CircuitTerminationType]
@strawberry_django.type(
models.CircuitGroup,
fields='__all__',
filters=CircuitGroupFilter
)
class CircuitGroupType(OrganizationalObjectType):
tenant: TenantType | None
@strawberry_django.type(
models.CircuitGroupAssignment,
fields='__all__',
filters=CircuitGroupAssignmentFilter
)
class CircuitGroupAssignmentType(TagsMixin, BaseObjectType):
group: Annotated["CircuitGroupType", strawberry.lazy('circuits.graphql.types')]
circuit: Annotated["CircuitType", strawberry.lazy('circuits.graphql.types')]

90
netbox/circuits/migrations/0044_circuitgroup_circuitgroupassignment_and_more.py Normal file
View File

@ -0,0 +1,90 @@
# Generated by Django 5.0.7 on 2024-07-22 06:27
import django.db.models.deletion
import taggit.managers
import utilities.json
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('circuits', '0043_circuittype_color'),
('extras', '0118_notifications'),
('tenancy', '0015_contactassignment_rename_content_type'),
]
operations = [
migrations.CreateModel(
name='CircuitGroup',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False)),
('created', models.DateTimeField(auto_now_add=True, null=True)),
('last_updated', models.DateTimeField(auto_now=True, null=True)),
(
'custom_field_data',
models.JSONField(blank=True, default=dict, encoder=utilities.json.CustomFieldJSONEncoder),
),
('name', models.CharField(max_length=100, unique=True)),
('slug', models.SlugField(max_length=100, unique=True)),
('description', models.CharField(blank=True, max_length=200)),
('tags', taggit.managers.TaggableManager(through='extras.TaggedItem', to='extras.Tag')),
(
'tenant',
models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.PROTECT,
related_name='circuit_groups',
to='tenancy.tenant',
),
),
],
options={
'verbose_name': 'Circuit group',
'verbose_name_plural': 'Circuit group',
'ordering': ('name',),
},
),
migrations.CreateModel(
name='CircuitGroupAssignment',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False)),
('created', models.DateTimeField(auto_now_add=True, null=True)),
('last_updated', models.DateTimeField(auto_now=True, null=True)),
(
'custom_field_data',
models.JSONField(blank=True, default=dict, encoder=utilities.json.CustomFieldJSONEncoder),
),
('priority', models.CharField(blank=True, max_length=50)),
(
'circuit',
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name='assignments',
to='circuits.circuit',
),
),
(
'group',
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name='assignments',
to='circuits.circuitgroup',
),
),
('tags', taggit.managers.TaggableManager(through='extras.TaggedItem', to='extras.Tag')),
],
options={
'verbose_name': 'Circuit group assignment',
'verbose_name_plural': 'Circuit group assignments',
'ordering': ('circuit', 'priority', 'pk'),
},
),
migrations.AddConstraint(
model_name='circuitgroupassignment',
constraint=models.UniqueConstraint(
fields=('circuit', 'group'), name='circuits_circuitgroupassignment_unique_circuit_group'
),
),
]

73
netbox/circuits/models/circuits.py
View File

@ -6,11 +6,13 @@ from django.utils.translation import gettext_lazy as _
from circuits.choices import *
from dcim.models import CabledObjectModel
from netbox.models import ChangeLoggedModel, OrganizationalModel, PrimaryModel
from netbox.models.features import ContactsMixin, CustomFieldsMixin, CustomLinksMixin, ImageAttachmentsMixin, TagsMixin
from netbox.models.features import ContactsMixin, CustomFieldsMixin, CustomLinksMixin, ExportTemplatesMixin, ImageAttachmentsMixin, TagsMixin
from utilities.fields import ColorField
__all__ = (
'Circuit',
'CircuitGroup',
'CircuitGroupAssignment',
'CircuitTermination',
'CircuitType',
)
@ -151,6 +153,75 @@ class Circuit(ContactsMixin, ImageAttachmentsMixin, PrimaryModel):
raise ValidationError({'provider_account': "The assigned account must belong to the assigned provider."})
class CircuitGroup(OrganizationalModel):
"""
An administrative grouping of Circuits.
"""
tenant = models.ForeignKey(
to='tenancy.Tenant',
on_delete=models.PROTECT,
related_name='circuit_groups',
blank=True,
null=True
)
class Meta:
ordering = ('name',)
verbose_name = _('circuit group')
verbose_name_plural = _('circuit groups')
def __str__(self):
return self.name
def get_absolute_url(self):
return reverse('circuits:circuitgroup', args=[self.pk])
class CircuitGroupAssignment(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, ChangeLoggedModel):
"""
Assignment of a Circuit to a CircuitGroup with an optional priority.
"""
circuit = models.ForeignKey(
Circuit,
on_delete=models.CASCADE,
related_name='assignments'
)
group = models.ForeignKey(
CircuitGroup,
on_delete=models.CASCADE,
related_name='assignments'
)
priority = models.CharField(
verbose_name=_('priority'),
max_length=50,
choices=CircuitPriorityChoices,
blank=True
)
prerequisite_models = (
'circuits.Circuit',
'circuits.CircuitGroup',
)
class Meta:
ordering = ('circuit', 'priority', 'pk')
constraints = (
models.UniqueConstraint(
fields=('circuit', 'group'),
name='%(app_label)s_%(class)s_unique_circuit_group'
),
)
verbose_name = _('Circuit group assignment')
verbose_name_plural = _('Circuit group assignments')
def __str__(self):
if self.priority:
return f"{self.group} ({self.get_priority_display()})"
return str(self.group)
def get_absolute_url(self):
return reverse('circuits:circuitgroupassignment', args=[self.pk])
class CircuitTermination(
CustomFieldsMixin,
CustomLinksMixin,

11
netbox/circuits/search.py
View File

@ -13,6 +13,17 @@ class CircuitIndex(SearchIndex):
display_attrs = ('provider', 'provider_account', 'type', 'status', 'tenant', 'description')
@register_search
class CircuitGroupIndex(SearchIndex):
model = models.CircuitGroup
fields = (
('name', 100),
('slug', 110),
('description', 500),
)
display_attrs = ('description',)
@register_search
class CircuitTerminationIndex(SearchIndex):
model = models.CircuitTermination

49
netbox/circuits/tables/circuits.py
View File

@ -9,6 +9,8 @@ from netbox.tables import NetBoxTable, columns
from .columns import CommitRateColumn
__all__ = (
'CircuitGroupAssignmentTable',
'CircuitGroupTable',
'CircuitTable',
'CircuitTerminationTable',
'CircuitTypeTable',
@ -119,3 +121,50 @@ class CircuitTerminationTable(NetBoxTable):
'xconnect_id', 'pp_info', 'description', 'created', 'last_updated', 'actions',
)
default_columns = ('pk', 'id', 'circuit', 'provider', 'term_side', 'description')
class CircuitGroupTable(NetBoxTable):
name = tables.Column(
verbose_name=_('Name'),
linkify=True
)
circuit_group_assignment_count = columns.LinkedCountColumn(
viewname='circuits:circuitgroupassignment_list',
url_params={'group_id': 'pk'},
verbose_name=_('Circuits')
)
tags = columns.TagColumn(
url_name='circuits:circuitgroup_list'
)
class Meta(NetBoxTable.Meta):
model = CircuitGroup
fields = (
'pk', 'name', 'description', 'circuit_group_assignment_count', 'tags',
'created', 'last_updated', 'actions',
)
default_columns = ('pk', 'name', 'description', 'circuit_group_assignment_count')
class CircuitGroupAssignmentTable(NetBoxTable):
group = tables.Column(
verbose_name=_('Group'),
linkify=True
)
circuit = tables.Column(
verbose_name=_('Circuit'),
linkify=True
)
priority = tables.Column(
verbose_name=_('Priority'),
)
tags = columns.TagColumn(
url_name='circuits:circuitgroupassignment_list'
)
class Meta(NetBoxTable.Meta):
model = CircuitGroupAssignment
fields = (
'pk', 'id', 'group', 'circuit', 'priority', 'created', 'last_updated', 'actions', 'tags',
)
default_columns = ('pk', 'group', 'circuit', 'priority')

103
netbox/circuits/tests/test_api.py
View File

@ -206,6 +206,38 @@ class CircuitTerminationTest(APIViewTestCases.APIViewTestCase):
}
class CircuitGroupTest(APIViewTestCases.APIViewTestCase):
model = CircuitGroup
brief_fields = ['display', 'id', 'name', 'url']
bulk_update_data = {
'description': 'New description',
}
@classmethod
def setUpTestData(cls):
circuit_groups = (
CircuitGroup(name="Circuit Group 1", slug='circuit-group-1'),
CircuitGroup(name="Circuit Group 2", slug='circuit-group-2'),
CircuitGroup(name="Circuit Group 3", slug='circuit-group-3'),
)
CircuitGroup.objects.bulk_create(circuit_groups)
cls.create_data = [
{
'name': 'Circuit Group 4',
'slug': 'circuit-group-4',
},
{
'name': 'Circuit Group 5',
'slug': 'circuit-group-5',
},
{
'name': 'Circuit Group 6',
'slug': 'circuit-group-6',
},
]
class ProviderAccountTest(APIViewTestCases.APIViewTestCase):
model = ProviderAccount
brief_fields = ['account', 'description', 'display', 'id', 'name', 'url']
@ -249,6 +281,77 @@ class ProviderAccountTest(APIViewTestCases.APIViewTestCase):
}
class CircuitGroupAssignmentTest(APIViewTestCases.APIViewTestCase):
model = CircuitGroupAssignment
brief_fields = ['circuit', 'display', 'group', 'id', 'priority', 'url']
bulk_update_data = {
'priority': CircuitPriorityChoices.PRIORITY_INACTIVE,
}
@classmethod
def setUpTestData(cls):
circuit_groups = (
CircuitGroup(name='Circuit Group 1', slug='circuit-group-1'),
CircuitGroup(name='Circuit Group 2', slug='circuit-group-2'),
CircuitGroup(name='Circuit Group 3', slug='circuit-group-3'),
CircuitGroup(name='Circuit Group 4', slug='circuit-group-4'),
CircuitGroup(name='Circuit Group 5', slug='circuit-group-5'),
CircuitGroup(name='Circuit Group 6', slug='circuit-group-6'),
)
CircuitGroup.objects.bulk_create(circuit_groups)
provider = Provider.objects.create(name='Provider 1', slug='provider-1')
circuittype = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1')
circuits = (
Circuit(cid='Circuit 1', provider=provider, type=circuittype),
Circuit(cid='Circuit 2', provider=provider, type=circuittype),
Circuit(cid='Circuit 3', provider=provider, type=circuittype),
Circuit(cid='Circuit 4', provider=provider, type=circuittype),
Circuit(cid='Circuit 5', provider=provider, type=circuittype),
Circuit(cid='Circuit 6', provider=provider, type=circuittype),
)
Circuit.objects.bulk_create(circuits)
assignments = (
CircuitGroupAssignment(
group=circuit_groups[0],
circuit=circuits[0],
priority=CircuitPriorityChoices.PRIORITY_PRIMARY
),
CircuitGroupAssignment(
group=circuit_groups[1],
circuit=circuits[1],
priority=CircuitPriorityChoices.PRIORITY_SECONDARY
),
CircuitGroupAssignment(
group=circuit_groups[2],
circuit=circuits[2],
priority=CircuitPriorityChoices.PRIORITY_TERTIARY
),
)
CircuitGroupAssignment.objects.bulk_create(assignments)
cls.create_data = [
{
'group': circuit_groups[3].pk,
'circuit': circuits[3].pk,
'priority': CircuitPriorityChoices.PRIORITY_PRIMARY,
},
{
'group': circuit_groups[4].pk,
'circuit': circuits[4].pk,
'priority': CircuitPriorityChoices.PRIORITY_SECONDARY,
},
{
'group': circuit_groups[5].pk,
'circuit': circuits[5].pk,
'priority': CircuitPriorityChoices.PRIORITY_TERTIARY,
},
]
class ProviderNetworkTest(APIViewTestCases.APIViewTestCase):
model = ProviderNetwork
brief_fields = ['description', 'display', 'id', 'name', 'url']

116
netbox/circuits/tests/test_filtersets.py
View File

@ -451,6 +451,122 @@ class CircuitTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 7)
class CircuitGroupTestCase(TestCase, ChangeLoggedFilterSetTests):
queryset = CircuitGroup.objects.all()
filterset = CircuitGroupFilterSet
@classmethod
def setUpTestData(cls):
tenant_groups = (
TenantGroup(name='Tenant group 1', slug='tenant-group-1'),
TenantGroup(name='Tenant group 2', slug='tenant-group-2'),
TenantGroup(name='Tenant group 3', slug='tenant-group-3'),
)
for tenantgroup in tenant_groups:
tenantgroup.save()
tenants = (
Tenant(name='Tenant 1', slug='tenant-1', group=tenant_groups[0]),
Tenant(name='Tenant 2', slug='tenant-2', group=tenant_groups[1]),
Tenant(name='Tenant 3', slug='tenant-3', group=tenant_groups[2]),
)
Tenant.objects.bulk_create(tenants)
CircuitGroup.objects.bulk_create((
CircuitGroup(name='Circuit Group 1', slug='circuit-group-1', description='foobar1', tenant=tenants[0]),
CircuitGroup(name='Circuit Group 2', slug='circuit-group-2', description='foobar2', tenant=tenants[1]),
CircuitGroup(name='Circuit Group 3', slug='circuit-group-3', tenant=tenants[1]),
))
def test_q(self):
params = {'q': 'foobar1'}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_name(self):
params = {'name': ['Circuit Group 1']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_slug(self):
params = {'slug': ['circuit-group-1']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_description(self):
params = {'description': ['foobar1', 'foobar2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_tenant(self):
tenants = Tenant.objects.all()[:2]
params = {'tenant_id': [tenants[0].pk, tenants[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
params = {'tenant': [tenants[0].slug, tenants[1].slug]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
def test_tenant_group(self):
tenant_groups = TenantGroup.objects.all()[:2]
params = {'tenant_group_id': [tenant_groups[0].pk, tenant_groups[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
params = {'tenant_group': [tenant_groups[0].slug, tenant_groups[1].slug]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
class CircuitGroupAssignmentTestCase(TestCase, ChangeLoggedFilterSetTests):
queryset = CircuitGroupAssignment.objects.all()
filterset = CircuitGroupAssignmentFilterSet
@classmethod
def setUpTestData(cls):
circuit_groups = (
CircuitGroup(name='Circuit Group 1', slug='circuit-group-1'),
CircuitGroup(name='Circuit Group 2', slug='circuit-group-2'),
CircuitGroup(name='Circuit Group 3', slug='circuit-group-3'),
CircuitGroup(name='Circuit Group 4', slug='circuit-group-4'),
)
CircuitGroup.objects.bulk_create(circuit_groups)
provider = Provider.objects.create(name='Provider 1', slug='provider-1')
circuittype = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1')
circuits = (
Circuit(cid='Circuit 1', provider=provider, type=circuittype),
Circuit(cid='Circuit 2', provider=provider, type=circuittype),
Circuit(cid='Circuit 3', provider=provider, type=circuittype),
Circuit(cid='Circuit 4', provider=provider, type=circuittype),
)
Circuit.objects.bulk_create(circuits)
assignments = (
CircuitGroupAssignment(
group=circuit_groups[0],
circuit=circuits[0],
priority=CircuitPriorityChoices.PRIORITY_PRIMARY
),
CircuitGroupAssignment(
group=circuit_groups[1],
circuit=circuits[1],
priority=CircuitPriorityChoices.PRIORITY_SECONDARY
),
CircuitGroupAssignment(
group=circuit_groups[2],
circuit=circuits[2],
priority=CircuitPriorityChoices.PRIORITY_TERTIARY
),
)
CircuitGroupAssignment.objects.bulk_create(assignments)
def test_group_id(self):
groups = CircuitGroup.objects.filter(name__in=['Circuit Group 1', 'Circuit Group 2'])
params = {'group_id': [groups[0].pk, groups[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'group': [groups[0].slug, groups[1].slug]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_circuit_id(self):
circuits = Circuit.objects.filter(cid__in=['Circuit 1', 'Circuit 2'])
params = {'circuit_id': [circuits[0].pk, circuits[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
class ProviderNetworkTestCase(TestCase, ChangeLoggedFilterSetTests):
queryset = ProviderNetwork.objects.all()
filterset = ProviderNetworkFilterSet

106
netbox/circuits/tests/test_views.py
View File

@ -404,3 +404,109 @@ class CircuitTerminationTestCase(ViewTestCases.PrimaryObjectViewTestCase):
response = self.client.get(reverse('circuits:circuittermination_trace', kwargs={'pk': circuittermination.pk}))
self.assertHttpStatus(response, 200)
class CircuitGroupTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
model = CircuitGroup
@classmethod
def setUpTestData(cls):
circuit_groups = (
CircuitGroup(name='Circuit Group 1', slug='circuit-group-1'),
CircuitGroup(name='Circuit Group 2', slug='circuit-group-2'),
CircuitGroup(name='Circuit Group 3', slug='circuit-group-3'),
)
CircuitGroup.objects.bulk_create(circuit_groups)
tags = create_tags('Alpha', 'Bravo', 'Charlie')
cls.form_data = {
'name': 'Circuit Group X',
'slug': 'circuit-group-x',
'description': 'A new Circuit Group',
'tags': [t.pk for t in tags],
}
cls.csv_data = (
"name,slug",
"Circuit Group 4,circuit-group-4",
"Circuit Group 5,circuit-group-5",
"Circuit Group 6,circuit-group-6",
)
cls.csv_update_data = (
"id,name,description",
f"{circuit_groups[0].pk},Circuit Group 7,New description7",
f"{circuit_groups[1].pk},Circuit Group 8,New description8",
f"{circuit_groups[2].pk},Circuit Group 9,New description9",
)
cls.bulk_edit_data = {
'description': 'Foo',
}
class CircuitGroupAssignmentTestCase(
ViewTestCases.CreateObjectViewTestCase,
ViewTestCases.EditObjectViewTestCase,
ViewTestCases.DeleteObjectViewTestCase,
ViewTestCases.ListObjectsViewTestCase,
ViewTestCases.BulkEditObjectsViewTestCase,
ViewTestCases.BulkDeleteObjectsViewTestCase
):
model = CircuitGroupAssignment
@classmethod
def setUpTestData(cls):
circuit_groups = (
CircuitGroup(name='Circuit Group 1', slug='circuit-group-1'),
CircuitGroup(name='Circuit Group 2', slug='circuit-group-2'),
CircuitGroup(name='Circuit Group 3', slug='circuit-group-3'),
CircuitGroup(name='Circuit Group 4', slug='circuit-group-4'),
)
CircuitGroup.objects.bulk_create(circuit_groups)
provider = Provider.objects.create(name='Provider 1', slug='provider-1')
circuittype = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1')
circuits = (
Circuit(cid='Circuit 1', provider=provider, type=circuittype),
Circuit(cid='Circuit 2', provider=provider, type=circuittype),
Circuit(cid='Circuit 3', provider=provider, type=circuittype),
Circuit(cid='Circuit 4', provider=provider, type=circuittype),
)
Circuit.objects.bulk_create(circuits)
assignments = (
CircuitGroupAssignment(
group=circuit_groups[0],
circuit=circuits[0],
priority=CircuitPriorityChoices.PRIORITY_PRIMARY
),
CircuitGroupAssignment(
group=circuit_groups[1],
circuit=circuits[1],
priority=CircuitPriorityChoices.PRIORITY_SECONDARY
),
CircuitGroupAssignment(
group=circuit_groups[2],
circuit=circuits[2],
priority=CircuitPriorityChoices.PRIORITY_TERTIARY
),
)
CircuitGroupAssignment.objects.bulk_create(assignments)
tags = create_tags('Alpha', 'Bravo', 'Charlie')
cls.form_data = {
'group': circuit_groups[3].pk,
'circuit': circuits[3].pk,
'priority': CircuitPriorityChoices.PRIORITY_INACTIVE,
'tags': [t.pk for t in tags],
}
cls.bulk_edit_data = {
'priority': CircuitPriorityChoices.PRIORITY_INACTIVE,
}

15
netbox/circuits/urls.py
View File

@ -55,4 +55,19 @@ urlpatterns = [
path('circuit-terminations/delete/', views.CircuitTerminationBulkDeleteView.as_view(), name='circuittermination_bulk_delete'),
path('circuit-terminations/<int:pk>/', include(get_model_urls('circuits', 'circuittermination'))),
# Circuit Groups
path('circuit-groups/', views.CircuitGroupListView.as_view(), name='circuitgroup_list'),
path('circuit-groups/add/', views.CircuitGroupEditView.as_view(), name='circuitgroup_add'),
path('circuit-groups/import/', views.CircuitGroupBulkImportView.as_view(), name='circuitgroup_import'),
path('circuit-groups/edit/', views.CircuitGroupBulkEditView.as_view(), name='circuitgroup_bulk_edit'),
path('circuit-groups/delete/', views.CircuitGroupBulkDeleteView.as_view(), name='circuitgroup_bulk_delete'),
path('circuit-groups/<int:pk>/', include(get_model_urls('circuits', 'circuitgroup'))),
# Circuit Group Assignments
path('circuit-group-assignments/', views.CircuitGroupAssignmentListView.as_view(), name='circuitgroupassignment_list'),
path('circuit-group-assignments/add/', views.CircuitGroupAssignmentEditView.as_view(), name='circuitgroupassignment_add'),
path('circuit-group-assignments/import/', views.CircuitGroupAssignmentBulkImportView.as_view(), name='circuitgroupassignment_import'),
path('circuit-group-assignments/edit/', views.CircuitGroupAssignmentBulkEditView.as_view(), name='circuitgroupassignment_bulk_edit'),
path('circuit-group-assignments/delete/', views.CircuitGroupAssignmentBulkDeleteView.as_view(), name='circuitgroupassignment_bulk_delete'),
path('circuit-group-assignments/<int:pk>/', include(get_model_urls('circuits', 'circuitgroupassignment'))),
]

97
netbox/circuits/views.py
View File

@ -440,3 +440,100 @@ class CircuitTerminationBulkDeleteView(generic.BulkDeleteView):
# Trace view
register_model_view(CircuitTermination, 'trace', kwargs={'model': CircuitTermination})(PathTraceView)
#
# Circuit Groups
#
class CircuitGroupListView(generic.ObjectListView):
queryset = CircuitGroup.objects.annotate(
circuit_group_assignment_count=count_related(CircuitGroupAssignment, 'group')
)
filterset = filtersets.CircuitGroupFilterSet
filterset_form = forms.CircuitGroupFilterForm
table = tables.CircuitGroupTable
@register_model_view(CircuitGroup)
class CircuitGroupView(GetRelatedModelsMixin, generic.ObjectView):
queryset = CircuitGroup.objects.all()
def get_extra_context(self, request, instance):
return {
'related_models': self.get_related_models(request, instance),
}
@register_model_view(CircuitGroup, 'edit')
class CircuitGroupEditView(generic.ObjectEditView):
queryset = CircuitGroup.objects.all()
form = forms.CircuitGroupForm
@register_model_view(CircuitGroup, 'delete')
class CircuitGroupDeleteView(generic.ObjectDeleteView):
queryset = CircuitGroup.objects.all()
class CircuitGroupBulkImportView(generic.BulkImportView):
queryset = CircuitGroup.objects.all()
model_form = forms.CircuitGroupImportForm
class CircuitGroupBulkEditView(generic.BulkEditView):
queryset = CircuitGroup.objects.all()
filterset = filtersets.CircuitGroupFilterSet
table = tables.CircuitGroupTable
form = forms.CircuitGroupBulkEditForm
class CircuitGroupBulkDeleteView(generic.BulkDeleteView):
queryset = CircuitGroup.objects.all()
filterset = filtersets.CircuitGroupFilterSet
table = tables.CircuitGroupTable
#
# Circuit Groups
#
class CircuitGroupAssignmentListView(generic.ObjectListView):
queryset = CircuitGroupAssignment.objects.all()
filterset = filtersets.CircuitGroupAssignmentFilterSet
filterset_form = forms.CircuitGroupAssignmentFilterForm
table = tables.CircuitGroupAssignmentTable
@register_model_view(CircuitGroupAssignment)
class CircuitGroupAssignmentView(generic.ObjectView):
queryset = CircuitGroupAssignment.objects.all()
@register_model_view(CircuitGroupAssignment, 'edit')
class CircuitGroupAssignmentEditView(generic.ObjectEditView):
queryset = CircuitGroupAssignment.objects.all()
form = forms.CircuitGroupAssignmentForm
@register_model_view(CircuitGroupAssignment, 'delete')
class CircuitGroupAssignmentDeleteView(generic.ObjectDeleteView):
queryset = CircuitGroupAssignment.objects.all()
class CircuitGroupAssignmentBulkImportView(generic.BulkImportView):
queryset = CircuitGroupAssignment.objects.all()
model_form = forms.CircuitGroupAssignmentImportForm
class CircuitGroupAssignmentBulkEditView(generic.BulkEditView):
queryset = CircuitGroupAssignment.objects.all()
filterset = filtersets.CircuitGroupAssignmentFilterSet
table = tables.CircuitGroupAssignmentTable
form = forms.CircuitGroupAssignmentBulkEditForm
class CircuitGroupAssignmentBulkDeleteView(generic.BulkDeleteView):
queryset = CircuitGroupAssignment.objects.all()
filterset = filtersets.CircuitGroupAssignmentFilterSet
table = tables.CircuitGroupAssignmentTable

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()

2
netbox/core/tables/change_logging.py
View File

@ -48,6 +48,6 @@ class ObjectChangeTable(NetBoxTable):
class Meta(NetBoxTable.Meta):
model = ObjectChange
fields = (
'pk', 'id', 'time', 'user_name', 'full_name', 'action', 'changed_object_type', 'object_repr', 'request_id',
'pk', 'time', 'user_name', 'full_name', 'action', 'changed_object_type', 'object_repr', 'request_id',
'actions',
)

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
@ -153,35 +154,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:

12
netbox/extras/signals.py
View File

@ -274,7 +274,11 @@ def process_job_start_event_rules(sender, **kwargs):
"""
Process event rules for jobs starting.
"""
event_rules = EventRule.objects.filter(type_job_start=True, enabled=True, object_types=sender.object_type)
event_rules = EventRule.objects.filter(
event_types__contains=[JOB_STARTED],
enabled=True,
object_types=sender.object_type
)
username = sender.user.username if sender.user else None
process_event_rules(
event_rules=event_rules,
@ -290,7 +294,11 @@ def process_job_end_event_rules(sender, **kwargs):
"""
Process event rules for jobs terminating.
"""
event_rules = EventRule.objects.filter(type_job_end=True, enabled=True, object_types=sender.object_type)
event_rules = EventRule.objects.filter(
event_types__contains=[JOB_COMPLETED],
enabled=True,
object_types=sender.object_type
)
username = sender.user.username if sender.user else None
process_event_rules(
event_rules=event_rules,

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)

57
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):
@ -1133,6 +1098,8 @@ class TagTestCase(TestCase, ChangeLoggedFilterSetTests):
'asnrange',
'cable',
'circuit',
'circuitgroup',
'circuitgroupassignment',
'circuittermination',
'circuittype',
'cluster',

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

@ -20,6 +20,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):

2
netbox/netbox/navigation/menu.py
View File

@ -259,6 +259,8 @@ CIRCUITS_MENU = Menu(
items=(
get_model_item('circuits', 'circuit', _('Circuits')),
get_model_item('circuits', 'circuittype', _('Circuit Types')),
get_model_item('circuits', 'circuitgroup', _('Circuit Groups')),
get_model_item('circuits', 'circuitgroupassignment', _('Group Assignments')),
get_model_item('circuits', 'circuittermination', _('Circuit Terminations')),
),
),

3
netbox/release.yaml
View File

@ -1,2 +1,3 @@
version: "4.1.0"
designation: "dev"
edition: Community
designation: dev

10
netbox/templates/base/layout.html
View File

@ -35,7 +35,8 @@ Blocks:
</h1>
{# User menu (mobile view) #}
<div class="navbar-nav flex-row d-lg-none">
<div class="navbar-nav flex-row align-items-center d-lg-none">
{% plugin_navbar %}
{% include 'inc/light_toggle.html' %}
{% include 'inc/user_menu.html' %}
</div>
@ -52,15 +53,10 @@ Blocks:
<header class="navbar navbar-expand-md sticky-top d-none d-lg-flex d-print-none">
<div class="container-fluid">
{# User menu (desktop view) #}
<div class="navbar-nav flex-row align-items-center order-md-last">
{# Plugin content #}
{% plugin_navbar %}
{# Dark/light mode toggle #}
{% include 'inc/light_toggle.html' %}
{# User menu #}
{% include 'inc/user_menu.html' %}
</div>

13
netbox/templates/circuits/circuit.html
View File

@ -61,6 +61,19 @@
</tr>
</table>
</div>
<div class="card">
<h5 class="card-header">
{% trans "Group Assignments" %}
{% if perms.circuits.add_circuitgroupassignment %}
<div class="card-actions">
<a href="{% url 'circuits:circuitgroupassignment_add' %}?circuit={{ object.pk }}&return_url={{ object.get_absolute_url }}" class="btn btn-ghost-primary btn-sm">
<span class="mdi mdi-plus-thick" aria-hidden="true"></span> {% trans "Assign Group" %}
</a>
</div>
{% endif %}
</h5>
{% htmx_table 'circuits:circuitgroupassignment_list' circuit_id=object.pk %}
</div>
{% include 'inc/panels/custom_fields.html' %}
{% include 'inc/panels/tags.html' %}
{% include 'inc/panels/comments.html' %}

56
netbox/templates/circuits/circuitgroup.html Normal file
View File

@ -0,0 +1,56 @@
{% extends 'generic/object.html' %}
{% load static %}
{% load helpers %}
{% load plugins %}
{% load render_table from django_tables2 %}
{% load i18n %}
{% block breadcrumbs %}
{{ block.super }}
<li class="breadcrumb-item"><a href="{% url 'circuits:circuitgroup_list' %}?circuitgroup_id={{ object.id }}">{{ object.name }}</a></li>
{% endblock %}
{% block extra_controls %}
{% if perms.circuit.add_circuitgroupassignment %}
<a href="{% url 'circuits:circuitgroupassignment_add' %}?group={{ object.pk }}" class="btn btn-primary">
<span class="mdi mdi-plus-thick" aria-hidden="true"></span> {% trans "Assign Circuit" %}
</a>
{% endif %}
{% endblock extra_controls %}
{% block content %}
<div class="row mb-3">
<div class="col col-md-6">
<div class="card">
<h5 class="card-header">{% trans "Circuit Group" %}</h5>
<table class="table table-hover attr-table">
<tr>
<th scope="row">{% trans "Name" %}</th>
<td>{{ object.name }}</td>
</tr>
<tr>
<th scope="row">{% trans "Description" %}</th>
<td>{{ object.description|placeholder }}</td>
</tr>
<tr>
<th scope="row">{% trans "Tenant" %}</th>
<td>
{% if object.tenant.group %}
{{ object.tenant.group|linkify }} /
{% endif %}
{{ object.tenant|linkify|placeholder }}
</td>
</tr>
</table>
</div>
{% include 'inc/panels/tags.html' %}
{% plugin_left_page object %}
</div>
<div class="col col-md-6">
{% include 'inc/panels/related_objects.html' %}
{% include 'inc/panels/comments.html' %}
{% include 'inc/panels/custom_fields.html' %}
{% plugin_right_page object %}
</div>
</div>
{% endblock %}

48
netbox/templates/circuits/circuitgroupassignment.html Normal file
View File

@ -0,0 +1,48 @@
{% extends 'generic/object.html' %}
{% load static %}
{% load helpers %}
{% load plugins %}
{% load render_table from django_tables2 %}
{% load i18n %}
{% block breadcrumbs %}
{{ block.super }}
<li class="breadcrumb-item">
<a href="{% url 'circuits:circuitgroupassignment_list' %}?group_id={{ object.group_id }}">{{ object.group }}</a>
</li>
{% endblock %}
{% block content %}
<div class="row mb-3">
<div class="col col-md-6">
<div class="card">
<h5 class="card-header">{% trans "Circuit Group Assignment" %}</h5>
<table class="table table-hover attr-table">
<tr>
<th scope="row">{% trans "Group" %}</th>
<td>{{ object.group }}</td>
</tr>
<tr>
<th scope="row">{% trans "Circuit" %}</th>
<td>{{ object.circuit }}</td>
</tr>
<tr>
<th scope="row">{% trans "Priority" %}</th>
<td>{{ object.priority }}</td>
</tr>
</table>
</div>
{% include 'inc/panels/tags.html' %}
{% include 'inc/panels/custom_fields.html' %}
{% plugin_left_page object %}
</div>
<div class="col col-md-6">
{% plugin_right_page object %}
</div>
</div>
<div class="row">
<div class="col col-md-12">
{% plugin_full_width_page object %}
</div>
</div>
{% endblock %}

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>

18
netbox/utilities/release.py
View File

@ -1,8 +1,8 @@
import datetime
import os
import yaml
from dataclasses import dataclass
from typing import Union
from dataclasses import dataclass, field
from typing import List, Union
from django.core.exceptions import ImproperlyConfigured
@ -15,9 +15,10 @@ LOCAL_RELEASE_PATH = 'local/release.yaml'
@dataclass
class ReleaseInfo:
version: str
edition: str = 'Community'
edition: str
published: Union[datetime.date, None] = None
designation: Union[str, None] = None
features: List = field(default_factory=list)
@property
def full_version(self):
@ -46,11 +47,12 @@ def load_release_data():
local_data = yaml.safe_load(release_file)
except FileNotFoundError:
local_data = {}
if type(local_data) is not dict:
raise ImproperlyConfigured(
f"{LOCAL_RELEASE_PATH}: Local release data must be defined as a dictionary."
)
data.update(local_data)
if local_data is not None:
if type(local_data) is not dict:
raise ImproperlyConfigured(
f"{LOCAL_RELEASE_PATH}: Local release data must be defined as a dictionary."
)
data.update(local_data)
# Convert the published date to a date object
if 'published' in data:

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