Fix netbox-community#12731: Add custom model validation from forms an…

…d serializers

This allows for validation of any m2m relation like tags.
The validator receives the cleaned data from the form/serializer.
The model instance is not consistently available in the model
forms/serializers so is ommitted from both.

Note: the available fields can vary per form/serializer for a model.
You should always check if a field is defined in the cleaned data.

netbox/extras/signals.py

@@ -9,7 +9,7 @@
 from extras.validators import CustomValidator
 from netbox.config import get_config
 from netbox.context import current_request, webhooks_queue
-from netbox.signals import post_clean
+from netbox.signals import post_clean, post_form_clean, post_serializer_clean
 from utilities.exceptions import AbortRequest
 from .choices import ObjectChangeActionChoices
 from .models import ConfigRevision, CustomField, ObjectChange, TaggedItem
@@ -178,12 +178,17 @@ def handle_cf_deleted(instance, **kwargs):
 # Custom validation
 #
 
-@receiver(post_clean)
-def run_custom_validators(sender, instance, **kwargs):
+@receiver([post_clean, post_form_clean, post_serializer_clean])
+def run_custom_validators(signal, sender, instance=None, data=None, **kwargs):
     config = get_config()
     model_name = f'{sender._meta.app_label}.{sender._meta.model_name}'
     validators = config.CUSTOM_VALIDATORS.get(model_name, [])
 
+    if signal is post_clean:
+        assert instance is not None
+    else:
+        assert data is not None
+
     for validator in validators:
 
         # Loading a validator class by dotted path
@@ -195,7 +200,12 @@ def run_custom_validators(sender, instance, **kwargs):
         elif type(validator) is dict:
             validator = CustomValidator(validator)
 
-        validator(instance)
+        if signal is post_form_clean:
+            validator.validate_form_data(data)
+        elif signal is post_serializer_clean:
+            validator.validate_serializer_data(data)
+        else:
+            validator(instance)
 
 
 #

netbox/extras/tests/test_customvalidator.py

@@ -4,6 +4,9 @@
 
 from ipam.models import ASN, RIR
 from dcim.models import Site
+from dcim.api.serializers import SiteSerializer
+from dcim.forms.model_forms import SiteForm
+from extras.models import Tag
 from extras.validators import CustomValidator
 
 
@@ -14,6 +17,33 @@ def validate(self, instance):
             self.fail("Name must be foo!")
 
 
