[Feature] Survey validator

Squest/settings.py

@@ -523,6 +523,7 @@ def backup_filename(databasename, servername, datetime, extension, content_type)
 # Plugins
 # -----------------------------------------
 FIELD_VALIDATOR_PATH = "plugins/field_validators"
+SURVEY_VALIDATOR_PATH = "plugins/survey_validators"
 
 # -----------------------------------------
 # SQL Debug

Squest/utils/plugin_controller.py

@@ -1,5 +1,9 @@
 import logging
+from importlib.machinery import SourceFileLoader
+import inspect
 import os
+from pydoc import locate
+import re
 
 from django.conf import settings
 
@@ -12,7 +16,7 @@
 class PluginController:
 
     @classmethod
-    def get_user_provisioned_validators(cls):
+    def get_user_provisioned_field_validators(cls):
         filepath = settings.FIELD_VALIDATOR_PATH
         file_list = os.listdir(filepath)
         for forbidden_word in ["__init__.py", "__pycache__"]:
@@ -23,25 +27,72 @@ def get_user_provisioned_validators(cls):
             returned_list.append(os.path.splitext(file_name)[0])
         returned_list.sort()
         return returned_list
+    @classmethod
+    def get_user_provisioned_survey_validators(cls):
+        def is_validator(obj):
+            """
+            Returns True if the object is a Script.
+            """
+            from service_catalog.forms import SurveyValidator
+            try:
+                return issubclass(obj, SurveyValidator) and obj != SurveyValidator
+            except TypeError:
+                return False
+
+        def python_name(full_path):
+            path, filename = os.path.split(full_path)
+            name = os.path.splitext(filename)[0]
+            if name == "__init__":
+                # File is a package
+                return os.path.basename(path)
+            else:
+                return name
+
+        def full_path_to_dotted_path(path):
+            path = re.sub(r"\.py", "", path)
+            path = re.sub(r"/", ".", path)
+            return path
+
+        scripts = list()
+        for filename in os.listdir(settings.SURVEY_VALIDATOR_PATH):
+            if filename in ["__init__.py", "__pycache__"]:
+                continue
+            full_path = os.path.join(settings.SURVEY_VALIDATOR_PATH, filename)
+            loader = SourceFileLoader(python_name(full_path), full_path)
+            module = loader.load_module()
+            for name, cls in inspect.getmembers(module, is_validator):
+                dotted_path = f"{full_path_to_dotted_path(full_path)}.{cls.__name__}"
+                scripts.append(dotted_path)
+        return scripts
 
     @classmethod
     def get_ui_field_validator_def(cls, validator_file):
-        return cls._load_validator_module(module_name=validator_file, definition_kind=VALIDATE_UI_DEFINITION_NAME)
+        return cls._load_field_validator_module(module_name=validator_file, definition_kind=VALIDATE_UI_DEFINITION_NAME)
 
     @classmethod
     def get_api_field_validator_def(cls, validator_file):
-        return cls._load_validator_module(module_name=validator_file, definition_kind=VALIDATE_API_DEFINITION_NAME)
+        return cls._load_field_validator_module(module_name=validator_file,
+                                                definition_kind=VALIDATE_API_DEFINITION_NAME)
 
     @classmethod
-    def _load_validator_module(cls, module_name, definition_kind):
+    def get_survey_validator_def(cls, validator_file):
+        return locate(validator_file)
+
+    @classmethod
+    def _load_field_validator_module(cls, module_name, definition_kind):
+        filepath = settings.FIELD_VALIDATOR_PATH
+        return cls._load_validator_module(module_name, definition_kind, filepath)
+
+    @staticmethod
+    def _load_validator_module(module_name, definition_kind, filepath):
         """
         Dynamically load a validator definition from a python file
-        :param module_name: name of the python file that contains field validator definitions
-        :param definition_kind: UI or API
+        :param module_name: name of the python file that contains validator definitions
+        :param definition_kind: UI or API or Survey
         :return:
         """
         try:
