diff --git a/.github/ISSUE_TEMPLATE/bug_report.yaml b/.github/ISSUE_TEMPLATE/bug_report.yaml index 8664768ee..d0ded0e4c 100644 --- a/.github/ISSUE_TEMPLATE/bug_report.yaml +++ b/.github/ISSUE_TEMPLATE/bug_report.yaml @@ -10,16 +10,25 @@ body: installation. If you're having trouble with installation or just looking for assistance with using NetBox, please visit our [discussion forum](https://github.com/netbox-community/netbox/discussions) instead. + - type: dropdown + attributes: + label: Deployment Type + description: How are you running NetBox? + options: + - Self-hosted + - NetBox Cloud + validations: + required: true - type: input attributes: - label: NetBox version + label: NetBox Version description: What version of NetBox are you currently running? - placeholder: v3.6.2 + placeholder: v3.7.1 validations: required: true - type: dropdown attributes: - label: Python version + label: Python Version description: What version of Python are you currently running? options: - "3.8" diff --git a/.github/ISSUE_TEMPLATE/config.yml b/.github/ISSUE_TEMPLATE/config.yml index e6a5e76c2..2ad52023e 100644 --- a/.github/ISSUE_TEMPLATE/config.yml +++ b/.github/ISSUE_TEMPLATE/config.yml @@ -7,6 +7,9 @@ contact_links: - name: ❓ Discussion url: https://github.com/netbox-community/netbox/discussions about: "If you're just looking for help, try starting a discussion instead." + - name: 🌎 Correct a Translation + url: https://explore.transifex.com/netbox-community/netbox/ + about: "Spot an incorrect translation? You can propose a fix on Transifex." - name: 💡 Plugin Idea url: https://plugin-ideas.netbox.dev about: "Have an idea for a plugin? Head over to the ideas board!" diff --git a/.github/ISSUE_TEMPLATE/feature_request.yaml b/.github/ISSUE_TEMPLATE/feature_request.yaml index 8e3af527a..5c4fc375e 100644 --- a/.github/ISSUE_TEMPLATE/feature_request.yaml +++ b/.github/ISSUE_TEMPLATE/feature_request.yaml @@ -14,7 +14,7 @@ body: attributes: label: NetBox version description: What version of NetBox are you currently running? - placeholder: v3.6.2 + placeholder: v3.7.1 validations: required: true - type: dropdown diff --git a/.github/ISSUE_TEMPLATE/translation.yaml b/.github/ISSUE_TEMPLATE/translation.yaml new file mode 100644 index 000000000..d07bc399d --- /dev/null +++ b/.github/ISSUE_TEMPLATE/translation.yaml @@ -0,0 +1,37 @@ +--- +name: 🌍 Translation +description: Request support for a new language in the user interface +labels: ["type: translation"] +body: + - type: markdown + attributes: + value: > + **NOTE:** This template is used only for proposing the addition of *new* languages. Please do + not use it to request changes to existing translations. + - type: input + attributes: + label: Language + description: What is the name of the language in English? + validations: + required: true + - type: input + attributes: + label: ISO 639-1 code + description: > + What is the two-letter [ISO 639-1 code](https://en.wikipedia.org/wiki/List_of_ISO_639-1_codes) + assigned to the language? + validations: + required: true + - type: dropdown + attributes: + label: Volunteer + description: Are you a fluent speaker of this language **and** willing to contribute a translation map? + options: + - "Yes" + - "No" + validations: + required: true + - type: textarea + attributes: + label: Comments + description: Any other notes you would like to share diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1d9692194..ed8c65b7d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,15 +31,15 @@ jobs: steps: - name: Check out repo - uses: actions/checkout@v2 + uses: actions/checkout@v4 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v2 + uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - name: Use Node.js ${{ matrix.node-version }} - uses: actions/setup-node@v2 + uses: actions/setup-node@v3 with: node-version: ${{ matrix.node-version }} @@ -47,7 +47,7 @@ jobs: run: npm install -g yarn - name: Setup Node.js with Yarn Caching - uses: actions/setup-node@v2 + uses: actions/setup-node@v3 with: node-version: ${{ matrix.node-version }} cache: yarn @@ -68,6 +68,9 @@ jobs: - name: Collect static files run: python netbox/manage.py collectstatic --no-input + - name: Check for missing migrations + run: python netbox/manage.py makemigrations --check + - name: Check PEP8 compliance run: pycodestyle --ignore=W504,E501 --exclude=node_modules netbox/ diff --git a/.github/workflows/lock.yml b/.github/workflows/lock.yml index 6019cef5d..ad3bf5d75 100644 --- a/.github/workflows/lock.yml +++ b/.github/workflows/lock.yml @@ -9,13 +9,15 @@ on: permissions: issues: write pull-requests: write + discussions: write jobs: lock: runs-on: ubuntu-latest steps: - - uses: dessant/lock-threads@v3 + - uses: dessant/lock-threads@v5 with: issue-inactive-days: 90 pr-inactive-days: 30 + discussion-inactive-days: 180 issue-lock-reason: 'resolved' diff --git a/.github/workflows/stale.yml b/.github/workflows/stale.yml index 3b37aae56..22de146a2 100644 --- a/.github/workflows/stale.yml +++ b/.github/workflows/stale.yml @@ -15,7 +15,7 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/stale@v6 + - uses: actions/stale@v8 with: close-issue-message: > This issue has been automatically closed due to lack of activity. In an diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 301fac079..471846427 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -36,6 +36,8 @@ NetBox users are welcome to participate in either role, on stage or in the crowd ## :bug: Reporting Bugs +:warning: Bug reports are used to call attention to some unintended or unexpected behavior in NetBox, such as when an error occurs or when the result of taking some action is inconsistent with the documentation. **Bug reports may not be used to suggest new functionality**; please see "feature requests" below if that is your goal. + * First, ensure that you're running the [latest stable version](https://github.com/netbox-community/netbox/releases) of NetBox. If you're running an older version, it's likely that the bug has already been fixed. * Next, search our [issues list](https://github.com/netbox-community/netbox/issues?q=is%3Aissue) to see if the bug you've found has already been reported. If you come across a bug report that seems to match, please click "add a reaction" in the top right corner of the issue and add a thumbs up (:thumbsup:). This will help draw more attention to it. Any comments you can add to provide additional information or context would also be much appreciated. diff --git a/README.md b/README.md index 6e50e5687..f166919c4 100644 --- a/README.md +++ b/README.md @@ -1,86 +1,129 @@
The premier source of truth powering network automation
- +The cornerstone of every automated network
+ + + + + ++ NetBox's Role | + Why NetBox? | + Getting Started | + Get Involved | + Project Stats | + Screenshots +
-![Screenshot of NetBox UI](docs/media/screenshots/netbox-ui.png "NetBox UI") ++ +
+ +## NetBox's Role + +NetBox functions as the **source of truth** for your network infrastructure. Its job is to define and validate the _intended state_ of all network components and resources. NetBox does not interact with network nodes directly; rather, it makes this data available programmatically to purpose-built automation, monitoring, and assurance tools. This separation of duties enables the construction of a robust yet flexible automation system. + ++ +
+ +The diagram above illustrates the recommended deployment architecture for an automated network, leveraging NetBox as the central authority for network state. This approach allows your team to swap out individual tools to meet changing needs while retaining a predictable, modular workflow. + +## Why NetBox? + +### Comprehensive Data Model + +Racks, devices, cables, IP addresses, VLANs, circuits, power, VPNs, and lots more: NetBox is built for networks. Its comprehensive and thoroughly inter-linked data model provides for natural and highly structured modeling of myriad network primitives that just isn't possible using general-purpose tools. And there's no need to waste time contemplating how to build out a database: Everything is ready to go upon installation. + +### Focused Development + +NetBox strives to meet a singular goal: Provide the best available solution for making network infrastructure programmatically accessible. Unlike "all-in-one" tools which awkwardly bolt on half-baked features in an attempt to check every box, NetBox is committed to its core function. NetBox provides the best possible solution for modeling network infrastructure, and provides rich APIs for integrating with tools that excel in other areas of network automation. + +### Extensible and Customizable + +No two networks are exactly the same. Users are empowered to extend NetBox's native data model with custom fields and tags to best suit their unique needs. You can even write your own plugins to introduce entirely new objects and functionality! + +### Flexible Permissions + +NetBox includes a fully customizable permission system, which affords administrators incredible granularity when assigning roles to users and groups. Want to restrict certain users to working only with cabling and not be able to change IP addresses? Or maybe each team should have access only to a particular tenant? NetBox enables you to craft roles as you see fit. + +### Custom Validation & Protection Rules + +The data you put into NetBox is crucial to network operations. In addition to its robust native validation rules, NetBox provides mechanisms for administrators to define their own custom validation rules for objects. Custom validation can be used both to ensure new or modified objects adhere to a set of rules, and to prevent the deletion of objects which don't meet certain criteria. (For example, you might want to prevent the deletion of a device with an "active" status.) + +### Device Configuration Rendering + +NetBox can render user-created Jinja2 templates to generate device configurations from its own data. Configuration templates can be uploaded individually or pulled automatically from an external source, such as a git repository. Rendered configurations can be retrieved via the REST API for application directly to network devices via a provisioning tool such as Ansible or Salt. + +### Custom Scripts + +Complex workflows, such as provisioning a new branch office, can be tedious to carry out via the user interface. NetBox allows you to write and upload custom scripts that can be run directly from the UI. Scripts prompt users for input and then automate the necessary tasks to greatly simplify otherwise burdensome processes. + +### Automated Events + +Users can define event rules to automatically trigger a custom script or outbound webhook in response to a NetBox event. For example, you might want to automatically update a network monitoring service whenever a new device is added to NetBox, or update a DHCP server when an IP range is allocated. + +### Comprehensive Change Logging + +NetBox automatically logs the creation, modification, and deletion of all managed objects, providing a thorough change history. Changes can be attributed to the executing user, and related changes are grouped automatically by request ID. + +> [!NOTE] +> A complete list of NetBox's myriad features can be found in [the introductory documentation](https://docs.netbox.dev/en/stable/introduction/). ## Getting Started -
+
+ Looking for an enterprise solution? Check out NetBox Cloud!
+
Stats via Repography
-
+ NetBox Dashboard (Light Mode)
+
+
+ NetBox Dashboard (Dark Mode)
+
+
+ Prefixes List
+
+
+ Rack View
+
+
+ Cable Trace
+
+
00ff00
'),
+ }
class CircuitImportForm(NetBoxModelImportForm):
diff --git a/netbox/circuits/forms/filtersets.py b/netbox/circuits/forms/filtersets.py
index 1fb239023..1e1abd068 100644
--- a/netbox/circuits/forms/filtersets.py
+++ b/netbox/circuits/forms/filtersets.py
@@ -7,7 +7,7 @@ from dcim.models import Region, Site, SiteGroup
from ipam.models import ASN
from netbox.forms import NetBoxModelFilterSetForm
from tenancy.forms import TenancyFilterForm, ContactModelFilterForm
-from utilities.forms.fields import DynamicModelMultipleChoiceField, TagFilterField
+from utilities.forms.fields import ColorField, DynamicModelMultipleChoiceField, TagFilterField
from utilities.forms.widgets import DatePicker, NumberWithOptions
__all__ = (
@@ -88,7 +88,7 @@ class ProviderNetworkFilterForm(NetBoxModelFilterSetForm):
label=_('Provider')
)
service_id = forms.CharField(
- label=_('Service id'),
+ label=_('Service ID'),
max_length=100,
required=False
)
@@ -97,8 +97,17 @@ class ProviderNetworkFilterForm(NetBoxModelFilterSetForm):
class CircuitTypeFilterForm(NetBoxModelFilterSetForm):
model = CircuitType
+ fieldsets = (
+ (None, ('q', 'filter_id', 'tag')),
+ (_('Attributes'), ('color',)),
+ )
tag = TagFilterField(model)
+ color = ColorField(
+ label=_('Color'),
+ required=False
+ )
+
class CircuitFilterForm(TenancyFilterForm, ContactModelFilterForm, NetBoxModelFilterSetForm):
model = Circuit
@@ -110,6 +119,7 @@ class CircuitFilterForm(TenancyFilterForm, ContactModelFilterForm, NetBoxModelFi
(_('Tenant'), ('tenant_group_id', 'tenant_id')),
(_('Contacts'), ('contact', 'contact_role', 'contact_group')),
)
+ selector_fields = ('filter_id', 'q', 'region_id', 'site_group_id', 'site_id', 'provider_id', 'provider_network_id')
type_id = DynamicModelMultipleChoiceField(
queryset=CircuitType.objects.all(),
required=False,
diff --git a/netbox/circuits/forms/model_forms.py b/netbox/circuits/forms/model_forms.py
index 8a540032e..0809cb2f4 100644
--- a/netbox/circuits/forms/model_forms.py
+++ b/netbox/circuits/forms/model_forms.py
@@ -76,14 +76,14 @@ class CircuitTypeForm(NetBoxModelForm):
fieldsets = (
(_('Circuit Type'), (
- 'name', 'slug', 'description', 'tags',
+ 'name', 'slug', 'color', 'description', 'tags',
)),
)
class Meta:
model = CircuitType
fields = [
- 'name', 'slug', 'description', 'tags',
+ 'name', 'slug', 'color', 'description', 'tags',
]
diff --git a/netbox/circuits/migrations/0043_circuittype_color.py b/netbox/circuits/migrations/0043_circuittype_color.py
new file mode 100644
index 000000000..6c4dffeb6
--- /dev/null
+++ b/netbox/circuits/migrations/0043_circuittype_color.py
@@ -0,0 +1,18 @@
+# Generated by Django 4.2.5 on 2023-10-20 21:25
+
+from django.db import migrations
+import utilities.fields
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ('circuits', '0042_provideraccount'),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name='circuittype',
+ name='color',
+ field=utilities.fields.ColorField(blank=True, max_length=6),
+ ),
+ ]
diff --git a/netbox/circuits/models/circuits.py b/netbox/circuits/models/circuits.py
index 0322b67c6..4dc775364 100644
--- a/netbox/circuits/models/circuits.py
+++ b/netbox/circuits/models/circuits.py
@@ -7,6 +7,7 @@ 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 utilities.fields import ColorField
__all__ = (
'Circuit',
@@ -20,6 +21,11 @@ class CircuitType(OrganizationalModel):
Circuits can be organized by their functional role. For example, a user might wish to define CircuitTypes named
"Long Haul," "Metro," or "Out-of-Band".
"""
+ color = ColorField(
+ verbose_name=_('color'),
+ blank=True
+ )
+
def get_absolute_url(self):
return reverse('circuits:circuittype', args=[self.pk])
diff --git a/netbox/circuits/search.py b/netbox/circuits/search.py
index b80f92d4d..c22b400eb 100644
--- a/netbox/circuits/search.py
+++ b/netbox/circuits/search.py
@@ -10,6 +10,7 @@ class CircuitIndex(SearchIndex):
('description', 500),
('comments', 5000),
)
+ display_attrs = ('provider', 'provider_account', 'type', 'status', 'tenant', 'description')
@register_search
@@ -22,6 +23,7 @@ class CircuitTerminationIndex(SearchIndex):
('port_speed', 2000),
('upstream_speed', 2000),
)
+ display_attrs = ('circuit', 'site', 'provider_network', 'description')
@register_search
@@ -32,6 +34,7 @@ class CircuitTypeIndex(SearchIndex):
('slug', 110),
('description', 500),
)
+ display_attrs = ('description',)
@register_search
@@ -42,6 +45,7 @@ class ProviderIndex(SearchIndex):
('description', 500),
('comments', 5000),
)
+ display_attrs = ('description',)
class ProviderAccountIndex(SearchIndex):
@@ -51,6 +55,7 @@ class ProviderAccountIndex(SearchIndex):
('account', 200),
('comments', 5000),
)
+ display_attrs = ('provider', 'account', 'description')
@register_search
@@ -62,3 +67,4 @@ class ProviderNetworkIndex(SearchIndex):
('description', 500),
('comments', 5000),
)
+ display_attrs = ('provider', 'service_id', 'description')
diff --git a/netbox/circuits/tables/circuits.py b/netbox/circuits/tables/circuits.py
index 6a05983e6..6ae727eca 100644
--- a/netbox/circuits/tables/circuits.py
+++ b/netbox/circuits/tables/circuits.py
@@ -28,6 +28,7 @@ class CircuitTypeTable(NetBoxTable):
linkify=True,
verbose_name=_('Name'),
)
+ color = columns.ColorColumn()
tags = columns.TagColumn(
url_name='circuits:circuittype_list'
)
@@ -40,7 +41,7 @@ class CircuitTypeTable(NetBoxTable):
class Meta(NetBoxTable.Meta):
model = CircuitType
fields = (
- 'pk', 'id', 'name', 'circuit_count', 'description', 'slug', 'tags', 'created', 'last_updated', 'actions',
+ 'pk', 'id', 'name', 'circuit_count', 'color', 'description', 'slug', 'tags', 'created', 'last_updated', 'actions',
)
default_columns = ('pk', 'name', 'circuit_count', 'description', 'slug')
diff --git a/netbox/circuits/tests/test_filtersets.py b/netbox/circuits/tests/test_filtersets.py
index e3380a1e5..6553179ec 100644
--- a/netbox/circuits/tests/test_filtersets.py
+++ b/netbox/circuits/tests/test_filtersets.py
@@ -25,8 +25,8 @@ class ProviderTestCase(TestCase, ChangeLoggedFilterSetTests):
ASN.objects.bulk_create(asns)
providers = (
- Provider(name='Provider 1', slug='provider-1'),
- Provider(name='Provider 2', slug='provider-2'),
+ Provider(name='Provider 1', slug='provider-1', description='foobar1'),
+ Provider(name='Provider 2', slug='provider-2', description='foobar2'),
Provider(name='Provider 3', slug='provider-3'),
Provider(name='Provider 4', slug='provider-4'),
Provider(name='Provider 5', slug='provider-5'),
@@ -74,6 +74,10 @@ class ProviderTestCase(TestCase, ChangeLoggedFilterSetTests):
CircuitTermination(circuit=circuits[1], site=sites[0], term_side='A'),
))
+ def test_q(self):
+ params = {'q': 'foobar1'}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
def test_name(self):
params = {'name': ['Provider 1', 'Provider 2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
@@ -82,6 +86,10 @@ class ProviderTestCase(TestCase, ChangeLoggedFilterSetTests):
params = {'slug': ['provider-1', 'provider-2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+ def test_description(self):
+ params = {'description': ['foobar1', 'foobar2']}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
def test_asn_id(self): # ASN object assignment
asns = ASN.objects.all()[:2]
params = {'asn_id': [asns[0].pk, asns[1].pk]}
@@ -122,6 +130,10 @@ class CircuitTypeTestCase(TestCase, ChangeLoggedFilterSetTests):
CircuitType(name='Circuit Type 3', slug='circuit-type-3'),
))
+ def test_q(self):
+ params = {'q': 'foobar1'}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
def test_name(self):
params = {'name': ['Circuit Type 1']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
@@ -227,6 +239,10 @@ class CircuitTestCase(TestCase, ChangeLoggedFilterSetTests):
))
CircuitTermination.objects.bulk_create(circuit_terminations)
+ def test_q(self):
+ params = {'q': 'foobar1'}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
def test_cid(self):
params = {'cid': ['Test Circuit 1', 'Test Circuit 2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
@@ -369,6 +385,10 @@ class CircuitTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
Cable(a_terminations=[circuit_terminations[0]], b_terminations=[circuit_terminations[1]]).save()
+ def test_q(self):
+ params = {'q': 'foobar1'}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
def test_term_side(self):
params = {'term_side': 'A'}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 7)
@@ -440,6 +460,10 @@ class ProviderNetworkTestCase(TestCase, ChangeLoggedFilterSetTests):
)
ProviderNetwork.objects.bulk_create(provider_networks)
+ def test_q(self):
+ params = {'q': 'foobar1'}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
def test_name(self):
params = {'name': ['Provider Network 1', 'Provider Network 2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
@@ -477,6 +501,10 @@ class ProviderAccountTestCase(TestCase, ChangeLoggedFilterSetTests):
)
ProviderAccount.objects.bulk_create(provider_accounts)
+ def test_q(self):
+ params = {'q': 'foobar1'}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
def test_name(self):
params = {'name': ['Provider Account 1', 'Provider Account 2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
diff --git a/netbox/core/api/serializers.py b/netbox/core/api/serializers.py
index 4117a609c..a16a06d62 100644
--- a/netbox/core/api/serializers.py
+++ b/netbox/core/api/serializers.py
@@ -4,6 +4,7 @@ from core.choices import *
from core.models import *
from netbox.api.fields import ChoiceField, ContentTypeField
from netbox.api.serializers import BaseModelSerializer, NetBoxModelSerializer
+from netbox.utils import get_data_backend_choices
from users.api.nested_serializers import NestedUserSerializer
from .nested_serializers import *
@@ -19,7 +20,7 @@ class DataSourceSerializer(NetBoxModelSerializer):
view_name='core-api:datasource-detail'
)
type = ChoiceField(
- choices=DataSourceTypeChoices
+ choices=get_data_backend_choices()
)
status = ChoiceField(
choices=DataSourceStatusChoices,
@@ -35,7 +36,7 @@ class DataSourceSerializer(NetBoxModelSerializer):
model = DataSource
fields = [
'id', 'url', 'display', 'name', 'type', 'source_url', 'enabled', 'status', 'description', 'comments',
- 'parameters', 'ignore_rules', 'created', 'last_updated', 'file_count',
+ 'parameters', 'ignore_rules', 'custom_fields', 'created', 'last_updated', 'file_count',
]
@@ -68,5 +69,5 @@ class JobSerializer(BaseModelSerializer):
model = Job
fields = [
'id', 'url', 'display', 'object_type', 'object_id', 'name', 'status', 'created', 'scheduled', 'interval',
- 'started', 'completed', 'user', 'data', 'job_id',
+ 'started', 'completed', 'user', 'data', 'error', 'job_id',
]
diff --git a/netbox/core/choices.py b/netbox/core/choices.py
index b5d9d0d90..8d7050414 100644
--- a/netbox/core/choices.py
+++ b/netbox/core/choices.py
@@ -7,18 +7,6 @@ from utilities.choices import ChoiceSet
# Data sources
#
-class DataSourceTypeChoices(ChoiceSet):
- LOCAL = 'local'
- GIT = 'git'
- AMAZON_S3 = 'amazon-s3'
-
- CHOICES = (
- (LOCAL, _('Local'), 'gray'),
- (GIT, 'Git', 'blue'),
- (AMAZON_S3, 'Amazon S3', 'blue'),
- )
-
-
class DataSourceStatusChoices(ChoiceSet):
NEW = 'new'
QUEUED = 'queued'
diff --git a/netbox/core/data_backends.py b/netbox/core/data_backends.py
index 82b3962dd..9ff0b4d63 100644
--- a/netbox/core/data_backends.py
+++ b/netbox/core/data_backends.py
@@ -10,61 +10,24 @@ from django import forms
from django.conf import settings
from django.utils.translation import gettext as _
-from netbox.registry import registry
-from .choices import DataSourceTypeChoices
+from netbox.data_backends import DataBackend
+from netbox.utils import register_data_backend
from .exceptions import SyncError
__all__ = (
- 'LocalBackend',
'GitBackend',
+ 'LocalBackend',
'S3Backend',
)
logger = logging.getLogger('netbox.data_backends')
-def register_backend(name):
- """
- Decorator for registering a DataBackend class.
- """
-
- def _wrapper(cls):
- registry['data_backends'][name] = cls
- return cls
-
- return _wrapper
-
-
-class DataBackend:
- parameters = {}
- sensitive_parameters = []
-
- # Prevent Django's template engine from calling the backend
- # class when referenced via DataSource.backend_class
- do_not_call_in_templates = True
-
- def __init__(self, url, **kwargs):
- self.url = url
- self.params = kwargs
- self.config = self.init_config()
-
- def init_config(self):
- """
- Hook to initialize the instance's configuration.
- """
- return
-
- @property
- def url_scheme(self):
- return urlparse(self.url).scheme.lower()
-
- @contextmanager
- def fetch(self):
- raise NotImplemented()
-
-
-@register_backend(DataSourceTypeChoices.LOCAL)
+@register_data_backend()
class LocalBackend(DataBackend):
+ name = 'local'
+ label = _('Local')
+ is_local = True
@contextmanager
def fetch(self):
@@ -74,8 +37,10 @@ class LocalBackend(DataBackend):
yield local_path
-@register_backend(DataSourceTypeChoices.GIT)
+@register_data_backend()
class GitBackend(DataBackend):
+ name = 'git'
+ label = 'Git'
parameters = {
'username': forms.CharField(
required=False,
@@ -144,8 +109,10 @@ class GitBackend(DataBackend):
local_path.cleanup()
-@register_backend(DataSourceTypeChoices.AMAZON_S3)
+@register_data_backend()
class S3Backend(DataBackend):
+ name = 'amazon-s3'
+ label = 'Amazon S3'
parameters = {
'aws_access_key_id': forms.CharField(
label=_('AWS access key ID'),
diff --git a/netbox/core/filtersets.py b/netbox/core/filtersets.py
index 62a58086a..902e240ee 100644
--- a/netbox/core/filtersets.py
+++ b/netbox/core/filtersets.py
@@ -4,10 +4,12 @@ from django.utils.translation import gettext as _
import django_filters
from netbox.filtersets import BaseFilterSet, ChangeLoggedModelFilterSet, NetBoxModelFilterSet
+from netbox.utils import get_data_backend_choices
from .choices import *
from .models import *
__all__ = (
+ 'ConfigRevisionFilterSet',
'DataFileFilterSet',
'DataSourceFilterSet',
'JobFilterSet',
@@ -16,7 +18,7 @@ __all__ = (
class DataSourceFilterSet(NetBoxModelFilterSet):
type = django_filters.MultipleChoiceFilter(
- choices=DataSourceTypeChoices,
+ choices=get_data_backend_choices,
null_value=None
)
status = django_filters.MultipleChoiceFilter(
@@ -26,7 +28,7 @@ class DataSourceFilterSet(NetBoxModelFilterSet):
class Meta:
model = DataSource
- fields = ('id', 'name', 'enabled')
+ fields = ('id', 'name', 'enabled', 'description')
def search(self, queryset, name, value):
if not value.strip():
@@ -122,3 +124,23 @@ class JobFilterSet(BaseFilterSet):
Q(user__username__icontains=value) |
Q(name__icontains=value)
)
+
+
+class ConfigRevisionFilterSet(BaseFilterSet):
+ q = django_filters.CharFilter(
+ method='search',
+ label=_('Search'),
+ )
+
+ class Meta:
+ model = ConfigRevision
+ fields = [
+ 'id',
+ ]
+
+ def search(self, queryset, name, value):
+ if not value.strip():
+ return queryset
+ return queryset.filter(
+ Q(comment__icontains=value)
+ )
diff --git a/netbox/core/forms/bulk_edit.py b/netbox/core/forms/bulk_edit.py
index a4ecd646f..dcc92c6f0 100644
--- a/netbox/core/forms/bulk_edit.py
+++ b/netbox/core/forms/bulk_edit.py
@@ -1,10 +1,9 @@
from django import forms
from django.utils.translation import gettext_lazy as _
-from core.choices import DataSourceTypeChoices
from core.models import *
from netbox.forms import NetBoxModelBulkEditForm
-from utilities.forms import add_blank_choice
+from netbox.utils import get_data_backend_choices
from utilities.forms.fields import CommentField
from utilities.forms.widgets import BulkEditNullBooleanSelect
@@ -16,9 +15,8 @@ __all__ = (
class DataSourceBulkEditForm(NetBoxModelBulkEditForm):
type = forms.ChoiceField(
label=_('Type'),
- choices=add_blank_choice(DataSourceTypeChoices),
- required=False,
- initial=''
+ choices=get_data_backend_choices,
+ required=False
)
enabled = forms.NullBooleanField(
required=False,
diff --git a/netbox/core/forms/filtersets.py b/netbox/core/forms/filtersets.py
index f7a6f3595..f21bd3f87 100644
--- a/netbox/core/forms/filtersets.py
+++ b/netbox/core/forms/filtersets.py
@@ -1,18 +1,18 @@
from django import forms
from django.contrib.auth import get_user_model
-from django.contrib.contenttypes.models import ContentType
from django.utils.translation import gettext_lazy as _
from core.choices import *
from core.models import *
-from extras.forms.mixins import SavedFiltersMixin
-from extras.utils import FeatureQuery
from netbox.forms import NetBoxModelFilterSetForm
+from netbox.forms.mixins import SavedFiltersMixin
+from netbox.utils import get_data_backend_choices
from utilities.forms import BOOLEAN_WITH_BLANK_CHOICES, FilterForm
from utilities.forms.fields import ContentTypeChoiceField, DynamicModelMultipleChoiceField
from utilities.forms.widgets import APISelectMultiple, DateTimePicker
__all__ = (
+ 'ConfigRevisionFilterForm',
'DataFileFilterForm',
'DataSourceFilterForm',
'JobFilterForm',
@@ -27,7 +27,7 @@ class DataSourceFilterForm(NetBoxModelFilterSetForm):
)
type = forms.MultipleChoiceField(
label=_('Type'),
- choices=DataSourceTypeChoices,
+ choices=get_data_backend_choices,
required=False
)
status = forms.MultipleChoiceField(
@@ -68,7 +68,7 @@ class JobFilterForm(SavedFiltersMixin, FilterForm):
)
object_type = ContentTypeChoiceField(
label=_('Object Type'),
- queryset=ContentType.objects.filter(FeatureQuery('jobs').get_query()),
+ queryset=ContentType.objects.with_feature('jobs'),
required=False,
)
status = forms.MultipleChoiceField(
@@ -124,3 +124,9 @@ class JobFilterForm(SavedFiltersMixin, FilterForm):
api_url='/api/users/users/',
)
)
+
+
+class ConfigRevisionFilterForm(SavedFiltersMixin, FilterForm):
+ fieldsets = (
+ (None, ('q', 'filter_id')),
+ )
diff --git a/netbox/core/forms/model_forms.py b/netbox/core/forms/model_forms.py
index 01d5474c6..652728734 100644
--- a/netbox/core/forms/model_forms.py
+++ b/netbox/core/forms/model_forms.py
@@ -1,23 +1,34 @@
import copy
+import json
from django import forms
+from django.conf import settings
from django.utils.translation import gettext_lazy as _
from core.forms.mixins import SyncedDataMixin
from core.models import *
+from netbox.config import get_config, PARAMS
from netbox.forms import NetBoxModelForm
from netbox.registry import registry
-from utilities.forms import get_field_value
+from netbox.utils import get_data_backend_choices
+from utilities.forms import BootstrapMixin, get_field_value
from utilities.forms.fields import CommentField
from utilities.forms.widgets import HTMXSelect
__all__ = (
+ 'ConfigRevisionForm',
'DataSourceForm',
'ManagedFileForm',
)
+EMPTY_VALUES = ('', None, [], ())
+
class DataSourceForm(NetBoxModelForm):
+ type = forms.ChoiceField(
+ choices=get_data_backend_choices,
+ widget=HTMXSelect()
+ )
comments = CommentField()
class Meta:
@@ -26,7 +37,6 @@ class DataSourceForm(NetBoxModelForm):
'name', 'type', 'source_url', 'enabled', 'description', 'comments', 'ignore_rules', 'tags',
]
widgets = {
- 'type': HTMXSelect(),
'ignore_rules': forms.Textarea(
attrs={
'rows': 5,
@@ -56,12 +66,13 @@ class DataSourceForm(NetBoxModelForm):
# Add backend-specific form fields
self.backend_fields = []
- for name, form_field in backend.parameters.items():
- field_name = f'backend_{name}'
- self.backend_fields.append(field_name)
- self.fields[field_name] = copy.copy(form_field)
- if self.instance and self.instance.parameters:
- self.fields[field_name].initial = self.instance.parameters.get(name)
+ if backend:
+ for name, form_field in backend.parameters.items():
+ field_name = f'backend_{name}'
+ self.backend_fields.append(field_name)
+ self.fields[field_name] = copy.copy(form_field)
+ if self.instance and self.instance.parameters:
+ self.fields[field_name].initial = self.instance.parameters.get(name)
def save(self, *args, **kwargs):
@@ -106,3 +117,113 @@ class ManagedFileForm(SyncedDataMixin, NetBoxModelForm):
new_file.write(self.cleaned_data['upload_file'].read())
return super().save(*args, **kwargs)
+
+
+class ConfigFormMetaclass(forms.models.ModelFormMetaclass):
+
+ def __new__(mcs, name, bases, attrs):
+
+ # Emulate a declared field for each supported configuration parameter
+ param_fields = {}
+ for param in PARAMS:
+ field_kwargs = {
+ 'required': False,
+ 'label': param.label,
+ 'help_text': param.description,
+ }
+ field_kwargs.update(**param.field_kwargs)
+ param_fields[param.name] = param.field(**field_kwargs)
+ attrs.update(param_fields)
+
+ return super().__new__(mcs, name, bases, attrs)
+
+
+class ConfigRevisionForm(BootstrapMixin, forms.ModelForm, metaclass=ConfigFormMetaclass):
+ """
+ Form for creating a new ConfigRevision.
+ """
+
+ fieldsets = (
+ (_('Rack Elevations'), ('RACK_ELEVATION_DEFAULT_UNIT_HEIGHT', 'RACK_ELEVATION_DEFAULT_UNIT_WIDTH')),
+ (_('Power'), ('POWERFEED_DEFAULT_VOLTAGE', 'POWERFEED_DEFAULT_AMPERAGE', 'POWERFEED_DEFAULT_MAX_UTILIZATION')),
+ (_('IPAM'), ('ENFORCE_GLOBAL_UNIQUE', 'PREFER_IPV4')),
+ (_('Security'), ('ALLOWED_URL_SCHEMES',)),
+ (_('Banners'), ('BANNER_LOGIN', 'BANNER_MAINTENANCE', 'BANNER_TOP', 'BANNER_BOTTOM')),
+ (_('Pagination'), ('PAGINATE_COUNT', 'MAX_PAGE_SIZE')),
+ (_('Validation'), ('CUSTOM_VALIDATORS', 'PROTECTION_RULES')),
+ (_('User Preferences'), ('DEFAULT_USER_PREFERENCES',)),
+ (_('Miscellaneous'), (
+ 'MAINTENANCE_MODE', 'GRAPHQL_ENABLED', 'CHANGELOG_RETENTION', 'JOB_RETENTION', 'MAPS_URL',
+ )),
+ (_('Config Revision'), ('comment',))
+ )
+
+ class Meta:
+ model = ConfigRevision
+ fields = '__all__'
+ widgets = {
+ 'BANNER_LOGIN': forms.Textarea(attrs={'class': 'font-monospace'}),
+ 'BANNER_MAINTENANCE': forms.Textarea(attrs={'class': 'font-monospace'}),
+ 'BANNER_TOP': forms.Textarea(attrs={'class': 'font-monospace'}),
+ 'BANNER_BOTTOM': forms.Textarea(attrs={'class': 'font-monospace'}),
+ 'CUSTOM_VALIDATORS': forms.Textarea(attrs={'class': 'font-monospace'}),
+ 'PROTECTION_RULES': forms.Textarea(attrs={'class': 'font-monospace'}),
+ 'comment': forms.Textarea(),
+ }
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ # Append current parameter values to form field help texts and check for static configurations
+ config = get_config()
+ for param in PARAMS:
+ value = getattr(config, param.name)
+
+ # Set the field's initial value, if it can be serialized. (This may not be the case e.g. for
+ # CUSTOM_VALIDATORS, which may reference Python objects.)
+ try:
+ json.dumps(value)
+ if type(value) in (tuple, list):
+ self.fields[param.name].initial = ', '.join(value)
+ else:
+ self.fields[param.name].initial = value
+ except TypeError:
+ pass
+
+ # Check whether this parameter is statically configured (e.g. in configuration.py)
+ if hasattr(settings, param.name):
+ self.fields[param.name].disabled = True
+ self.fields[param.name].help_text = _(
+ 'This parameter has been defined statically and cannot be modified.'
+ )
+ continue
+
+ # Set the field's help text
+ help_text = self.fields[param.name].help_text
+ if help_text:
+ help_text += 'choice1:First Choice
')
)
class Meta:
model = CustomFieldChoiceSet
fields = ('name', 'description', 'base_choices', 'extra_choices', 'order_alphabetically')
+ def __init__(self, *args, initial=None, **kwargs):
+ super().__init__(*args, initial=initial, **kwargs)
+
+ # Escape colons in extra_choices
+ if 'extra_choices' in self.initial and self.initial['extra_choices']:
+ choices = []
+ for choice in self.initial['extra_choices']:
+ choice = (choice[0].replace(':', '\\:'), choice[1].replace(':', '\\:'))
+ choices.append(choice)
+
+ self.initial['extra_choices'] = choices
+
def clean_extra_choices(self):
data = []
for line in self.cleaned_data['extra_choices'].splitlines():
try:
- value, label = line.split(',', maxsplit=1)
+ value, label = re.split(r'(?JSON format.')
+ )
+ action_data = JSONField(
+ required=False,
+ help_text=_('Enter parameters to pass to the action in JSON format.')
+ )
+
+ fieldsets = (
+ (_('Event Rule'), ('name', 'description', 'content_types', 'enabled', 'tags')),
+ (_('Events'), ('type_create', 'type_update', 'type_delete', 'type_job_start', 'type_job_end')),
+ (_('Conditions'), ('conditions',)),
+ (_('Action'), (
+ 'action_type', 'action_choice', 'action_object_type', 'action_object_id', 'action_data',
+ )),
+ )
+
+ class Meta:
+ model = EventRule
+ fields = (
+ 'content_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'
+ )
labels = {
'type_create': _('Creations'),
'type_update': _('Updates'),
@@ -250,18 +288,90 @@ class WebhookForm(NetBoxModelForm):
'type_job_end': _('Job terminations'),
}
widgets = {
- 'additional_headers': forms.Textarea(attrs={'class': 'font-monospace'}),
- 'body_template': forms.Textarea(attrs={'class': 'font-monospace'}),
'conditions': forms.Textarea(attrs={'class': 'font-monospace'}),
+ 'action_type': HTMXSelect(),
+ 'action_object_type': forms.HiddenInput,
+ 'action_object_id': forms.HiddenInput,
}
+ def init_script_choice(self):
+ choices = []
+ for module in ScriptModule.objects.all():
+ scripts = []
+ for script_name in module.scripts.keys():
+ name = f"{str(module.pk)}:{script_name}"
+ scripts.append((name, script_name))
+ if scripts:
+ choices.append((str(module), scripts))
+ self.fields['action_choice'].choices = choices
+
+ if self.instance.action_type == EventRuleActionChoices.SCRIPT and self.instance.action_parameters:
+ scriptmodule_id = self.instance.action_object_id
+ script_name = self.instance.action_parameters.get('script_name')
+ self.fields['action_choice'].initial = f'{scriptmodule_id}:{script_name}'
+
+ def init_webhook_choice(self):
+ initial = None
+ if self.instance.action_type == EventRuleActionChoices.WEBHOOK:
+ webhook_id = get_field_value(self, 'action_object_id')
+ initial = Webhook.objects.get(pk=webhook_id) if webhook_id else None
+ self.fields['action_choice'] = DynamicModelChoiceField(
+ label=_('Webhook'),
+ queryset=Webhook.objects.all(),
+ required=True,
+ initial=initial
+ )
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.fields['action_object_type'].required = False
+ self.fields['action_object_id'].required = False
+
+ # Determine the action type
+ action_type = get_field_value(self, 'action_type')
+
+ if action_type == EventRuleActionChoices.WEBHOOK:
+ self.init_webhook_choice()
+ elif action_type == EventRuleActionChoices.SCRIPT:
+ self.init_script_choice()
+
+ def clean(self):
+ super().clean()
+
+ action_choice = self.cleaned_data.get('action_choice')
+ # Webhook
+ if self.cleaned_data.get('action_type') == EventRuleActionChoices.WEBHOOK:
+ self.cleaned_data['action_object_type'] = ContentType.objects.get_for_model(action_choice)
+ self.cleaned_data['action_object_id'] = action_choice.id
+ # Script
+ elif self.cleaned_data.get('action_type') == EventRuleActionChoices.SCRIPT:
+ self.cleaned_data['action_object_type'] = ContentType.objects.get_for_model(
+ ScriptModule,
+ for_concrete_model=False
+ )
+ module_id, script_name = action_choice.split(":", maxsplit=1)
+ self.cleaned_data['action_object_id'] = module_id
+
+ return self.cleaned_data
+
+ def save(self, *args, **kwargs):
+ # Set action_parameters on the instance
+ if self.cleaned_data['action_type'] == EventRuleActionChoices.SCRIPT:
+ module_id, script_name = self.cleaned_data.get('action_choice').split(":", maxsplit=1)
+ self.instance.action_parameters = {
+ 'script_name': script_name,
+ }
+ else:
+ self.instance.action_parameters = None
+
+ return super().save(*args, **kwargs)
+
class TagForm(BootstrapMixin, forms.ModelForm):
slug = SlugField()
object_types = ContentTypeMultipleChoiceField(
label=_('Object types'),
- queryset=ContentType.objects.all(),
- limit_choices_to=FeatureQuery('tags'),
+ queryset=ContentType.objects.with_feature('tags'),
required=False
)
@@ -455,115 +565,3 @@ class JournalEntryForm(NetBoxModelForm):
'assigned_object_type': forms.HiddenInput,
'assigned_object_id': forms.HiddenInput,
}
-
-
-EMPTY_VALUES = ('', None, [], ())
-
-
-class ConfigFormMetaclass(forms.models.ModelFormMetaclass):
-
- def __new__(mcs, name, bases, attrs):
-
- # Emulate a declared field for each supported configuration parameter
- param_fields = {}
- for param in PARAMS:
- field_kwargs = {
- 'required': False,
- 'label': param.label,
- 'help_text': param.description,
- }
- field_kwargs.update(**param.field_kwargs)
- param_fields[param.name] = param.field(**field_kwargs)
- attrs.update(param_fields)
-
- return super().__new__(mcs, name, bases, attrs)
-
-
-class ConfigRevisionForm(BootstrapMixin, forms.ModelForm, metaclass=ConfigFormMetaclass):
- """
- Form for creating a new ConfigRevision.
- """
-
- fieldsets = (
- (_('Rack Elevations'), ('RACK_ELEVATION_DEFAULT_UNIT_HEIGHT', 'RACK_ELEVATION_DEFAULT_UNIT_WIDTH')),
- (_('Power'), ('POWERFEED_DEFAULT_VOLTAGE', 'POWERFEED_DEFAULT_AMPERAGE', 'POWERFEED_DEFAULT_MAX_UTILIZATION')),
- (_('IPAM'), ('ENFORCE_GLOBAL_UNIQUE', 'PREFER_IPV4')),
- (_('Security'), ('ALLOWED_URL_SCHEMES',)),
- (_('Banners'), ('BANNER_LOGIN', 'BANNER_MAINTENANCE', 'BANNER_TOP', 'BANNER_BOTTOM')),
- (_('Pagination'), ('PAGINATE_COUNT', 'MAX_PAGE_SIZE')),
- (_('Validation'), ('CUSTOM_VALIDATORS',)),
- (_('User Preferences'), ('DEFAULT_USER_PREFERENCES',)),
- (_('Miscellaneous'), (
- 'MAINTENANCE_MODE', 'GRAPHQL_ENABLED', 'CHANGELOG_RETENTION', 'JOB_RETENTION', 'MAPS_URL',
- )),
- (_('Config Revision'), ('comment',))
- )
-
- class Meta:
- model = ConfigRevision
- fields = '__all__'
- widgets = {
- 'BANNER_LOGIN': forms.Textarea(attrs={'class': 'font-monospace'}),
- 'BANNER_MAINTENANCE': forms.Textarea(attrs={'class': 'font-monospace'}),
- 'BANNER_TOP': forms.Textarea(attrs={'class': 'font-monospace'}),
- 'BANNER_BOTTOM': forms.Textarea(attrs={'class': 'font-monospace'}),
- 'CUSTOM_VALIDATORS': forms.Textarea(attrs={'class': 'font-monospace'}),
- 'comment': forms.Textarea(),
- }
-
- def __init__(self, *args, **kwargs):
- super().__init__(*args, **kwargs)
-
- # Append current parameter values to form field help texts and check for static configurations
- config = get_config()
- for param in PARAMS:
- value = getattr(config, param.name)
-
- # Set the field's initial value, if it can be serialized. (This may not be the case e.g. for
- # CUSTOM_VALIDATORS, which may reference Python objects.)
- try:
- json.dumps(value)
- if type(value) in (tuple, list):
- self.fields[param.name].initial = ', '.join(value)
- else:
- self.fields[param.name].initial = value
- except TypeError:
- pass
-
- # Check whether this parameter is statically configured (e.g. in configuration.py)
- if hasattr(settings, param.name):
- self.fields[param.name].disabled = True
- self.fields[param.name].help_text = _(
- 'This parameter has been defined statically and cannot be modified.'
- )
- continue
-
- # Set the field's help text
- help_text = self.fields[param.name].help_text
- if help_text:
- help_text += '{stacktrace}") logger.error(f"Exception raised during report execution: {e}") - job.terminate(status=JobStatusChoices.STATUS_ERRORED) + job.terminate(status=JobStatusChoices.STATUS_ERRORED, error=repr(e)) # Perform any post-run tasks self.post_run() diff --git a/netbox/extras/scripts.py b/netbox/extras/scripts.py index e93326ddc..f28465547 100644 --- a/netbox/extras/scripts.py +++ b/netbox/extras/scripts.py @@ -17,13 +17,13 @@ from core.models import Job from extras.api.serializers import ScriptOutputSerializer from extras.choices import LogLevelChoices from extras.models import ScriptModule -from extras.signals import clear_webhooks +from extras.signals import clear_events from ipam.formfields import IPAddressFormField, IPNetworkFormField from ipam.validators import MaxPrefixLengthValidator, MinPrefixLengthValidator, prefix_validator from utilities.exceptions import AbortScript, AbortTransaction from utilities.forms import add_blank_choice from utilities.forms.fields import DynamicModelChoiceField, DynamicModelMultipleChoiceField -from .context_managers import change_logging +from .context_managers import event_tracking from .forms import ScriptForm __all__ = ( @@ -472,10 +472,16 @@ def get_module_and_script(module_name, script_name): return module, script -def run_script(data, request, job, commit=True, **kwargs): +def run_script(data, job, request=None, commit=True, **kwargs): """ A wrapper for calling Script.run(). This performs error handling and provides a hook for committing changes. It exists outside the Script class to ensure it cannot be overridden by a script author. + + Args: + data: A dictionary of data to be passed to the script upon execution + job: The Job associated with this execution + request: The WSGI request associated with this execution (if any) + commit: Passed through to Script.run() """ job.start() @@ -486,9 +492,10 @@ def run_script(data, request, job, commit=True, **kwargs): logger.info(f"Running script (commit={commit})") # Add files to form data - files = request.FILES - for field_name, fileobj in files.items(): - data[field_name] = fileobj + if request: + files = request.FILES + for field_name, fileobj in files.items(): + data[field_name] = fileobj # Add the current request as a property of the script script.request = request @@ -496,7 +503,7 @@ def run_script(data, request, job, commit=True, **kwargs): def _run_script(): """ Core script execution task. We capture this within a subfunction to allow for conditionally wrapping it with - the change_logging context manager (which is bypassed if commit == False). + the event_tracking context manager (which is bypassed if commit == False). """ try: try: @@ -506,7 +513,8 @@ def run_script(data, request, job, commit=True, **kwargs): raise AbortTransaction() except AbortTransaction: script.log_info("Database changes have been reverted automatically.") - clear_webhooks.send(request) + if request: + clear_events.send(request) job.data = ScriptOutputSerializer(script).data job.terminate() except Exception as e: @@ -519,15 +527,16 @@ def run_script(data, request, job, commit=True, **kwargs): logger.error(f"Exception raised during script execution: {e}") script.log_info("Database changes have been reverted due to error.") job.data = ScriptOutputSerializer(script).data - job.terminate(status=JobStatusChoices.STATUS_ERRORED) - clear_webhooks.send(request) + job.terminate(status=JobStatusChoices.STATUS_ERRORED, error=repr(e)) + if request: + clear_events.send(request) logger.info(f"Script completed in {job.duration}") - # Execute the script. If commit is True, wrap it with the change_logging context manager to ensure we process - # change logging, webhooks, etc. + # Execute the script. If commit is True, wrap it with the event_tracking context manager to ensure we process + # change logging, event rules, etc. if commit: - with change_logging(request): + with event_tracking(request): _run_script() else: _run_script() diff --git a/netbox/extras/search.py b/netbox/extras/search.py index da4aa1c84..fff59fa77 100644 --- a/netbox/extras/search.py +++ b/netbox/extras/search.py @@ -9,3 +9,14 @@ class JournalEntryIndex(SearchIndex): ('comments', 5000), ) category = 'Journal' + display_attrs = ('kind', 'created_by') + + +@register_search +class WebhookEntryIndex(SearchIndex): + model = models.Webhook + fields = ( + ('name', 100), + ('description', 500), + ) + display_attrs = ('description',) diff --git a/netbox/extras/signals.py b/netbox/extras/signals.py index d6550309f..d1b20961a 100644 --- a/netbox/extras/signals.py +++ b/netbox/extras/signals.py @@ -2,25 +2,31 @@ import importlib import logging from django.contrib.contenttypes.models import ContentType +from django.core.exceptions import ValidationError from django.db.models.signals import m2m_changed, post_save, pre_delete from django.dispatch import receiver, Signal +from django.utils.translation import gettext_lazy as _ from django_prometheus.models import model_deletes, model_inserts, model_updates +from core.signals import job_end, job_start +from extras.constants import EVENT_JOB_END, EVENT_JOB_START +from extras.events import process_event_rules +from extras.models import EventRule from extras.validators import CustomValidator from netbox.config import get_config -from netbox.context import current_request, webhooks_queue +from netbox.context import current_request, events_queue from netbox.signals import post_clean from utilities.exceptions import AbortRequest from .choices import ObjectChangeActionChoices -from .models import ConfigRevision, CustomField, ObjectChange, TaggedItem -from .webhooks import enqueue_object, get_snapshots, serialize_for_webhook +from .events import enqueue_object, get_snapshots, serialize_for_event +from .models import CustomField, ObjectChange, TaggedItem # # Change logging/webhooks # -# Define a custom signal that can be sent to clear any queued webhooks -clear_webhooks = Signal() +# Define a custom signal that can be sent to clear any queued events +clear_events = Signal() def is_same_object(instance, webhook_data, request_id): @@ -63,30 +69,30 @@ def handle_changed_object(sender, instance, **kwargs): return # Record an ObjectChange if applicable - if hasattr(instance, 'to_objectchange'): - if m2m_changed: - ObjectChange.objects.filter( - changed_object_type=ContentType.objects.get_for_model(instance), - changed_object_id=instance.pk, - request_id=request.id - ).update( - postchange_data=instance.to_objectchange(action).postchange_data - ) - else: - objectchange = instance.to_objectchange(action) + if m2m_changed: + ObjectChange.objects.filter( + changed_object_type=ContentType.objects.get_for_model(instance), + changed_object_id=instance.pk, + request_id=request.id + ).update( + postchange_data=instance.to_objectchange(action).postchange_data + ) + else: + objectchange = instance.to_objectchange(action) + if objectchange and objectchange.has_changes: objectchange.user = request.user objectchange.request_id = request.id objectchange.save() # If this is an M2M change, update the previously queued webhook (from post_save) - queue = webhooks_queue.get() + queue = events_queue.get() if m2m_changed and queue and is_same_object(instance, queue[-1], request.id): instance.refresh_from_db() # Ensure that we're working with fresh M2M assignments - queue[-1]['data'] = serialize_for_webhook(instance) + queue[-1]['data'] = serialize_for_event(instance) queue[-1]['snapshots']['postchange'] = get_snapshots(instance, action)['postchange'] else: enqueue_object(queue, instance, request.user, request.id, action) - webhooks_queue.set(queue) + events_queue.set(queue) # Increment metric counters if action == ObjectChangeActionChoices.ACTION_CREATE: @@ -115,22 +121,22 @@ def handle_deleted_object(sender, instance, **kwargs): objectchange.save() # Enqueue webhooks - queue = webhooks_queue.get() + queue = events_queue.get() enqueue_object(queue, instance, request.user, request.id, ObjectChangeActionChoices.ACTION_DELETE) - webhooks_queue.set(queue) + events_queue.set(queue) # Increment metric counters model_deletes.labels(instance._meta.model_name).inc() -@receiver(clear_webhooks) -def clear_webhook_queue(sender, **kwargs): +@receiver(clear_events) +def clear_events_queue(sender, **kwargs): """ - Delete any queued webhooks (e.g. because of an aborted bulk transaction) + Delete any queued events (e.g. because of an aborted bulk transaction) """ - logger = logging.getLogger('webhooks') - logger.info(f"Clearing {len(webhooks_queue.get())} queued webhooks ({sender})") - webhooks_queue.set([]) + logger = logging.getLogger('events') + logger.info(f"Clearing {len(events_queue.get())} queued events ({sender})") + events_queue.set([]) # @@ -178,11 +184,7 @@ m2m_changed.connect(handle_cf_removed_obj_types, sender=CustomField.content_type # Custom validation # -@receiver(post_clean) -def run_custom_validators(sender, instance, **kwargs): - config = get_config() - model_name = f'{sender._meta.app_label}.{sender._meta.model_name}' - validators = config.CUSTOM_VALIDATORS.get(model_name, []) +def run_validators(instance, validators): for validator in validators: @@ -198,16 +200,27 @@ def run_custom_validators(sender, instance, **kwargs): validator(instance) -# -# Dynamic configuration -# +@receiver(post_clean) +def run_save_validators(sender, instance, **kwargs): + model_name = f'{sender._meta.app_label}.{sender._meta.model_name}' + validators = get_config().CUSTOM_VALIDATORS.get(model_name, []) -@receiver(post_save, sender=ConfigRevision) -def update_config(sender, instance, **kwargs): - """ - Update the cached NetBox configuration when a new ConfigRevision is created. - """ - instance.activate() + run_validators(instance, validators) + + +@receiver(pre_delete) +def run_delete_validators(sender, instance, **kwargs): + model_name = f'{sender._meta.app_label}.{sender._meta.model_name}' + validators = get_config().PROTECTION_RULES.get(model_name, []) + + try: + run_validators(instance, validators) + except ValidationError as e: + raise AbortRequest( + _("Deletion is prevented by a protection rule: {message}").format( + message=e + ) + ) # @@ -226,3 +239,27 @@ def validate_assigned_tags(sender, instance, action, model, pk_set, **kwargs): for tag in model.objects.filter(pk__in=pk_set, object_types__isnull=False).prefetch_related('object_types'): if ct not in tag.object_types.all(): raise AbortRequest(f"Tag {tag} cannot be assigned to {ct.model} objects.") + + +# +# Event rules +# + +@receiver(job_start) +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, content_types=sender.object_type) + username = sender.user.username if sender.user else None + process_event_rules(event_rules, sender.object_type.model, EVENT_JOB_START, sender.data, username) + + +@receiver(job_end) +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, content_types=sender.object_type) + username = sender.user.username if sender.user else None + process_event_rules(event_rules, sender.object_type.model, EVENT_JOB_END, sender.data, username) diff --git a/netbox/extras/tables/tables.py b/netbox/extras/tables/tables.py index 9e14a2d27..8482c5e24 100644 --- a/netbox/extras/tables/tables.py +++ b/netbox/extras/tables/tables.py @@ -11,11 +11,11 @@ from .template_code import * __all__ = ( 'BookmarkTable', 'ConfigContextTable', - 'ConfigRevisionTable', 'ConfigTemplateTable', 'CustomFieldChoiceSetTable', 'CustomFieldTable', 'CustomLinkTable', + 'EventRuleTable', 'ExportTemplateTable', 'ImageAttachmentTable', 'JournalEntryTable', @@ -34,31 +34,6 @@ IMAGEATTACHMENT_IMAGE = ''' {% endif %} ''' -REVISION_BUTTONS = """ -{% if not record.is_active %} - - - -{% endif %} -""" - - -class ConfigRevisionTable(NetBoxTable): - is_active = columns.BooleanColumn( - verbose_name=_('Is Active'), - ) - actions = columns.ActionsColumn( - actions=('delete',), - extra_buttons=REVISION_BUTTONS - ) - - class Meta(NetBoxTable.Meta): - model = ConfigRevision - fields = ( - 'pk', 'id', 'is_active', 'created', 'comment', - ) - default_columns = ('pk', 'id', 'is_active', 'created', 'comment') - class CustomFieldTable(NetBoxTable): name = tables.Column( @@ -71,8 +46,11 @@ class CustomFieldTable(NetBoxTable): required = columns.BooleanColumn( verbose_name=_('Required') ) - ui_visibility = columns.ChoiceFieldColumn( - verbose_name=_('UI Visibility') + ui_visible = columns.ChoiceFieldColumn( + verbose_name=_('Visible') + ) + ui_editable = columns.ChoiceFieldColumn( + verbose_name=_('Editable') ) description = columns.MarkdownColumn( verbose_name=_('Description') @@ -94,8 +72,8 @@ class CustomFieldTable(NetBoxTable): model = CustomField fields = ( 'pk', 'id', 'name', 'content_types', 'label', 'type', 'group_name', 'required', 'default', 'description', - 'search_weight', 'filter_logic', 'ui_visibility', 'is_cloneable', 'weight', 'choice_set', 'choices', - 'created', 'last_updated', + 'search_weight', 'filter_logic', 'ui_visible', 'ui_editable', 'is_cloneable', 'weight', 'choice_set', + 'choices', 'created', 'last_updated', ) default_columns = ('pk', 'name', 'content_types', 'label', 'group_name', 'type', 'required', 'description') @@ -273,6 +251,36 @@ class WebhookTable(NetBoxTable): verbose_name=_('Name'), linkify=True ) + ssl_validation = columns.BooleanColumn( + verbose_name=_('SSL Validation') + ) + tags = columns.TagColumn( + url_name='extras:webhook_list' + ) + + class Meta(NetBoxTable.Meta): + model = Webhook + fields = ( + 'pk', 'id', 'name', 'http_method', 'payload_url', 'http_content_type', 'secret', 'ssl_verification', + 'ca_file_path', 'description', 'tags', 'created', 'last_updated', + ) + default_columns = ( + 'pk', 'name', 'http_method', 'payload_url', 'description', + ) + + +class EventRuleTable(NetBoxTable): + name = tables.Column( + verbose_name=_('Name'), + linkify=True + ) + action_type = tables.Column( + verbose_name=_('Type'), + ) + action_object = tables.Column( + linkify=True, + verbose_name=_('Object'), + ) content_types = columns.ContentTypesColumn( verbose_name=_('Content Types'), ) @@ -294,23 +302,20 @@ class WebhookTable(NetBoxTable): type_job_end = columns.BooleanColumn( verbose_name=_('Job End') ) - ssl_validation = columns.BooleanColumn( - verbose_name=_('SSL Validation') - ) tags = columns.TagColumn( url_name='extras:webhook_list' ) class Meta(NetBoxTable.Meta): - model = Webhook + model = EventRule fields = ( - 'pk', 'id', 'name', 'content_types', 'enabled', 'type_create', 'type_update', 'type_delete', - 'type_job_start', 'type_job_end', 'http_method', 'payload_url', 'secret', 'ssl_validation', 'ca_file_path', - 'tags', 'created', 'last_updated', + 'pk', 'id', 'name', 'enabled', 'description', 'action_type', 'action_object', 'content_types', + 'type_create', 'type_update', 'type_delete', 'type_job_start', 'type_job_end', 'tags', 'created', + 'last_updated', ) default_columns = ( - 'pk', 'name', 'content_types', 'enabled', 'type_create', 'type_update', 'type_delete', 'type_job_start', - 'type_job_end', 'http_method', 'payload_url', + 'pk', 'name', 'enabled', 'action_type', 'action_object', 'content_types', 'type_create', 'type_update', + 'type_delete', 'type_job_start', 'type_job_end', ) diff --git a/netbox/extras/tests/test_api.py b/netbox/extras/tests/test_api.py index 255457f21..f40372a8f 100644 --- a/netbox/extras/tests/test_api.py +++ b/netbox/extras/tests/test_api.py @@ -8,12 +8,12 @@ from rest_framework import status from core.choices import ManagedFileRootPathChoices from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Rack, Location, RackRole, Site +from extras.choices import * from extras.models import * from extras.reports import Report from extras.scripts import BooleanVar, IntegerVar, Script, StringVar from utilities.testing import APITestCase, APIViewTestCases - User = get_user_model() @@ -32,53 +32,119 @@ class WebhookTest(APIViewTestCases.APIViewTestCase): brief_fields = ['display', 'id', 'name', 'url'] create_data = [ { - 'content_types': ['dcim.device', 'dcim.devicetype'], 'name': 'Webhook 4', - 'type_create': True, 'payload_url': 'http://example.com/?4', }, { - 'content_types': ['dcim.device', 'dcim.devicetype'], 'name': 'Webhook 5', - 'type_update': True, 'payload_url': 'http://example.com/?5', }, { - 'content_types': ['dcim.device', 'dcim.devicetype'], 'name': 'Webhook 6', - 'type_delete': True, 'payload_url': 'http://example.com/?6', }, ] bulk_update_data = { + 'description': 'New description', 'ssl_verification': False, } @classmethod def setUpTestData(cls): - site_ct = ContentType.objects.get_for_model(Site) - rack_ct = ContentType.objects.get_for_model(Rack) webhooks = ( Webhook( name='Webhook 1', - type_create=True, payload_url='http://example.com/?1', ), Webhook( name='Webhook 2', - type_update=True, payload_url='http://example.com/?1', ), Webhook( name='Webhook 3', - type_delete=True, payload_url='http://example.com/?1', ), ) Webhook.objects.bulk_create(webhooks) - for webhook in webhooks: - webhook.content_types.add(site_ct, rack_ct) + + +class EventRuleTest(APIViewTestCases.APIViewTestCase): + model = EventRule + brief_fields = ['display', 'id', 'name', 'url'] + bulk_update_data = { + 'enabled': False, + 'description': 'New description', + } + update_data = { + 'name': 'Event Rule X', + 'enabled': False, + 'description': 'New description', + } + + @classmethod + def setUpTestData(cls): + webhooks = ( + Webhook( + name='Webhook 1', + payload_url='http://example.com/?1', + ), + Webhook( + name='Webhook 2', + payload_url='http://example.com/?1', + ), + Webhook( + name='Webhook 3', + payload_url='http://example.com/?1', + ), + Webhook( + name='Webhook 4', + payload_url='http://example.com/?1', + ), + Webhook( + name='Webhook 5', + payload_url='http://example.com/?1', + ), + Webhook( + name='Webhook 6', + payload_url='http://example.com/?1', + ), + ) + 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.objects.bulk_create(event_rules) + + cls.create_data = [ + { + 'name': 'EventRule 4', + 'content_types': ['dcim.device', 'dcim.devicetype'], + 'type_create': True, + 'action_type': EventRuleActionChoices.WEBHOOK, + 'action_object_type': 'extras.webhook', + 'action_object_id': webhooks[3].pk, + }, + { + 'name': 'EventRule 5', + 'content_types': ['dcim.device', 'dcim.devicetype'], + 'type_create': True, + 'action_type': EventRuleActionChoices.WEBHOOK, + 'action_object_type': 'extras.webhook', + 'action_object_id': webhooks[4].pk, + }, + { + 'name': 'EventRule 6', + 'content_types': ['dcim.device', 'dcim.devicetype'], + 'type_create': True, + 'action_type': EventRuleActionChoices.WEBHOOK, + 'action_object_type': 'extras.webhook', + 'action_object_id': webhooks[5].pk, + }, + ] class CustomFieldTest(APIViewTestCases.APIViewTestCase): @@ -184,6 +250,23 @@ class CustomFieldChoiceSetTest(APIViewTestCases.APIViewTestCase): ) CustomFieldChoiceSet.objects.bulk_create(choice_sets) + def test_invalid_choice_items(self): + """ + Attempting to define each choice as a single-item list should return a 400 error. + """ + self.add_permissions('extras.add_customfieldchoiceset') + data = { + "name": "test", + "extra_choices": [ + ["choice1"], + ["choice2"], + ["choice3"], + ] + } + + response = self.client.post(self._get_list_url(), data, format='json', **self.header) + self.assertEqual(response.status_code, 400) + class CustomLinkTest(APIViewTestCases.APIViewTestCase): model = CustomLink diff --git a/netbox/extras/tests/test_changelog.py b/netbox/extras/tests/test_changelog.py index 34fd72b2b..e144c5dee 100644 --- a/netbox/extras/tests/test_changelog.py +++ b/netbox/extras/tests/test_changelog.py @@ -1,4 +1,5 @@ from django.contrib.contenttypes.models import ContentType +from django.test import override_settings from django.urls import reverse from rest_framework import status @@ -207,6 +208,66 @@ class ChangeLogViewTest(ModelViewTestCase): self.assertEqual(objectchange.prechange_data['slug'], sites[0].slug) self.assertEqual(objectchange.postchange_data, None) + @override_settings(CHANGELOG_SKIP_EMPTY_CHANGES=False) + def test_update_object_change(self): + # Create a Site + site = Site.objects.create( + name='Site 1', + slug='site-1', + status=SiteStatusChoices.STATUS_PLANNED, + custom_field_data={ + 'cf1': None, + 'cf2': None + } + ) + + # Update it with the same field values + form_data = { + 'name': site.name, + 'slug': site.slug, + 'status': SiteStatusChoices.STATUS_PLANNED, + } + request = { + 'path': self._get_url('edit', instance=site), + 'data': post_data(form_data), + } + self.add_permissions('dcim.change_site', 'extras.view_tag') + response = self.client.post(**request) + self.assertHttpStatus(response, 302) + + # Check that an ObjectChange record has been created + self.assertEqual(ObjectChange.objects.count(), 1) + + @override_settings(CHANGELOG_SKIP_EMPTY_CHANGES=True) + def test_update_object_nochange(self): + # Create a Site + site = Site.objects.create( + name='Site 1', + slug='site-1', + status=SiteStatusChoices.STATUS_PLANNED, + custom_field_data={ + 'cf1': None, + 'cf2': None + } + ) + + # Update it with the same field values + form_data = { + 'name': site.name, + 'slug': site.slug, + 'status': SiteStatusChoices.STATUS_PLANNED, + } + request = { + 'path': self._get_url('edit', instance=site), + 'data': post_data(form_data), + } + self.add_permissions('dcim.change_site', 'extras.view_tag') + response = self.client.post(**request) + self.assertHttpStatus(response, 302) + + # Check that no ObjectChange records have been created + self.assertEqual(ObjectChange.objects.count(), 0) + class ChangeLogAPITest(APITestCase): diff --git a/netbox/extras/tests/test_custom_validation.py b/netbox/extras/tests/test_custom_validation.py new file mode 100644 index 000000000..e375b49f5 --- /dev/null +++ b/netbox/extras/tests/test_custom_validation.py @@ -0,0 +1,265 @@ +from django.test import TestCase +from django.test import override_settings + +from circuits.api.serializers import ProviderSerializer +from circuits.forms import ProviderForm +from circuits.models import Provider +from ipam.models import ASN, RIR +from utilities.choices import CSVDelimiterChoices, ImportFormatChoices +from utilities.testing import APITestCase, ModelViewTestCase, create_tags, post_data + + +class ModelFormCustomValidationTest(TestCase): + + @override_settings(CUSTOM_VALIDATORS={ + 'circuits.provider': [ + {'tags': {'required': True}} + ] + }) + def test_tags_validation(self): + """ + Check that custom validation rules work for tag assignment. + """ + data = { + 'name': 'Provider 1', + 'slug': 'provider-1', + } + form = ProviderForm(data) + self.assertFalse(form.is_valid()) + + tags = create_tags('Tag1', 'Tag2', 'Tag3') + data['tags'] = [tag.pk for tag in tags] + form = ProviderForm(data) + self.assertTrue(form.is_valid()) + + @override_settings(CUSTOM_VALIDATORS={ + 'circuits.provider': [ + {'asns': {'required': True}} + ] + }) + def test_m2m_validation(self): + """ + Check that custom validation rules work for many-to-many fields. + """ + data = { + 'name': 'Provider 1', + 'slug': 'provider-1', + } + form = ProviderForm(data) + self.assertFalse(form.is_valid()) + + rir = RIR.objects.create(name='RIR 1', slug='rir-1') + asns = ASN.objects.bulk_create(( + ASN(rir=rir, asn=65001), + ASN(rir=rir, asn=65002), + ASN(rir=rir, asn=65003), + )) + data['asns'] = [asn.pk for asn in asns] + form = ProviderForm(data) + self.assertTrue(form.is_valid()) + + +class BulkEditCustomValidationTest(ModelViewTestCase): + model = Provider + + @classmethod + def setUpTestData(cls): + rir = RIR.objects.create(name='RIR 1', slug='rir-1') + asns = ASN.objects.bulk_create(( + ASN(rir=rir, asn=65001), + ASN(rir=rir, asn=65002), + ASN(rir=rir, asn=65003), + )) + + providers = ( + Provider(name='Provider 1', slug='provider-1'), + Provider(name='Provider 2', slug='provider-2'), + Provider(name='Provider 3', slug='provider-3'), + ) + Provider.objects.bulk_create(providers) + for provider in providers: + provider.asns.set(asns) + + @override_settings(CUSTOM_VALIDATORS={ + 'circuits.provider': [ + {'asns': {'required': True}} + ] + }) + def test_bulk_edit_without_m2m(self): + """ + Check that custom validation rules do not interfere with bulk editing. + """ + data = { + 'pk': list(Provider.objects.values_list('pk', flat=True)), + '_apply': '', + 'description': 'New description', + } + self.add_permissions( + 'circuits.view_provider', + 'circuits.change_provider', + ) + + # Bulk edit the description without changing ASN assignments + request = { + 'path': self._get_url('bulk_edit'), + 'data': post_data(data), + } + response = self.client.post(**request) + self.assertHttpStatus(response, 302) + self.assertEqual( + Provider.objects.filter(description=data['description']).count(), + len(data['pk']) + ) + + @override_settings(CUSTOM_VALIDATORS={ + 'circuits.provider': [ + {'asns': {'required': True}} + ] + }) + def test_bulk_edit_m2m(self): + """ + Test that custom validation rules are enforced during bulk editing. + """ + data = { + 'pk': list(Provider.objects.values_list('pk', flat=True)), + '_apply': '', + 'description': 'New description', + } + self.add_permissions( + 'circuits.view_provider', + 'circuits.change_provider', + 'ipam.view_asn', + ) + + # Change the ASN assignments + asn = ASN.objects.first() + data['asns'] = [asn.pk] + request = { + 'path': self._get_url('bulk_edit'), + 'data': post_data(data), + } + response = self.client.post(**request) + self.assertHttpStatus(response, 302) + for provider in Provider.objects.all(): + self.assertEqual(len(provider.asns.all()), 1) + + # Attempt to remove the ASN assignments + data.pop('asns') + data['_nullify'] = 'asns' + request = { + 'path': self._get_url('bulk_edit'), + 'data': post_data(data), + } + response = self.client.post(**request) + self.assertHttpStatus(response, 200) + for provider in Provider.objects.all(): + self.assertTrue(provider.asns.exists()) + + +class BulkImportCustomValidationTest(ModelViewTestCase): + model = Provider + + @classmethod + def setUpTestData(cls): + create_tags('Tag1', 'Tag2', 'Tag3') + + @override_settings(CUSTOM_VALIDATORS={ + 'circuits.provider': [ + {'tags': {'required': True}} + ] + }) + def test_bulk_import_invalid(self): + """ + Test that custom validation rules are enforced during bulk import. + """ + csv_data = ( + "name,slug", + "Provider 1,provider-1", + "Provider 2,provider-2", + "Provider 3,provider-3", + ) + data = { + 'data': '\n'.join(csv_data), + 'format': ImportFormatChoices.CSV, + 'csv_delimiter': CSVDelimiterChoices.COMMA, + } + self.add_permissions( + 'circuits.view_provider', + 'circuits.add_provider', + 'extras.view_tag', + ) + + # Attempt to import providers without tags + request = { + 'path': self._get_url('import'), + 'data': post_data(data), + } + response = self.client.post(**request) + self.assertHttpStatus(response, 200) + self.assertFalse(Provider.objects.exists()) + + # Import providers successfully with tag assignments + csv_data = ( + "name,slug,tags", + "Provider 1,provider-1,tag1", + "Provider 2,provider-2,tag2", + "Provider 3,provider-3,tag3", + ) + data['data'] = '\n'.join(csv_data) + request = { + 'path': self._get_url('import'), + 'data': post_data(data), + } + response = self.client.post(**request) + self.assertHttpStatus(response, 302) + self.assertTrue(Provider.objects.exists()) + + +class APISerializerCustomValidationTest(APITestCase): + + @override_settings(CUSTOM_VALIDATORS={ + 'circuits.provider': [ + {'tags': {'required': True}} + ] + }) + def test_tags_validation(self): + """ + Check that custom validation rules work for tag assignment. + """ + data = { + 'name': 'Provider 1', + 'slug': 'provider-1', + } + serializer = ProviderSerializer(data=data) + self.assertFalse(serializer.is_valid()) + + tags = create_tags('Tag1', 'Tag2', 'Tag3') + data['tags'] = [tag.pk for tag in tags] + serializer = ProviderSerializer(data=data) + self.assertTrue(serializer.is_valid()) + + @override_settings(CUSTOM_VALIDATORS={ + 'circuits.provider': [ + {'asns': {'required': True}} + ] + }) + def test_m2m_validation(self): + """ + Check that custom validation rules work for many-to-many fields. + """ + data = { + 'name': 'Provider 1', + 'slug': 'provider-1', + } + serializer = ProviderSerializer(data=data) + self.assertFalse(serializer.is_valid()) + + rir = RIR.objects.create(name='RIR 1', slug='rir-1') + asns = ASN.objects.bulk_create(( + ASN(rir=rir, asn=65001), + ASN(rir=rir, asn=65002), + ASN(rir=rir, asn=65003), + )) + data['asns'] = [asn.pk for asn in asns] + serializer = ProviderSerializer(data=data) + self.assertTrue(serializer.is_valid()) diff --git a/netbox/extras/tests/test_customfields.py b/netbox/extras/tests/test_customfields.py index 7ac6b2035..574452a81 100644 --- a/netbox/extras/tests/test_customfields.py +++ b/netbox/extras/tests/test_customfields.py @@ -1329,7 +1329,7 @@ class CustomFieldModelFilterTest(TestCase): choice_set = CustomFieldChoiceSet.objects.create( name='Custom Field Choice Set 1', - extra_choices=(('a', 'A'), ('b', 'B'), ('c', 'C'), ('x', 'X')) + extra_choices=(('a', 'A'), ('b', 'B'), ('c', 'C')) ) # Integer filtering @@ -1435,7 +1435,7 @@ class CustomFieldModelFilterTest(TestCase): 'cf7': 'http://a.example.com', 'cf8': 'http://a.example.com', 'cf9': 'A', - 'cf10': ['A', 'X'], + 'cf10': ['A', 'B'], 'cf11': manufacturers[0].pk, 'cf12': [manufacturers[0].pk, manufacturers[3].pk], }), @@ -1449,7 +1449,7 @@ class CustomFieldModelFilterTest(TestCase): 'cf7': 'http://b.example.com', 'cf8': 'http://b.example.com', 'cf9': 'B', - 'cf10': ['B', 'X'], + 'cf10': ['B', 'C'], 'cf11': manufacturers[1].pk, 'cf12': [manufacturers[1].pk, manufacturers[3].pk], }), @@ -1463,7 +1463,7 @@ class CustomFieldModelFilterTest(TestCase): 'cf7': 'http://c.example.com', 'cf8': 'http://c.example.com', 'cf9': 'C', - 'cf10': ['C', 'X'], + 'cf10': None, 'cf11': manufacturers[2].pk, 'cf12': [manufacturers[2].pk, manufacturers[3].pk], }), @@ -1531,8 +1531,9 @@ class CustomFieldModelFilterTest(TestCase): self.assertEqual(self.filterset({'cf_cf9': ['A', 'B']}, self.queryset).qs.count(), 2) def test_filter_multiselect(self): - self.assertEqual(self.filterset({'cf_cf10': ['A', 'B']}, self.queryset).qs.count(), 2) - self.assertEqual(self.filterset({'cf_cf10': ['X']}, self.queryset).qs.count(), 3) + self.assertEqual(self.filterset({'cf_cf10': ['A']}, self.queryset).qs.count(), 1) + self.assertEqual(self.filterset({'cf_cf10': ['A', 'C']}, self.queryset).qs.count(), 2) + self.assertEqual(self.filterset({'cf_cf10': ['null']}, self.queryset).qs.count(), 1) def test_filter_object(self): manufacturer_ids = Manufacturer.objects.values_list('id', flat=True) diff --git a/netbox/extras/tests/test_customvalidator.py b/netbox/extras/tests/test_customvalidation.py similarity index 64% rename from netbox/extras/tests/test_customvalidator.py rename to netbox/extras/tests/test_customvalidation.py index 0fe507b67..d74ad599b 100644 --- a/netbox/extras/tests/test_customvalidator.py +++ b/netbox/extras/tests/test_customvalidation.py @@ -1,10 +1,13 @@ from django.conf import settings from django.core.exceptions import ValidationError +from django.db import transaction from django.test import TestCase, override_settings from ipam.models import ASN, RIR +from dcim.choices import SiteStatusChoices from dcim.models import Site from extras.validators import CustomValidator +from utilities.exceptions import AbortRequest class MyValidator(CustomValidator): @@ -14,6 +17,20 @@ class MyValidator(CustomValidator): self.fail("Name must be foo!") +eq_validator = CustomValidator({ + 'asn': { + 'eq': 100 + } +}) + + +neq_validator = CustomValidator({ + 'asn': { + 'neq': 100 + } +}) + + min_validator = CustomValidator({ 'asn': { 'min': 65000 @@ -77,6 +94,18 @@ class CustomValidatorTest(TestCase): validator = settings.CUSTOM_VALIDATORS['ipam.asn'][0] self.assertIsInstance(validator, CustomValidator) + @override_settings(CUSTOM_VALIDATORS={'ipam.asn': [eq_validator]}) + def test_eq(self): + ASN(asn=100, rir=RIR.objects.first()).clean() + with self.assertRaises(ValidationError): + ASN(asn=99, rir=RIR.objects.first()).clean() + + @override_settings(CUSTOM_VALIDATORS={'ipam.asn': [neq_validator]}) + def test_neq(self): + ASN(asn=99, rir=RIR.objects.first()).clean() + with self.assertRaises(ValidationError): + ASN(asn=100, rir=RIR.objects.first()).clean() + @override_settings(CUSTOM_VALIDATORS={'ipam.asn': [min_validator]}) def test_min(self): with self.assertRaises(ValidationError): @@ -147,7 +176,7 @@ class CustomValidatorConfigTest(TestCase): @override_settings( CUSTOM_VALIDATORS={ 'dcim.site': ( - 'extras.tests.test_customvalidator.MyValidator', + 'extras.tests.test_customvalidation.MyValidator', ) } ) @@ -159,3 +188,62 @@ class CustomValidatorConfigTest(TestCase): Site(name='foo', slug='foo').clean() with self.assertRaises(ValidationError): Site(name='bar', slug='bar').clean() + + +class ProtectionRulesConfigTest(TestCase): + + @override_settings( + PROTECTION_RULES={ + 'dcim.site': [ + {'status': {'eq': SiteStatusChoices.STATUS_DECOMMISSIONING}} + ] + } + ) + def test_plain_data(self): + """ + Test custom validator configuration using plain data (as opposed to a CustomValidator + class) + """ + # Create a site with a protected status + site = Site(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) + site.save() + + # Try to delete it + with self.assertRaises(AbortRequest): + with transaction.atomic(): + site.delete() + + # Change its status to an allowed value + site.status = SiteStatusChoices.STATUS_DECOMMISSIONING + site.save() + + # Deletion should now succeed + site.delete() + + @override_settings( + PROTECTION_RULES={ + 'dcim.site': ( + 'extras.tests.test_customvalidation.MyValidator', + ) + } + ) + def test_dotted_path(self): + """ + Test custom validator configuration using a dotted path (string) reference to a + CustomValidator class. + """ + # Create a site with a protected name + site = Site(name='bar', slug='bar') + site.save() + + # Try to delete it + with self.assertRaises(AbortRequest): + with transaction.atomic(): + site.delete() + + # Change the name to an allowed value + site.name = site.slug = 'foo' + site.save() + + # Deletion should now succeed + site.delete() diff --git a/netbox/extras/tests/test_webhooks.py b/netbox/extras/tests/test_event_rules.py similarity index 72% rename from netbox/extras/tests/test_webhooks.py rename to netbox/extras/tests/test_event_rules.py index ef7637765..549c33478 100644 --- a/netbox/extras/tests/test_webhooks.py +++ b/netbox/extras/tests/test_event_rules.py @@ -3,22 +3,21 @@ import uuid from unittest.mock import patch import django_rq +from dcim.choices import SiteStatusChoices +from dcim.models import Site from django.contrib.contenttypes.models import ContentType from django.http import HttpResponse from django.urls import reverse +from extras.choices import EventRuleActionChoices, ObjectChangeActionChoices +from extras.events import enqueue_object, flush_events, serialize_for_event +from extras.models import EventRule, Tag, Webhook +from extras.webhooks import generate_signature, send_webhook from requests import Session from rest_framework import status - -from dcim.choices import SiteStatusChoices -from dcim.models import Site -from extras.choices import ObjectChangeActionChoices -from extras.models import Tag, Webhook -from extras.webhooks import enqueue_object, flush_webhooks, generate_signature, serialize_for_webhook -from extras.webhooks_worker import eval_conditions, process_webhook from utilities.testing import APITestCase -class WebhookTest(APITestCase): +class EventRuleTest(APITestCase): def setUp(self): super().setUp() @@ -35,12 +34,37 @@ class WebhookTest(APITestCase): DUMMY_SECRET = 'LOOKATMEIMASECRETSTRING' webhooks = Webhook.objects.bulk_create(( - Webhook(name='Webhook 1', type_create=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET, additional_headers='X-Foo: Bar'), - Webhook(name='Webhook 2', type_update=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET), - Webhook(name='Webhook 3', type_delete=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET), + Webhook(name='Webhook 1', payload_url=DUMMY_URL, secret=DUMMY_SECRET, additional_headers='X-Foo: Bar'), + Webhook(name='Webhook 2', payload_url=DUMMY_URL, secret=DUMMY_SECRET), + Webhook(name='Webhook 3', payload_url=DUMMY_URL, secret=DUMMY_SECRET), )) - for webhook in webhooks: - webhook.content_types.set([site_ct]) + + ct = ContentType.objects.get(app_label='extras', model='webhook') + event_rules = EventRule.objects.bulk_create(( + EventRule( + name='Webhook Event 1', + type_create=True, + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=ct, + action_object_id=webhooks[0].id + ), + EventRule( + name='Webhook Event 2', + type_update=True, + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=ct, + action_object_id=webhooks[0].id + ), + EventRule( + name='Webhook Event 3', + type_delete=True, + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=ct, + action_object_id=webhooks[0].id + ), + )) + for event_rule in event_rules: + event_rule.content_types.set([site_ct]) Tag.objects.bulk_create(( Tag(name='Foo', slug='foo'), @@ -48,7 +72,42 @@ class WebhookTest(APITestCase): Tag(name='Baz', slug='baz'), )) - def test_enqueue_webhook_create(self): + def test_eventrule_conditions(self): + """ + Test evaluation of EventRule conditions. + """ + event_rule = EventRule( + name='Event Rule 1', + type_create=True, + type_update=True, + conditions={ + 'and': [ + { + 'attr': 'status.value', + 'value': 'active', + } + ] + } + ) + + # Create a Site to evaluate + site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_STAGING) + data = serialize_for_event(site) + + # Evaluate the conditions (status='staging') + self.assertFalse(event_rule.eval_conditions(data)) + + # Change the site's status + site.status = SiteStatusChoices.STATUS_ACTIVE + data = serialize_for_event(site) + + # Evaluate the conditions (status='active') + self.assertTrue(event_rule.eval_conditions(data)) + + def test_single_create_process_eventrule(self): + """ + Check that creating an object with an applicable EventRule queues a background task for the rule's action. + """ # Create an object via the REST API data = { 'name': 'Site 1', @@ -65,10 +124,10 @@ class WebhookTest(APITestCase): self.assertEqual(Site.objects.count(), 1) self.assertEqual(Site.objects.first().tags.count(), 2) - # Verify that a job was queued for the object creation webhook + # 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['webhook'], Webhook.objects.get(type_create=True)) + self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_create=True)) self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_CREATE) self.assertEqual(job.kwargs['model_name'], 'site') self.assertEqual(job.kwargs['data']['id'], response.data['id']) @@ -76,7 +135,11 @@ class WebhookTest(APITestCase): self.assertEqual(job.kwargs['snapshots']['postchange']['name'], 'Site 1') self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Bar', 'Foo']) - def test_enqueue_webhook_bulk_create(self): + def test_bulk_create_process_eventrule(self): + """ + Check that bulk creating multiple objects with an applicable EventRule queues a background task for each + new object. + """ # Create multiple objects via the REST API data = [ { @@ -111,10 +174,10 @@ class WebhookTest(APITestCase): self.assertEqual(Site.objects.count(), 3) self.assertEqual(Site.objects.first().tags.count(), 2) - # Verify that a webhook was queued for each object + # 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['webhook'], Webhook.objects.get(type_create=True)) + self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_create=True)) self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_CREATE) self.assertEqual(job.kwargs['model_name'], 'site') self.assertEqual(job.kwargs['data']['id'], response.data[i]['id']) @@ -122,7 +185,10 @@ class WebhookTest(APITestCase): self.assertEqual(job.kwargs['snapshots']['postchange']['name'], response.data[i]['name']) self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Bar', 'Foo']) - def test_enqueue_webhook_update(self): + def test_single_update_process_eventrule(self): + """ + Check that updating an object with an applicable EventRule queues a background task for the rule's action. + """ site = Site.objects.create(name='Site 1', slug='site-1') site.tags.set(Tag.objects.filter(name__in=['Foo', 'Bar'])) @@ -139,10 +205,10 @@ class WebhookTest(APITestCase): response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) - # Verify that a job was queued for the object update webhook + # 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['webhook'], Webhook.objects.get(type_update=True)) + self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_update=True)) self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_UPDATE) self.assertEqual(job.kwargs['model_name'], 'site') self.assertEqual(job.kwargs['data']['id'], site.pk) @@ -152,7 +218,11 @@ class WebhookTest(APITestCase): self.assertEqual(job.kwargs['snapshots']['postchange']['name'], 'Site X') self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Baz']) - def test_enqueue_webhook_bulk_update(self): + def test_bulk_update_process_eventrule(self): + """ + Check that bulk updating multiple objects with an applicable EventRule queues a background task for each + updated object. + """ sites = ( Site(name='Site 1', slug='site-1'), Site(name='Site 2', slug='site-2'), @@ -191,10 +261,10 @@ class WebhookTest(APITestCase): response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) - # Verify that a job was queued for the object update webhook + # 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['webhook'], Webhook.objects.get(type_update=True)) + self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_update=True)) self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_UPDATE) self.assertEqual(job.kwargs['model_name'], 'site') self.assertEqual(job.kwargs['data']['id'], data[i]['id']) @@ -204,7 +274,10 @@ class WebhookTest(APITestCase): self.assertEqual(job.kwargs['snapshots']['postchange']['name'], response.data[i]['name']) self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Baz']) - def test_enqueue_webhook_delete(self): + def test_single_delete_process_eventrule(self): + """ + Check that deleting an object with an applicable EventRule queues a background task for the rule's action. + """ site = Site.objects.create(name='Site 1', slug='site-1') site.tags.set(Tag.objects.filter(name__in=['Foo', 'Bar'])) @@ -214,17 +287,21 @@ class WebhookTest(APITestCase): response = self.client.delete(url, **self.header) self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT) - # Verify that a job was queued for the object update webhook + # 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['webhook'], Webhook.objects.get(type_delete=True)) + self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_delete=True)) self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_DELETE) self.assertEqual(job.kwargs['model_name'], 'site') self.assertEqual(job.kwargs['data']['id'], site.pk) self.assertEqual(job.kwargs['snapshots']['prechange']['name'], 'Site 1') self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo']) - def test_enqueue_webhook_bulk_delete(self): + def test_bulk_delete_process_eventrule(self): + """ + Check that bulk deleting multiple objects with an applicable EventRule queues a background task for each + deleted object. + """ sites = ( Site(name='Site 1', slug='site-1'), Site(name='Site 2', slug='site-2'), @@ -243,49 +320,17 @@ class WebhookTest(APITestCase): response = self.client.delete(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT) - # Verify that a job was queued for the object update webhook + # 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['webhook'], Webhook.objects.get(type_delete=True)) + self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(type_delete=True)) self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_DELETE) self.assertEqual(job.kwargs['model_name'], 'site') self.assertEqual(job.kwargs['data']['id'], sites[i].pk) self.assertEqual(job.kwargs['snapshots']['prechange']['name'], sites[i].name) self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo']) - def test_webhook_conditions(self): - # Create a conditional Webhook - webhook = Webhook( - name='Conditional Webhook', - type_create=True, - type_update=True, - payload_url='http://localhost:9000/', - conditions={ - 'and': [ - { - 'attr': 'status.value', - 'value': 'active', - } - ] - } - ) - - # Create a Site to evaluate - site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_STAGING) - data = serialize_for_webhook(site) - - # Evaluate the conditions (status='staging') - self.assertFalse(eval_conditions(webhook, data)) - - # Change the site's status - site.status = SiteStatusChoices.STATUS_ACTIVE - data = serialize_for_webhook(site) - - # Evaluate the conditions (status='active') - self.assertTrue(eval_conditions(webhook, data)) - - def test_webhooks_worker(self): - + def test_send_webhook(self): request_id = uuid.uuid4() def dummy_send(_, request, **kwargs): @@ -293,7 +338,8 @@ class WebhookTest(APITestCase): A dummy implementation of Session.send() to be used for testing. Always returns a 200 HTTP response. """ - webhook = Webhook.objects.get(type_create=True) + event = EventRule.objects.get(type_create=True) + webhook = event.action_object signature = generate_signature(request.body, webhook.secret) # Validate the outgoing request headers @@ -322,11 +368,11 @@ class WebhookTest(APITestCase): request_id=request_id, action=ObjectChangeActionChoices.ACTION_CREATE ) - flush_webhooks(webhooks_queue) + flush_events(webhooks_queue) # Retrieve the job from queue job = self.queue.jobs[0] # Patch the Session object with our dummy_send() method, then process the webhook for sending with patch.object(Session, 'send', dummy_send) as mock_send: - process_webhook(**job.kwargs) + send_webhook(**job.kwargs) diff --git a/netbox/extras/tests/test_filtersets.py b/netbox/extras/tests/test_filtersets.py index 69111e6a7..ef8aedcbd 100644 --- a/netbox/extras/tests/test_filtersets.py +++ b/netbox/extras/tests/test_filtersets.py @@ -6,6 +6,7 @@ from django.contrib.contenttypes.models import ContentType from django.test import TestCase from circuits.models import Provider +from core.choices import ManagedFileRootPathChoices from dcim.filtersets import SiteFilterSet from dcim.models import DeviceRole, DeviceType, Manufacturer, Platform, Rack, Region, Site, SiteGroup from dcim.models import Location @@ -40,7 +41,9 @@ class CustomFieldTestCase(TestCase, BaseFilterSetTests): required=True, weight=100, filter_logic=CustomFieldFilterLogicChoices.FILTER_LOOSE, - ui_visibility=CustomFieldVisibilityChoices.VISIBILITY_READ_WRITE + ui_visible=CustomFieldUIVisibleChoices.ALWAYS, + ui_editable=CustomFieldUIEditableChoices.YES, + description='foobar1' ), CustomField( name='Custom Field 2', @@ -48,7 +51,9 @@ class CustomFieldTestCase(TestCase, BaseFilterSetTests): required=False, weight=200, filter_logic=CustomFieldFilterLogicChoices.FILTER_EXACT, - ui_visibility=CustomFieldVisibilityChoices.VISIBILITY_READ_ONLY + ui_visible=CustomFieldUIVisibleChoices.IF_SET, + ui_editable=CustomFieldUIEditableChoices.NO, + description='foobar2' ), CustomField( name='Custom Field 3', @@ -56,7 +61,9 @@ class CustomFieldTestCase(TestCase, BaseFilterSetTests): required=False, weight=300, filter_logic=CustomFieldFilterLogicChoices.FILTER_DISABLED, - ui_visibility=CustomFieldVisibilityChoices.VISIBILITY_HIDDEN + ui_visible=CustomFieldUIVisibleChoices.HIDDEN, + ui_editable=CustomFieldUIEditableChoices.HIDDEN, + description='foobar3' ), CustomField( name='Custom Field 4', @@ -64,7 +71,8 @@ class CustomFieldTestCase(TestCase, BaseFilterSetTests): required=False, weight=400, filter_logic=CustomFieldFilterLogicChoices.FILTER_DISABLED, - ui_visibility=CustomFieldVisibilityChoices.VISIBILITY_HIDDEN, + ui_visible=CustomFieldUIVisibleChoices.HIDDEN, + ui_editable=CustomFieldUIEditableChoices.HIDDEN, choice_set=choice_sets[0] ), CustomField( @@ -73,7 +81,8 @@ class CustomFieldTestCase(TestCase, BaseFilterSetTests): required=False, weight=500, filter_logic=CustomFieldFilterLogicChoices.FILTER_DISABLED, - ui_visibility=CustomFieldVisibilityChoices.VISIBILITY_HIDDEN, + ui_visible=CustomFieldUIVisibleChoices.HIDDEN, + ui_editable=CustomFieldUIEditableChoices.HIDDEN, choice_set=choice_sets[1] ), ) @@ -84,6 +93,10 @@ class CustomFieldTestCase(TestCase, BaseFilterSetTests): custom_fields[3].content_types.add(ContentType.objects.get_by_natural_key('dcim', 'device')) custom_fields[4].content_types.add(ContentType.objects.get_by_natural_key('dcim', 'device')) + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Custom Field 1', 'Custom Field 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -106,8 +119,12 @@ class CustomFieldTestCase(TestCase, BaseFilterSetTests): params = {'filter_logic': CustomFieldFilterLogicChoices.FILTER_LOOSE} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) - def test_ui_visibility(self): - params = {'ui_visibility': CustomFieldVisibilityChoices.VISIBILITY_READ_WRITE} + def test_ui_visible(self): + params = {'ui_visible': CustomFieldUIVisibleChoices.ALWAYS} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + + def test_ui_editable(self): + params = {'ui_editable': CustomFieldUIEditableChoices.YES} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) def test_choice_set(self): @@ -116,6 +133,10 @@ class CustomFieldTestCase(TestCase, BaseFilterSetTests): params = {'choice_set_id': CustomFieldChoiceSet.objects.values_list('pk', flat=True)} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_description(self): + params = {'description': ['foobar1', 'foobar2']} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + class CustomFieldChoiceSetTestCase(TestCase, BaseFilterSetTests): queryset = CustomFieldChoiceSet.objects.all() @@ -124,12 +145,16 @@ class CustomFieldChoiceSetTestCase(TestCase, BaseFilterSetTests): @classmethod def setUpTestData(cls): choice_sets = ( - CustomFieldChoiceSet(name='Choice Set 1', extra_choices=['A', 'B', 'C']), - CustomFieldChoiceSet(name='Choice Set 2', extra_choices=['D', 'E', 'F']), - CustomFieldChoiceSet(name='Choice Set 3', extra_choices=['G', 'H', 'I']), + CustomFieldChoiceSet(name='Choice Set 1', extra_choices=['A', 'B', 'C'], description='foobar1'), + CustomFieldChoiceSet(name='Choice Set 2', extra_choices=['D', 'E', 'F'], description='foobar2'), + CustomFieldChoiceSet(name='Choice Set 3', extra_choices=['G', 'H', 'I'], description='foobar3'), ) CustomFieldChoiceSet.objects.bulk_create(choice_sets) + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Choice Set 1', 'Choice Set 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -138,6 +163,10 @@ class CustomFieldChoiceSetTestCase(TestCase, BaseFilterSetTests): params = {'choice': ['A', 'D']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_description(self): + params = {'description': ['foobar1', 'foobar2']} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + class WebhookTestCase(TestCase, BaseFilterSetTests): queryset = Webhook.objects.all() @@ -150,82 +179,196 @@ class WebhookTestCase(TestCase, BaseFilterSetTests): webhooks = ( Webhook( name='Webhook 1', - type_create=True, - type_update=False, - type_delete=False, - type_job_start=False, - type_job_end=False, payload_url='http://example.com/?1', - enabled=True, http_method='GET', ssl_verification=True, + description='foobar1' ), Webhook( name='Webhook 2', - type_create=False, - type_update=True, - type_delete=False, - type_job_start=False, - type_job_end=False, payload_url='http://example.com/?2', - enabled=True, http_method='POST', ssl_verification=True, + description='foobar2' ), Webhook( name='Webhook 3', - type_create=False, - type_update=False, - type_delete=True, - type_job_start=False, - type_job_end=False, payload_url='http://example.com/?3', - enabled=False, http_method='PATCH', ssl_verification=False, + description='foobar3' ), Webhook( name='Webhook 4', - type_create=False, - type_update=False, - type_delete=False, - type_job_start=True, - type_job_end=False, payload_url='http://example.com/?4', - enabled=False, http_method='PATCH', ssl_verification=False, ), Webhook( name='Webhook 5', - type_create=False, - type_update=False, - type_delete=False, - type_job_start=False, - type_job_end=True, payload_url='http://example.com/?5', - enabled=False, http_method='PATCH', ssl_verification=False, ), ) Webhook.objects.bulk_create(webhooks) - webhooks[0].content_types.add(content_types[0]) - webhooks[1].content_types.add(content_types[1]) - webhooks[2].content_types.add(content_types[2]) - webhooks[3].content_types.add(content_types[3]) - webhooks[4].content_types.add(content_types[4]) + + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) def test_name(self): params = {'name': ['Webhook 1', 'Webhook 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_description(self): + params = {'description': ['foobar1', 'foobar2']} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + + def test_http_method(self): + params = {'http_method': ['GET', 'POST']} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + + def test_ssl_verification(self): + params = {'ssl_verification': True} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + + +class EventRuleTestCase(TestCase, BaseFilterSetTests): + queryset = EventRule.objects.all() + filterset = EventRuleFilterSet + + @classmethod + def setUpTestData(cls): + content_types = ContentType.objects.filter( + model__in=['region', 'site', 'rack', 'location', 'device'] + ) + + webhooks = ( + Webhook( + name='Webhook 1', + payload_url='http://example.com/?1', + ), + Webhook( + name='Webhook 2', + payload_url='http://example.com/?2', + ), + Webhook( + name='Webhook 3', + payload_url='http://example.com/?3', + ), + ) + Webhook.objects.bulk_create(webhooks) + + scripts = ( + ScriptModule( + file_root=ManagedFileRootPathChoices.SCRIPTS, + file_path='/var/tmp/script1.py' + ), + ScriptModule( + file_root=ManagedFileRootPathChoices.SCRIPTS, + file_path='/var/tmp/script2.py' + ), + ) + ScriptModule.objects.bulk_create(scripts) + + event_rules = ( + EventRule( + 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, + action_type=EventRuleActionChoices.WEBHOOK, + description='foobar1' + ), + EventRule( + 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, + action_type=EventRuleActionChoices.WEBHOOK, + description='foobar2' + ), + EventRule( + 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, + action_type=EventRuleActionChoices.WEBHOOK, + description='foobar3' + ), + EventRule( + 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, + 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, + action_type=EventRuleActionChoices.SCRIPT, + ), + ) + EventRule.objects.bulk_create(event_rules) + event_rules[0].content_types.add(content_types[0]) + event_rules[1].content_types.add(content_types[1]) + event_rules[2].content_types.add(content_types[2]) + event_rules[3].content_types.add(content_types[3]) + event_rules[4].content_types.add(content_types[4]) + + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + + def test_name(self): + params = {'name': ['Event Rule 1', 'Event Rule 2']} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + + def test_description(self): + params = {'description': ['foobar1', 'foobar2']} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_content_types(self): params = {'content_types': 'dcim.region'} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) params = {'content_type_id': [ContentType.objects.get_for_model(Region).pk]} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_action_type(self): + params = {'action_type': [EventRuleActionChoices.WEBHOOK]} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3) + params = {'action_type': [EventRuleActionChoices.SCRIPT]} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + + def test_enabled(self): + params = {'enabled': True} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + 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) @@ -246,18 +389,6 @@ class WebhookTestCase(TestCase, BaseFilterSetTests): params = {'type_job_end': True} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) - def test_enabled(self): - params = {'enabled': True} - self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) - - def test_http_method(self): - params = {'http_method': ['GET', 'POST']} - self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) - - def test_ssl_verification(self): - params = {'ssl_verification': True} - self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) - class CustomLinkTestCase(TestCase, BaseFilterSetTests): queryset = CustomLink.objects.all() @@ -297,6 +428,10 @@ class CustomLinkTestCase(TestCase, BaseFilterSetTests): for i, custom_link in enumerate(custom_links): custom_link.content_types.set([content_types[i]]) + def test_q(self): + params = {'q': 'Custom Link 1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Custom Link 1', 'Custom Link 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -347,7 +482,8 @@ class SavedFilterTestCase(TestCase, BaseFilterSetTests): weight=100, enabled=True, shared=True, - parameters={'status': ['active']} + parameters={'status': ['active']}, + description='foobar1' ), SavedFilter( name='Saved Filter 2', @@ -356,7 +492,8 @@ class SavedFilterTestCase(TestCase, BaseFilterSetTests): weight=200, enabled=True, shared=True, - parameters={'status': ['planned']} + parameters={'status': ['planned']}, + description='foobar2' ), SavedFilter( name='Saved Filter 3', @@ -365,13 +502,18 @@ class SavedFilterTestCase(TestCase, BaseFilterSetTests): weight=300, enabled=False, shared=False, - parameters={'status': ['retired']} + parameters={'status': ['retired']}, + description='foobar3' ), ) SavedFilter.objects.bulk_create(saved_filters) for i, savedfilter in enumerate(saved_filters): savedfilter.content_types.set([content_types[i]]) + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Saved Filter 1', 'Saved Filter 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -380,6 +522,10 @@ class SavedFilterTestCase(TestCase, BaseFilterSetTests): params = {'slug': ['saved-filter-1', 'saved-filter-2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_description(self): + params = {'description': ['foobar1', 'foobar2']} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_content_types(self): params = {'content_types': 'dcim.site'} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) @@ -423,8 +569,6 @@ class BookmarkTestCase(TestCase, BaseFilterSetTests): @classmethod def setUpTestData(cls): - content_types = ContentType.objects.filter(model__in=['site', 'rack', 'device']) - users = ( User(username='User 1'), User(username='User 2'), @@ -505,6 +649,10 @@ class ExportTemplateTestCase(TestCase, BaseFilterSetTests): for i, et in enumerate(export_templates): et.content_types.set([content_types[i]]) + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Export Template 1', 'Export Template 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -578,6 +726,10 @@ class ImageAttachmentTestCase(TestCase, BaseFilterSetTests): ) ImageAttachment.objects.bulk_create(image_attachments) + def test_q(self): + params = {'q': 'Attachment 1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Image Attachment 1', 'Image Attachment 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -630,41 +782,45 @@ class JournalEntryTestCase(TestCase, ChangeLoggedFilterSetTests): assigned_object=sites[0], created_by=users[0], kind=JournalEntryKindChoices.KIND_INFO, - comments='New journal entry' + comments='foobar1' ), JournalEntry( assigned_object=sites[0], created_by=users[1], kind=JournalEntryKindChoices.KIND_SUCCESS, - comments='New journal entry' + comments='foobar2' ), JournalEntry( assigned_object=sites[1], created_by=users[2], kind=JournalEntryKindChoices.KIND_WARNING, - comments='New journal entry' + comments='foobar3' ), JournalEntry( assigned_object=racks[0], created_by=users[0], kind=JournalEntryKindChoices.KIND_INFO, - comments='New journal entry' + comments='foobar4' ), JournalEntry( assigned_object=racks[0], created_by=users[1], kind=JournalEntryKindChoices.KIND_SUCCESS, - comments='New journal entry' + comments='foobar5' ), JournalEntry( assigned_object=racks[1], created_by=users[2], kind=JournalEntryKindChoices.KIND_WARNING, - comments='New journal entry' + comments='foobar6' ), ) JournalEntry.objects.bulk_create(journal_entries) + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_created_by(self): users = User.objects.filter(username__in=['Alice', 'Bob']) params = {'created_by': [users[0].username, users[1].username]} @@ -800,9 +956,10 @@ class ConfigContextTestCase(TestCase, ChangeLoggedFilterSetTests): for i in range(0, 3): is_active = bool(i % 2) c = ConfigContext.objects.create( - name='Config Context {}'.format(i + 1), + name=f"Config Context {i + 1}", is_active=is_active, - data='{"foo": 123}' + data='{"foo": 123}', + description=f"foobar{i + 1}" ) c.regions.set([regions[i]]) c.site_groups.set([site_groups[i]]) @@ -818,6 +975,10 @@ class ConfigContextTestCase(TestCase, ChangeLoggedFilterSetTests): c.tenants.set([tenants[i]]) c.tags.set([tags[i]]) + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Config Context 1', 'Config Context 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -828,6 +989,10 @@ class ConfigContextTestCase(TestCase, ChangeLoggedFilterSetTests): params = {'is_active': False} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_description(self): + params = {'description': ['foobar1', 'foobar2']} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) + def test_region(self): regions = Region.objects.all()[:2] params = {'region_id': [regions[0].pk, regions[1].pk]} @@ -929,6 +1094,10 @@ class ConfigTemplateTestCase(TestCase, BaseFilterSetTests): ) ConfigTemplate.objects.bulk_create(config_templates) + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Config Template 1', 'Config Template 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -965,6 +1134,10 @@ class TagTestCase(TestCase, ChangeLoggedFilterSetTests): site.tags.set([tags[0]]) provider.tags.set([tags[1]]) + def test_q(self): + params = {'q': 'foobar1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) + def test_name(self): params = {'name': ['Tag 1', 'Tag 2']} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2) @@ -1076,6 +1249,10 @@ class ObjectChangeTestCase(TestCase, BaseFilterSetTests): ) ObjectChange.objects.bulk_create(object_changes) + def test_q(self): + params = {'q': 'Site 1'} + self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3) + def test_user(self): params = {'user_id': User.objects.filter(username__in=['user1', 'user2']).values_list('pk', flat=True)} self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4) diff --git a/netbox/extras/tests/test_views.py b/netbox/extras/tests/test_views.py index 296ed9f4d..d720560e4 100644 --- a/netbox/extras/tests/test_views.py +++ b/netbox/extras/tests/test_views.py @@ -1,4 +1,3 @@ -import json import urllib.parse import uuid @@ -11,7 +10,6 @@ from extras.choices import * from extras.models import * from utilities.testing import ViewTestCases, TestCase - User = get_user_model() @@ -50,15 +48,16 @@ class CustomFieldTestCase(ViewTestCases.PrimaryObjectViewTestCase): 'default': None, 'weight': 200, 'required': True, - 'ui_visibility': CustomFieldVisibilityChoices.VISIBILITY_READ_WRITE, + 'ui_visible': CustomFieldUIVisibleChoices.ALWAYS, + 'ui_editable': CustomFieldUIEditableChoices.YES, } cls.csv_data = ( - 'name,label,type,content_types,object_type,weight,search_weight,filter_logic,choice_set,validation_minimum,validation_maximum,validation_regex,ui_visibility', - 'field4,Field 4,text,dcim.site,,100,1000,exact,,,,[a-z]{3},read-write', - 'field5,Field 5,integer,dcim.site,,100,2000,exact,,1,100,,read-write', - 'field6,Field 6,select,dcim.site,,100,3000,exact,Choice Set 1,,,,read-write', - 'field7,Field 7,object,dcim.site,dcim.region,100,4000,exact,,,,,read-write', + 'name,label,type,content_types,object_type,weight,search_weight,filter_logic,choice_set,validation_minimum,validation_maximum,validation_regex,ui_visible,ui_editable', + 'field4,Field 4,text,dcim.site,,100,1000,exact,,,,[a-z]{3},always,yes', + 'field5,Field 5,integer,dcim.site,,100,2000,exact,,1,100,,always,yes', + 'field6,Field 6,select,dcim.site,,100,3000,exact,Choice Set 1,,,,always,yes', + 'field7,Field 7,object,dcim.site,dcim.region,100,4000,exact,,,,,always,yes', ) cls.csv_update_data = ( @@ -93,19 +92,24 @@ class CustomFieldChoiceSetTestCase(ViewTestCases.PrimaryObjectViewTestCase): name='Choice Set 3', extra_choices=(('C1', 'Choice 1'), ('C2', 'Choice 2'), ('C3', 'Choice 3')) ), + CustomFieldChoiceSet( + name='Choice Set 4', + extra_choices=(('D1', 'Choice 1'), ('D2', 'Choice 2'), ('D3', 'Choice 3')) + ), ) CustomFieldChoiceSet.objects.bulk_create(choice_sets) cls.form_data = { 'name': 'Choice Set X', - 'extra_choices': '\n'.join(['X1,Choice 1', 'X2,Choice 2', 'X3,Choice 3']) + 'extra_choices': '\n'.join(['X1:Choice 1', 'X2:Choice 2', 'X3:Choice 3']) } cls.csv_data = ( 'name,extra_choices', - 'Choice Set 4,"D1,D2,D3"', - 'Choice Set 5,"E1,E2,E3"', - 'Choice Set 6,"F1,F2,F3"', + 'Choice Set 5,"D1,D2,D3"', + 'Choice Set 6,"E1,E2,E3"', + 'Choice Set 7,"F1,F2,F3"', + 'Choice Set 8,"F1:L1,F2:L2,F3:L3"', ) cls.csv_update_data = ( @@ -113,12 +117,20 @@ class CustomFieldChoiceSetTestCase(ViewTestCases.PrimaryObjectViewTestCase): f'{choice_sets[0].pk},"A,B,C"', f'{choice_sets[1].pk},"A,B,C"', f'{choice_sets[2].pk},"A,B,C"', + f'{choice_sets[3].pk},"A:L1,B:L2,C:L3"', ) cls.bulk_edit_data = { 'description': 'New description', } + # This is here as extra_choices field splits on colon, but is returned + # from DB as comma separated. + def assertInstanceEqual(self, instance, data, exclude=None, api=False): + if 'extra_choices' in data: + data['extra_choices'] = data['extra_choices'].replace(':', ',') + return super().assertInstanceEqual(instance, data, exclude, api) + class CustomLinkTestCase(ViewTestCases.PrimaryObjectViewTestCase): model = CustomLink @@ -335,48 +347,94 @@ class WebhookTestCase(ViewTestCases.PrimaryObjectViewTestCase): @classmethod def setUpTestData(cls): - site_ct = ContentType.objects.get_for_model(Site) webhooks = ( - Webhook(name='Webhook 1', payload_url='http://example.com/?1', type_create=True, http_method='POST'), - Webhook(name='Webhook 2', payload_url='http://example.com/?2', type_create=True, http_method='POST'), - Webhook(name='Webhook 3', payload_url='http://example.com/?3', type_create=True, http_method='POST'), + Webhook(name='Webhook 1', payload_url='http://example.com/?1', http_method='POST'), + Webhook(name='Webhook 2', payload_url='http://example.com/?2', http_method='POST'), + Webhook(name='Webhook 3', payload_url='http://example.com/?3', http_method='POST'), ) for webhook in webhooks: webhook.save() - webhook.content_types.add(site_ct) cls.form_data = { 'name': 'Webhook X', + 'payload_url': 'http://example.com/?x', + 'http_method': 'GET', + 'http_content_type': 'application/foo', + 'description': 'My webhook', + } + + cls.csv_data = ( + "name,payload_url,http_method,http_content_type,description", + "Webhook 4,http://example.com/?4,GET,application/json,Foo", + "Webhook 5,http://example.com/?5,GET,application/json,Bar", + "Webhook 6,http://example.com/?6,GET,application/json,Baz", + ) + + cls.csv_update_data = ( + "id,name,description", + f"{webhooks[0].pk},Webhook 7,Foo", + f"{webhooks[1].pk},Webhook 8,Bar", + f"{webhooks[2].pk},Webhook 9,Baz", + ) + + cls.bulk_edit_data = { + 'http_method': 'GET', + } + + +class EventRulesTestCase(ViewTestCases.PrimaryObjectViewTestCase): + model = EventRule + + @classmethod + def setUpTestData(cls): + + webhooks = ( + Webhook(name='Webhook 1', payload_url='http://example.com/?1', http_method='POST'), + Webhook(name='Webhook 2', payload_url='http://example.com/?2', http_method='POST'), + Webhook(name='Webhook 3', payload_url='http://example.com/?3', http_method='POST'), + ) + for webhook in webhooks: + webhook.save() + + site_ct = ContentType.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]), + ) + for event in event_rules: + event.save() + event.content_types.add(site_ct) + + webhook_ct = ContentType.objects.get_for_model(Webhook) + cls.form_data = { + 'name': 'Event X', 'content_types': [site_ct.pk], 'type_create': False, 'type_update': True, 'type_delete': True, - 'payload_url': 'http://example.com/?x', - 'http_method': 'GET', - 'http_content_type': 'application/foo', 'conditions': None, + 'action_type': 'webhook', + 'action_object_type': webhook_ct.pk, + 'action_object_id': webhooks[0].pk, + 'action_choice': webhooks[0], + 'description': 'New description', } cls.csv_data = ( - "name,content_types,type_create,payload_url,http_method,http_content_type", - "Webhook 4,dcim.site,True,http://example.com/?4,GET,application/json", - "Webhook 5,dcim.site,True,http://example.com/?5,GET,application/json", - "Webhook 6,dcim.site,True,http://example.com/?6,GET,application/json", + "name,content_types,type_create,action_type,action_object", + "Webhook 4,dcim.site,True,webhook,Webhook 1", ) cls.csv_update_data = ( "id,name", - f"{webhooks[0].pk},Webhook 7", - f"{webhooks[1].pk},Webhook 8", - f"{webhooks[2].pk},Webhook 9", + f"{event_rules[0].pk},Event 7", + f"{event_rules[1].pk},Event 8", + f"{event_rules[2].pk},Event 9", ) cls.bulk_edit_data = { - 'enabled': False, - 'type_create': False, 'type_update': True, - 'type_delete': True, - 'http_method': 'GET', } @@ -457,7 +515,7 @@ class ConfigContextTestCase( 'platforms': [], 'tenant_groups': [], 'tenants': [], - 'device_types': [devicetype.id,], + 'device_types': [devicetype.id], 'tags': [], 'data': '{"foo": 123}', } diff --git a/netbox/extras/urls.py b/netbox/extras/urls.py index fd95186e4..0a1786f1f 100644 --- a/netbox/extras/urls.py +++ b/netbox/extras/urls.py @@ -61,6 +61,14 @@ urlpatterns = [ path('webhooks/delete/', views.WebhookBulkDeleteView.as_view(), name='webhook_bulk_delete'), path('webhooks/