+class FooTagValidation:
+
+    def validate_foo_tag(self, data):
+        if data['name'] != 'foo':
+            for tag in data['tags']:
+                if tag.name == 'FOO':
+                    self.fail('FOO tag is reserved for site foo', 'tags')
+
+
+class MyDataValidator(FooTagValidation, CustomValidator):
+
+    def validate_data(self, data):
+        self.validate_foo_tag(data)
+
+
+class MyFormValidator(FooTagValidation, CustomValidator):
+
+    def validate_form_data(self, data):
+        self.validate_foo_tag(data)
+
+
+class MySerializerValidator(FooTagValidation, CustomValidator):
+
+    def validate_serializer_data(self, data):
+        self.validate_foo_tag(data)
+
+
 min_validator = CustomValidator({
     'asn': {
         'min': 65000
@@ -64,12 +94,31 @@ def validate(self, instance):
 
 custom_validator = MyValidator()
 
+custom_data_validator = MyDataValidator()
+
+custom_form_validator = MyFormValidator()
+
+custom_serializer_validator = MySerializerValidator()
+
 
 class CustomValidatorTest(TestCase):
 
     @classmethod
     def setUpTestData(cls):
         RIR.objects.create(name='RIR 1', slug='rir-1')
+        tag = Tag.objects.create(name='FOO', slug='foo')
+        cls.valid_data = {
+            'name': 'foo',
+            'slug': 'foo',
+            'status': 'active',
+            'tags': [tag.pk],
+        }
+        cls.invalid_data = {
+            'name': 'abc',
+            'slug': 'abc',
+            'status': 'active',
+            'tags': [tag.pk],
+        }
 
     @override_settings(CUSTOM_VALIDATORS={'ipam.asn': [min_validator]})
     def test_configuration(self):
@@ -125,6 +174,54 @@ def test_custom_invalid(self):
     def test_custom_valid(self):
         Site(name='foo', slug='foo').clean()
 
+    @override_settings(CUSTOM_VALIDATORS={'dcim.site': [custom_data_validator]})
+    def test_custom_data_invalid(self):
+        form = SiteForm(self.invalid_data)
+        self.assertFalse(form.is_valid())
+        self.assertIn('tags', form.errors)
+        serializer = SiteSerializer(data=self.invalid_data)
+        self.assertFalse(serializer.is_valid())
+        self.assertIn('tags', serializer.errors)
+
+    @override_settings(CUSTOM_VALIDATORS={'dcim.site': [custom_data_validator]})
+    def test_custom_data_valid(self):
+        form = SiteForm(self.valid_data)
+        self.assertTrue(form.is_valid(), form.errors.as_data())
+        serializer = SiteSerializer(data=self.valid_data)
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+
+    @override_settings(CUSTOM_VALIDATORS={'dcim.site': [custom_form_validator]})
+    def test_custom_form_invalid(self):
+        form = SiteForm(self.invalid_data)
+        self.assertFalse(form.is_valid())
+        self.assertIn('tags', form.errors)
+        # Form validator does not affect serializer validation.
+        serializer = SiteSerializer(data=self.invalid_data)
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+
+    @override_settings(CUSTOM_VALIDATORS={'dcim.site': [custom_form_validator]})
+    def test_custom_form_valid(self):
+        form = SiteForm(self.valid_data)
+        self.assertTrue(form.is_valid(), form.errors.as_data())
+        serializer = SiteSerializer(data=self.valid_data)
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+
+    @override_settings(CUSTOM_VALIDATORS={'dcim.site': [custom_serializer_validator]})
+    def test_custom_serializer_invalid(self):
+        # Serializer validator does not affect form validation.
+        form = SiteForm(self.invalid_data)
+        self.assertTrue(form.is_valid(), form.errors.as_data())
+        serializer = SiteSerializer(data=self.invalid_data)
+        self.assertFalse(serializer.is_valid())
+        self.assertIn('tags', serializer.errors)
+
+    @override_settings(CUSTOM_VALIDATORS={'dcim.site': [custom_serializer_validator]})
+    def test_custom_serializer_valid(self):
+        form = SiteForm(self.valid_data)
+        self.assertTrue(form.is_valid(), form.errors.as_data())
+        serializer = SiteSerializer(data=self.valid_data)
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+
 
 class CustomValidatorConfigTest(TestCase):
 

netbox/extras/validators.py

@@ -98,6 +98,27 @@ def validate(self, instance):
         """
         return
 
+    def validate_data(self, data):
+        """
+        Custom validation method for model forms and model serializers, to be overridden by the user.
+        Validation failures should raise a ValidationError exception.
+        """
+        return
+
+    def validate_form_data(self, data):
+        """
+        Custom validation method for model forms, to be overridden by the user.
+        Validation failures should raise a ValidationError exception.
+        """
+        return self.validate_data(data)
+
+    def validate_serializer_data(self, data):
+        """
+        Custom validation method for model serializers, to be overridden by the user.
+        Validation failures should raise a ValidationError exception.
+        """
+        return self.validate_data(data)
+
     def fail(self, message, field=None):
         """
         Raise a ValidationError exception. Associate the provided message with a form/serializer field if specified.

netbox/netbox/api/serializers/base.py

@@ -3,6 +3,8 @@
 from drf_spectacular.utils import extend_schema_field
 from drf_spectacular.types import OpenApiTypes
 
+from netbox.signals import post_serializer_clean
+
 __all__ = (
     'BaseModelSerializer',
     'ValidatedModelSerializer',
@@ -43,4 +45,7 @@ def validate(self, data):
                 setattr(instance, k, v)
         instance.full_clean()
 
+        # Send the post_serializer_clean signal
+        post_serializer_clean.send(sender=self.Meta.model, data=data)
+
         return data

netbox/netbox/forms/base.py

@@ -6,6 +6,7 @@
 from extras.choices import CustomFieldFilterLogicChoices, CustomFieldTypeChoices, CustomFieldVisibilityChoices
 from extras.forms.mixins import CustomFieldsMixin, SavedFiltersMixin, TagsMixin
 from extras.models import CustomField, Tag
+from netbox.signals import post_form_clean
 from utilities.forms import CSVModelForm
 from utilities.forms.fields import CSVModelMultipleChoiceField, DynamicModelMultipleChoiceField
 from utilities.forms.mixins import BootstrapMixin, CheckLastUpdatedMixin
@@ -55,6 +56,9 @@ def clean(self):
             else:
                 self.instance.custom_field_data[key] = customfield.serialize(value)
 
+        # Send the post_form_clean signal
+        post_form_clean.send(sender=self._meta.model, data=self.cleaned_data)
+
         return super().clean()
 
 

netbox/netbox/signals.py

@@ -3,3 +3,9 @@
 
 # Signals that a model has completed its clean() method
 post_clean = Signal()
+
+# Signals that a model form has completed its clean() method
+post_form_clean = Signal()
+
+# Signals that a model serializer has completed its validate() method
+post_serializer_clean = Signal()