-            mod = __import__(f"{settings.FIELD_VALIDATOR_PATH.replace('/','.')}.{module_name}",
+            mod = __import__(f"{filepath.replace('/', '.')}.{module_name}",
                              fromlist=[definition_kind])
             klass = getattr(mod, definition_kind)
             return klass

Squest/utils/squest_form.py

@@ -16,8 +16,9 @@ def __init__(self, *args, **kwargs):
     def is_valid(self):
         returned_value = super(SquestForm, self).is_valid()
         for field_name in self.errors.keys():
-            current_class = self.fields.get(field_name).widget.attrs.get('class', '')
-            self.fields.get(field_name).widget.attrs['class'] = f"{current_class} {self.error_css_class}"
+            if field_name != "__all__":
+                current_class = self.fields.get(field_name).widget.attrs.get('class', '')
+                self.fields.get(field_name).widget.attrs['class'] = f"{current_class} {self.error_css_class}"
         return returned_value
 
     def beautify(self):

docker-compose.yml

@@ -32,6 +32,7 @@ services:
       - django_media:/app/media
       - backup:/app/backup
       - ./plugins/field_validators:/app/plugins/field_validators
+      - ./plugins/survey_validators:/app/plugins/survey_validators
     depends_on:
       - db
       - rabbitmq

docs/configuration/squest_settings.md

@@ -341,6 +341,12 @@ Change the format of all date in Squest UI. Based on Python [strftime](https://s
 
 Path to form field validation modules.
 
+### SURVEY_VALIDATOR_PATH
+
+**Default:** `plugins/survey_validators`
+
+Path to form validation modules.
+
 ## Redis
 
 ### REDIS_CACHE_USER

docs/manual/advanced/survey_validators.md

@@ -0,0 +1,93 @@
+# Survey validators
+
+Survey validators are Python modules that can be added as plugin. It allows users to implement their own validation
+logic on a day1 or day2 operation.
+
+## Creating survey validator
+
+Create a Python file in **SURVEY_VALIDATOR_PATH** (default is `plugins/survey_validators`).
+Create Python class that inherit from SurveyValidator with a method `validate_survey`.
+
+```python
+# plugins/survey_validators/MySurveyValidator.py
+from service_catalog.forms import SurveyValidator
+
+class MyCustomValidatorFoo(SurveyValidator):
+    def validate_survey(self):
+        # Implement your own logic here
+        pass
+```
+
+## SurveyValidator attributes
+
+### survey
+
+This is a dict containing all the survey + request_comment. Keys are variable name.  
+type: dict
+
+```bash
+>>> print(self.survey)
+{
+  "request_comment": "commentary sent by user"
+  "ram_gb": 8,
+  "vcpu": 4
+}
+```
+
+### user
+
+User requesting operation.  
+type: django.contrib.auth.models.User
+
+### operation
+
+Operation requested.   
+type: service_catalog.models.Operation
+
+### instance
+
+Instance targeted.  
+type: service_catalog.models.Instance
+
+!!!note
+    For day 1 operation `self.instance` is a fake Instance object that contain only **name** and **quota_scope**
+    The real Instance object is created after validation.
+
+## SurveyValidator method
+
+### validate_survey(self)
+
+Redefine it to implement your own logic.
+
+### fail(self, message, field="\_\_all\_\_")
+
+Raise an exception and display message on UI/API
+
+## Set validator to a form field
+
+In Squest, edit an Operation to set validators. Multiples validators can be added, validators are executed in alphabetical order by script name and class name.
+
+## Example
+
+This validator will always fail if:
+
+- ram and cpu are both equal 1
+- Numbers of stars on Squest is under 250
+
+```python
+import requests
+from service_catalog.forms import SurveyValidator
+
+class ValidatorForVM(SurveyValidator):
+    def validate_survey(self):
+        if self.survey.get("ram") == 1 and  self.survey.get("vcpu") == 1:
+            self.fail("Forbidden: you cannot use ram=1 and cpu=1")
+
+        url = 'https://api.github.com/repos/HewlettPackard/squest'
+        headers = {'Accept': 'application/vnd.github.preview'}
+        response = requests.get(url, headers=headers).json()
+        squest_stars_on_github = response.get('watchers_count', 0)
+
+        if squest_stars_on_github <= 250:
+            self.fail("Number of stars on Squest is below 250, come back when it's 251")
+```

docs/manual/service_catalog/survey.md

@@ -50,7 +50,7 @@ Full `instance` and `user `object definition can be retrieved through the [API d
 ## Validators
 
 Field validators are python modules that can be added as plugin to perform a custom check on a form field.
-See related [documentation here](../advanced/validators.md).
+See related [documentation here](../advanced/field_validators.md).
 
 
 ## Attribute definition

mkdocs.yml

@@ -54,7 +54,8 @@ nav:
     - Advanced:
         - manual/advanced/filters.md
         - manual/advanced/jinja.md
-        - manual/advanced/validators.md
+        - manual/advanced/field_validators.md
+        - manual/advanced/survey_validators.md
         - manual/advanced/ldap.md
     - Notifications: manual/notifications.md
   - Administration:

service_catalog/api/serializers/request_serializers.py

@@ -1,13 +1,14 @@
 from json import dumps, loads
 
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import ValidationError, PermissionDenied
 from rest_framework.relations import PrimaryKeyRelatedField
 from rest_framework.serializers import ModelSerializer, CharField
 
 from Squest.utils.squest_encoder import SquestEncoder
 from profiles.api.serializers.user_serializers import UserSerializerNested
 from profiles.models import Scope
 from service_catalog.api.serializers import DynamicSurveySerializer, InstanceReadSerializer
+from service_catalog.models import InstanceState, OperationType
 from service_catalog.models.instance import Instance
 from service_catalog.models.message import RequestMessage
 from service_catalog.models.request import Request
@@ -41,21 +42,32 @@ def __init__(self, *args, **kwargs):
     def validate(self, data):
         super(ServiceRequestSerializer, self).validate(data)
         quota_scope = data.get("quota_scope")
-        fill_in_survey = data.get("fill_in_survey")
+        fill_in_survey = data.get("fill_in_survey",{})
+        request_comment = data.get("request_comment")
         # validate the quota if set on one of the fill_in_survey
-        if fill_in_survey is not None:
-            for field_name, value in fill_in_survey.items():
-                # get the tower field
-                tower_field = self.operation.tower_survey_fields.get(variable=field_name)
-                if tower_field.attribute_definition is not None:
-                    # try to find the field in the quota linked to the scope
-                    quota_set_on_attribute = quota_scope.quotas.filter(attribute_definition=tower_field.attribute_definition)
-                    if quota_set_on_attribute.exists():
-                        quota_set_on_attribute = quota_set_on_attribute.first()
-                        if value > quota_set_on_attribute.available:
-                            raise ValidationError({"fill_in_survey":
-                                                       f"Quota limit reached on '{field_name}'. "
-                                                       f"Available: {quota_set_on_attribute.available}"})
+        for field_name, value in fill_in_survey.items():
+            # get the tower field
+            tower_field = self.operation.tower_survey_fields.get(variable=field_name)
+            if tower_field.attribute_definition is not None:
+                # try to find the field in the quota linked to the scope
+                quota_set_on_attribute = quota_scope.quotas.filter(
+                    attribute_definition=tower_field.attribute_definition)
+                if quota_set_on_attribute.exists():
+                    quota_set_on_attribute = quota_set_on_attribute.first()
+                    if value > quota_set_on_attribute.available:
+                        raise ValidationError({"fill_in_survey":
+                                                   f"Quota limit reached on '{field_name}'. "
+                                                   f"Available: {quota_set_on_attribute.available}"})
+        fill_in_survey.update({"request_comment": request_comment})
+        for validators in self.operation.get_validators():
+            # load dynamically the user provided validator
+            validators(
+                survey=fill_in_survey,
+                user=self.user,
+                operation=self.operation,
+                instance=Instance(quota_scope=quota_scope, name=data.get("squest_instance_name")),
+                form=None
+            )._validate()
         return data
 
     def save(self):
@@ -117,6 +129,36 @@ def save(self, **kwargs):
         send_mail_request_update(target_request=new_request, user_applied_state=new_request.user, message=message)
         return new_request
 
+    def validate(self, data):
+        super(OperationRequestSerializer, self).validate(data)
+
+        if self.operation.is_admin_operation and not self.user.has_perm("service_catalog.admin_request_on_instance"):
+            raise PermissionDenied
+        if not self.operation.is_admin_operation and not self.user.has_perm("service_catalog.request_on_instance"):
+            raise PermissionDenied
+        if self.squest_instance.state not in [InstanceState.AVAILABLE]:
+            raise PermissionDenied("Instance not available")
+        if self.operation.enabled is False:
+            raise PermissionDenied(f"Operation is not enabled.")
+        if self.operation.service.id != self.squest_instance.service.id:
+            raise PermissionDenied("Operation service and instance service doesn't match")
+        if self.operation.type not in [OperationType.UPDATE, OperationType.DELETE]:
+            raise PermissionDenied("Operation type UPDATE and DELETE only")
+        fill_in_survey = data.get("fill_in_survey")
+        request_comment = data.get("request_comment")
+        fill_in_survey.update({"request_comment": request_comment})
+
+        for validators in self.operation.get_validators():
+            # load dynamically the user provided validator
+            validators(
+                survey=fill_in_survey,
+                user=self.user,
+                operation=self.operation,
+                instance=self.squest_instance,
+                form=None
+            )._validate()
+        return data
+
 
 class RequestSerializer(ModelSerializer):
     class Meta:

service_catalog/api/views/request_api_views.py

@@ -18,11 +18,12 @@ class RequestList(SquestListAPIView):
 
     def get_queryset(self):
         return Request.get_queryset_for_user(self.request.user, 'service_catalog.view_request').prefetch_related(
-                "user", "operation", "instance__requester","instance__requester__profile","instance__resources","instance__requester__groups" ,"instance__quota_scope", "instance__service",
-                "operation__service", "approval_workflow_state", "approval_workflow_state__approval_workflow",
-                "approval_workflow_state__current_step",
-                "approval_workflow_state__current_step__approval_step", "approval_workflow_state__approval_step_states"
-            )
+            "user", "operation", "instance__requester", "instance__requester__profile", "instance__resources",
+            "instance__requester__groups", "instance__quota_scope", "instance__service",
+            "operation__service", "approval_workflow_state", "approval_workflow_state__approval_workflow",
+            "approval_workflow_state__current_step",
+            "approval_workflow_state__current_step__approval_step", "approval_workflow_state__approval_step_states"
+        )
 
     def get_serializer_class(self):
         if self.request.user.has_perm("service_catalog.view_admin_survey"):
@@ -63,11 +64,6 @@ def create(self, request, *args, **kwargs):
             Operation,
             id=kwargs.get('operation_id'), type__in=[OperationType.UPDATE, OperationType.DELETE], enabled=True)
 
-        if operation.is_admin_operation and not self.request.user.has_perm("service_catalog.admin_request_on_instance"):
-            raise PermissionDenied
-        if not operation.is_admin_operation and not self.request.user.has_perm("service_catalog.request_on_instance"):
-            raise PermissionDenied
-
         serializer = self.get_serializer(operation=operation, instance=self.get_object(), user=request.user,
                                          data=request.data)
         serializer.is_valid(raise_exception=True)

service_catalog/forms/__init__.py

@@ -11,3 +11,4 @@
 from .support_message_forms import *
 from .support_request_forms import *
 from .tower_server_forms import *
+from .form_utils import *