Bulk import (#105)

* bulk import

* tests

Author: amyasnikov

validity/api/helpers.py

@@ -82,18 +82,23 @@ class SubformValidationMixin:
     Serializer Mixin. Validates JSON field according to a subform
     """
 
+    def _validate(self, attrs):
+        instance = self.instance or self.Meta.model()
+        for field, field_value in attrs.items():
+            if not isinstance(instance._meta.get_field(field), ManyToManyField):
+                setattr(instance, field, field_value)
+        if not instance.subform_type:
+            return
+        subform = instance.subform_cls(instance.subform_json)
+        if not subform.is_valid():
+            errors = [
+                ": ".join((field, err[0])) if field != "__all__" else err for field, err in subform.errors.items()
+            ]
+            raise ValidationError({instance.subform_json_field: errors})
+
     def validate(self, attrs):
         if isinstance(attrs, dict):
-            instance = self.instance or self.Meta.model()
-            for field, field_value in attrs.items():
-                if not isinstance(instance._meta.get_field(field), ManyToManyField):
-                    setattr(instance, field, field_value)
-            subform = instance.subform_cls(instance.subform_json)
-            if not subform.is_valid():
-                errors = [
-                    ": ".join((field, err[0])) if field != "__all__" else err for field, err in subform.errors.items()
-                ]
-                raise ValidationError({instance.subform_json_field: errors})
+            self._validate(attrs)
         return attrs
 
 

validity/forms/__init__.py

@@ -1,3 +1,11 @@
+from .bulk_import import (
+    CommandImportForm,
+    ComplianceSelectorImportForm,
+    ComplianceTestImportForm,
+    NameSetImportForm,
+    PollerImportForm,
+    SerializerImportForm,
+)
 from .filterset import (
     CommandFilterForm,
     ComplianceReportFilerForm,

validity/forms/bulk_import.py

@@ -0,0 +1,209 @@
+from core.models import DataFile, DataSource
+from dcim.choices import DeviceStatusChoices
+from dcim.models import DeviceType, Location, Manufacturer, Platform, Site
+from django.forms import Form
+from django.utils.translation import gettext_lazy as _
+from extras.models import Tag
+from netbox.forms import NetBoxModelImportForm
+from tenancy.models import Tenant
+from utilities.forms.fields import CSVChoiceField, CSVModelChoiceField, CSVModelMultipleChoiceField, JSONField
+
+from validity import choices, models
+from validity.api.helpers import SubformValidationMixin
+from .mixins import PollerCleanMixin
+
+
+class SubFormMixin(SubformValidationMixin):
+    def clean(self):
+        validated_data = {k: v for k, v in self.cleaned_data.items() if not k.startswith("_")}
+        self.validate(validated_data)
+        return self.cleaned_data
+
+
+class DataSourceMixin(Form):
+    data_source = CSVModelChoiceField(
+        queryset=DataSource.objects.all(),
+        required=False,
+        to_field_name="name",
+        help_text=_("Data Source"),
+    )
+    data_file = CSVModelChoiceField(
+        queryset=DataFile.objects.all(),
+        required=False,
+        to_field_name="path",
+        help_text=_("File from Data Source"),
+    )
+
+    def clean_data_source(self):
+        """
+        Filters data file by known data source
+        data_source MUST go before data_file in Meta.fields
+        """
+        data_source = self.cleaned_data["data_source"]
+        if data_source is not None:
+            datafile_field = self.fields["data_file"]
+            datafile_field.queryset = datafile_field.queryset.filter(source=data_source)
+        return data_source
+
+
+class ComplianceTestImportForm(DataSourceMixin, NetBoxModelImportForm):
+    severity = CSVChoiceField(choices=choices.SeverityChoices.choices, help_text=_("Test Severity"))
+    selectors = CSVModelMultipleChoiceField(
+        queryset=models.ComplianceSelector.objects.all(),
+        to_field_name="name",
+        help_text=_("Compliance Selector names separated by commas, encased with double quotes"),
+    )
+
+    class Meta:
+        model = models.ComplianceTest
+        fields = ("name", "severity", "description", "selectors", "expression", "data_source", "data_file", "tags")
+
+
+class NameSetImportForm(DataSourceMixin, NetBoxModelImportForm):
+    tests = CSVModelMultipleChoiceField(
+        queryset=models.ComplianceTest.objects.all(),
+        to_field_name="name",
+        help_text=_("Compliance Test names separated by commas, encased with double quotes"),
+        required=False,
+    )
+
+    class Meta:
+        model = models.NameSet
+        fields = ("name", "description", "_global", "tests", "definitions", "data_source", "data_file")
+
+    def __init__(self, *args, headers=None, **kwargs):
+        base_fields = {"global": self.base_fields["_global"]} | self.base_fields
+        base_fields.pop("_global")
+        self.base_fields = base_fields
+        super().__init__(*args, headers=headers, **kwargs)
+
+    def save(self, commit=True) -> choices.Any:
+        if (_global := self.cleaned_data.get("global")) is not None:
+            self.instance._global = _global
+        return super().save(commit)
+
+
+class SerializerImportForm(SubFormMixin, DataSourceMixin, NetBoxModelImportForm):
+    extraction_method = CSVChoiceField(
+        choices=choices.ExtractionMethodChoices.choices, help_text=_("Extraction Method")
+    )
+    parameters = JSONField(
+        help_text=_(
+            "JSON-encoded Serializer parameters depending on Extraction Method value. "
+            "See REST API to check for specific keys/values"
+        )
+    )
+
+    class Meta:
+        model = models.Serializer
+        fields = ("name", "extraction_method", "template", "parameters", "data_source", "data_file")
+
+
+class ComplianceSelectorImportForm(NetBoxModelImportForm):
+    filter_operation = CSVChoiceField(
+        choices=choices.BoolOperationChoices.choices, help_text=_("Filter Join Operation")
+    )
+    tag_filter = CSVModelMultipleChoiceField(
+        queryset=Tag.objects.all(),
+        to_field_name="slug",
+        help_text=_("Tag slugs separated by commas, encased with double quotes"),
+        required=False,
+    )
+    manufacturer_filter = CSVModelMultipleChoiceField(
+        queryset=Manufacturer.objects.all(),
+        to_field_name="slug",
+        help_text=_("Manufacturer slugs separated by commas, encased with double quotes"),
+        required=False,
+    )
+    type_filter = CSVModelMultipleChoiceField(
+        queryset=DeviceType.objects.all(),
+        to_field_name="slug",
+        help_text=_("Device Type slugs separated by commas, encased with double quotes"),
+        required=False,
+    )
+    platform_filter = CSVModelMultipleChoiceField(
+        queryset=Platform.objects.all(),
+        to_field_name="slug",
+        help_text=_("Platform slugs separated by commas, encased with double quotes"),
+        required=False,
+    )
+    status_filter = CSVChoiceField(choices=DeviceStatusChoices, help_text=_("Device Status Filter"), required=False)
+    location_filter = CSVModelMultipleChoiceField(
+        queryset=Location.objects.all(),
+        to_field_name="slug",
+        help_text=_("Location slugs separated by commas, encased with double quotes"),
+        required=False,
+    )
+    site_filter = CSVModelMultipleChoiceField(
+        queryset=Site.objects.all(),
+        to_field_name="slug",
+        help_text=_("Site slugs separated by commas, encased with double quotes"),
+        required=False,
+    )
+    tenant_filter = CSVModelMultipleChoiceField(
+        queryset=Tenant.objects.all(),
+        to_field_name="slug",
+        help_text=("Tenant slugs separated by commas, encased with double quotes"),
+        required=False,
+    )
+    dynamic_pairs = CSVChoiceField(
+        choices=choices.DynamicPairsChoices.choices, required=False, help_text=_("Dynamic Pairs")
+    )
+
+    class Meta:
+        model = models.ComplianceSelector
+        fields = (
+            "name",
+            "filter_operation",
+            "name_filter",
+            "tag_filter",
+            "manufacturer_filter",
+            "type_filter",
+            "platform_filter",
+            "status_filter",
+            "location_filter",
+            "site_filter",
+            "tenant_filter",
+            "dynamic_pairs",
+            "dp_tag_prefix",
+        )
+
+
+class CommandImportForm(SubFormMixin, NetBoxModelImportForm):
+    serializer = CSVModelChoiceField(
+        queryset=models.Serializer.objects.all(), to_field_name="name", help_text=_("Serializer"), required=False
+    )
+    type = CSVChoiceField(choices=choices.CommandTypeChoices.choices, help_text=_("Command Type"))
+    parameters = JSONField(
+        help_text=_(
+            "JSON-encoded Command parameters depending on Type value. See REST API to check for specific keys/values"
+        )
+    )
+
+    class Meta:
+        model = models.Command
+        fields = ("name", "label", "retrieves_config", "serializer", "type", "parameters")
+
+
+class PollerImportForm(PollerCleanMixin, NetBoxModelImportForm):
+    connection_type = CSVChoiceField(choices=choices.ConnectionTypeChoices.choices, help_text=_("Connection Type"))
+    commands = CSVModelMultipleChoiceField(
+        queryset=models.Command.objects.all(),
+        to_field_name="label",
+        help_text=_("Command labels separated by commas, encased with double quotes"),
+    )
+    public_credentials = JSONField(help_text=_("Public Credentials"), required=False)
+    private_credentials = JSONField(
+        help_text=_(
+            "Private Credentials. ATTENTION: encryption depends on Django's SECRET_KEY var, "
+            "values from another NetBox may not be decrypted properly"
+        ),
+        required=False,
+    )
+
+    def full_clean(self) -> None:
+        return super().full_clean()
+
+    class Meta:
+        model = models.Poller
+        fields = ("name", "connection_type", "commands", "public_credentials", "private_credentials")

validity/forms/general.py

@@ -7,15 +7,15 @@
 from extras.models import Tag
 from netbox.forms import NetBoxModelForm
 from tenancy.models import Tenant
-from utilities.forms import add_blank_choice, get_field_value
+from utilities.forms import add_blank_choice
 from utilities.forms.fields import DynamicModelChoiceField, DynamicModelMultipleChoiceField
 from utilities.forms.widgets import HTMXSelect
 
 from validity import models
 from validity.choices import ConnectionTypeChoices, ExplanationVerbosityChoices
 from validity.netbox_changes import FieldSet
 from .fields import DynamicModelChoicePropertyField, DynamicModelMultipleChoicePropertyField
-from .mixins import SubformMixin
+from .mixins import PollerCleanMixin, SubformMixin
 from .widgets import PrettyJSONWidget
 
 
@@ -133,7 +133,7 @@ class Meta:
         fields = ("name", "description", "_global", "tests", "definitions", "data_source", "data_file", "tags")
 
 
-class PollerForm(NetBoxModelForm):
+class PollerForm(PollerCleanMixin, NetBoxModelForm):
     connection_type = ChoiceField(
         choices=add_blank_choice(ConnectionTypeChoices.choices), widget=Select(attrs={"id": "connection_type_select"})
     )
@@ -147,11 +147,6 @@ class Meta:
             "private_credentials": PrettyJSONWidget(),
         }
 
-    def clean(self):
-        connection_type = self.cleaned_data.get("connection_type") or get_field_value(self, "connection_type")
-        models.Poller.validate_commands(connection_type, self.cleaned_data["commands"])
-        return super().clean()
-
 
 class CommandForm(SubformMixin, NetBoxModelForm):
     serializer = DynamicModelChoiceField(queryset=models.Serializer.objects.all(), required=False)

validity/forms/mixins.py

@@ -4,6 +4,7 @@
 from utilities.forms import get_field_value
 from utilities.forms.fields import DynamicModelMultipleChoiceField
 
+from validity.models import Poller
 from validity.netbox_changes import FieldSet
 
 
@@ -24,6 +25,13 @@ def __init__(self, *args, exclude: Sequence[str] = (), **kwargs) -> None:
             self.fields.pop(field, None)
 
 
+class PollerCleanMixin:
+    def clean(self):
+        connection_type = self.cleaned_data.get("connection_type") or get_field_value(self, "connection_type")
+        Poller.validate_commands(connection_type, self.cleaned_data.get("commands", []))
+        return super().clean()
+
+
 class SubformMixin:
     main_fieldsets: Sequence[FieldSet | Literal["__subform__"]]
 

validity/navigation.py

@@ -6,7 +6,17 @@ def model_add_button(entity):
         link=f"plugins:validity:{entity}_add",
         title="Add",
         icon_class="mdi mdi-plus-thick",
-        color=ButtonColorChoices.GREEN,
+        color=ButtonColorChoices.TEAL,
+        permissions=[f"validity.add_{entity}"],
+    )
+
+
+def model_import_button(entity):
+    return plugins.PluginMenuButton(
+        link=f"plugins:validity:{entity}_import",
+        title="Import",
+        icon_class="mdi mdi-upload",
+        color=ButtonColorChoices.CYAN,
         permissions=[f"validity.add_{entity}"],
     )
 
@@ -25,21 +35,21 @@ def model_menu_item(entity, title, buttons=()):
     link="plugins:validity:compliancetest_run",
     title="Run",
     icon_class="mdi mdi-rocket-launch",
-    color=ButtonColorChoices.CYAN,
+    color=ButtonColorChoices.PINK,
 )
 
 validity_menu_items = (
-    model_menu_item("complianceselector", "Selectors", [model_add_button]),
-    model_menu_item("compliancetest", "Tests", [run_tests_button, model_add_button]),
+    model_menu_item("complianceselector", "Selectors", [model_add_button, model_import_button]),
+    model_menu_item("compliancetest", "Tests", [run_tests_button, model_add_button, model_import_button]),
     model_menu_item("compliancetestresult", "Test Results"),
     model_menu_item("compliancereport", "Reports"),
-    model_menu_item("serializer", "Serializers", [model_add_button]),
-    model_menu_item("nameset", "Name Sets", [model_add_button]),
+    model_menu_item("serializer", "Serializers", [model_add_button, model_import_button]),
+    model_menu_item("nameset", "Name Sets", [model_add_button, model_import_button]),
 )
 
 polling_menu_items = (
-    model_menu_item("command", "Commands", [model_add_button]),
-    model_menu_item("poller", "Pollers", [model_add_button]),
+    model_menu_item("command", "Commands", [model_add_button, model_import_button]),
+    model_menu_item("poller", "Pollers", [model_add_button, model_import_button]),
 )
 
 menu = plugins.PluginMenu(

validity/template_content.py

@@ -54,7 +54,7 @@ class ComplianceTestExtension(PluginTemplateExtension):
     def list_buttons(self):
         run_tests_url = reverse("plugins:validity:compliancetest_run")
         icon = '<i class="mdi mdi-rocket-launch"></i>'
-        return f'<a class="btn btn-cyan" href="{run_tests_url}">{icon} Run Tests</a>'
+        return f'<a class="btn btn-pink" href="{run_tests_url}">{icon} Run Tests</a>'
 
 
 template_extensions = [DataSourceTenantExtension, PollingInfoExtension, ComplianceTestExtension]