bulk import

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,197 @@
+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 to get test expression from"),
+    )
+    data_file = CSVModelChoiceField(
+        queryset=DataFile.objects.all(),
+        required=False,
+        to_field_name="path",
+        help_text=_("File to get test expression from"),
+    )
+
+    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):
+        super().__init__(*args, headers=headers, **kwargs)
+        self.base_fields["global"] = self.base_fields.pop("_global")
+
+
+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,
+    )
+
+    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["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]

validity/urls.py

@@ -8,21 +8,25 @@
     path("selectors/", views.ComplianceSelectorListView.as_view(), name="complianceselector_list"),
     path("selectors/add/", views.ComplianceSelectorEditView.as_view(), name="complianceselector_add"),
     path("selectors/delete/", views.ComplianceSelectorBulkDeleteView.as_view(), name="complianceselector_bulk_delete"),
+    path("selectors/import/", views.ComplianceSelectorBulkImportView.as_view(), name="complianceselector_import"),
     path("selectors/<int:pk>/", include(get_model_urls("validity", "complianceselector"))),
     path("tests/", views.ComplianceTestListView.as_view(), name="compliancetest_list"),
     path("tests/run/", views.RunTestsView.as_view(), name="compliancetest_run"),
     path("tests/add/", views.ComplianceTestEditView.as_view(), name="compliancetest_add"),
     path("tests/delete/", views.ComplianceTestBulkDeleteView.as_view(), name="compliancetest_bulk_delete"),
+    path("tests/import/", views.ComplianceTestBulkImportView.as_view(), name="compliancetest_import"),
     path("tests/<int:pk>/", include(get_model_urls("validity", "compliancetest"))),
     path("test-results/", views.ComplianceResultListView.as_view(), name="compliancetestresult_list"),
     path("test-results/<int:pk>/", include(get_model_urls("validity", "compliancetestresult"))),
     path("serializers/", views.SerializerListView.as_view(), name="serializer_list"),
     path("serializers/add/", views.SerializerEditView.as_view(), name="serializer_add"),
     path("serializers/delete/", views.SerializerBulkDeleteView.as_view(), name="serializer_bulk_delete"),
+    path("serializers/import/", views.SerializerBulkImportView.as_view(), name="serializer_import"),
     path("serializers/<int:pk>/", include(get_model_urls("validity", "serializer"))),
     path("namesets/", views.NameSetListView.as_view(), name="nameset_list"),
     path("namesets/add/", views.NameSetEditView.as_view(), name="nameset_add"),
     path("namesets/delete/", views.NameSetBulkDeleteView.as_view(), name="nameset_bulk_delete"),
+    path("namesets/import/", views.NameSetBulkImportView.as_view(), name="nameset_import"),
     path("namesets/<int:pk>/", include(get_model_urls("validity", "nameset"))),
     path("reports/", views.ComplianceReportListView.as_view(), name="compliancereport_list"),
     path("reports/<int:pk>/", include(get_model_urls("validity", "compliancereport"))),
@@ -31,10 +35,12 @@
     path("pollers/", views.PollerListView.as_view(), name="poller_list"),
     path("pollers/add/", views.PollerEditView.as_view(), name="poller_add"),
     path("pollers/delete/", views.PollerBulkDeleteView.as_view(), name="poller_bulk_delete"),
+    path("pollers/import/", views.PollerBulkImportView.as_view(), name="poller_import"),
     path("pollers/<int:pk>/", include(get_model_urls("validity", "poller"))),
     path("commands/", views.CommandListView.as_view(), name="command_list"),
     path("commands/add/", views.CommandEditView.as_view(), name="command_add"),
     path("commands/delete/", views.CommandBulkDeleteView.as_view(), name="command_bulk_delete"),
+    path("commands/import/", views.CommandBulkImportView.as_view(), name="command_import"),
     path("commands/<int:pk>/", include(get_model_urls("validity", "command"))),
     path("scripts/results/<int:pk>/", views.ScriptResultView.as_view(), name="script_result"),
 ]

validity/views/__init__.py

@@ -1,3 +1,11 @@
+from .bulk_import import (
+    CommandBulkImportView,
+    ComplianceSelectorBulkImportView,
+    ComplianceTestBulkImportView,
+    NameSetBulkImportView,
+    PollerBulkImportView,
+    SerializerBulkImportView,
+)
 from .command import CommandBulkDeleteView, CommandDeleteView, CommandEditView, CommandListView, CommandView
 from .data_source import DataSourceBoundDevicesView
 from .device import DeviceSerializedStateView, TestResultView