feat: group access request - allow user to view new group access request form

User can create request to add access to a group. Added the form to capture the group access request

Author: Kartikkumar-Shetty

Access/accessrequest_helper.py

@@ -42,6 +42,37 @@
     "msg": "User Identity for module {access_tag} not setup by the user",
 }
 
+REQUEST_SUCCESS_MSG = {
+    "title": "{request_id}  Request Submitted",
+    "msg": "Once approved you will receive the update. {access_label}",
+}
+REQUEST_DUPLICATE_ERR_MSG = {
+    "title": "{access_tag}: Duplicate Request not submitted",
+    "msg": "Access already granted or request in pending state. {access_label}",
+}
+REQUEST_ERR_MSG = {
+    "error_msg": "Invalid Request",
+    "msg": "Please Contact Admin",
+}
+REQUEST_EMPTY_FORM_ERR_MSG = {
+    "error_msg": "The submitted form is empty. Tried direct access to reqeust access page",
+    "msg": "Error Occured while submitting your Request. Please contact the Admin",
+}
+
+REQUEST_ACCESS_AUTO_APPROVED_MSG = {
+    "title": "{request_id}  Request Approved",
+    "msg": "Once granted you will receive the update",
+}
+
+REQUEST_DB_ERR_MSG = {
+    "error_msg": "Error Saving Request",
+    "msg": "Please Contact Admin",
+}
+REQUEST_IDENTITY_NOT_SETUP_ERR_MSG = {
+    "error_msg": "Identity not setup",
+    "msg": "User Identity for module {access_tag} not setup by the user",
+}
+
 
 def requestAccessGet(request):
     context = {}
@@ -155,7 +186,10 @@ def get_pending_accesses_from_modules(access_user):
     group_requests = {}
 
     logger.info("Start looping all access modules")
-    for access_module_tag, access_module in helpers.get_available_access_modules().items():
+    for (
+        access_module_tag,
+        access_module,
+    ) in helpers.get_available_access_modules().items():
         access_module_start_time = time.time()
 
         try:
@@ -270,10 +304,10 @@ def create_request(auth_user, access_request_form):
     json_response = {}
     json_response["status"] = []
     json_response["status_list"] = []
-    extra_fields = _get_extra_fields(access_request=access_request)
+    extra_fields = get_extra_fields(access_request=access_request)
 
     for index1, access_type in enumerate(access_request["accessRequests"]):
-        access_labels = _validate_access_labels(
+        access_labels = validate_access_labels(
             access_labels_json=access_request["accessLabel"][index1],
             access_type=access_type,
         )
@@ -296,7 +330,7 @@ def create_request(auth_user, access_request_form):
             access_labels, auth_user, is_group=False
         )
 
-        extra_field_labels = _get_extra_field_labels(access_module)
+        extra_field_labels = get_extra_field_labels(access_module)
 
         if extra_fields and extra_field_labels:
             for field in extra_field_labels:
@@ -395,14 +429,14 @@ def _create_access_mapping(
     return access
 
 
-def _get_extra_field_labels(access_module):
+def get_extra_field_labels(access_module):
     try:
         return access_module.get_extra_fields()
     except Exception:
         return []
 
 
-def _get_extra_fields(access_request):
+def get_extra_fields(access_request):
     if "extraFields" in access_request:
         return access_request["extraFields"]
     return []
@@ -430,7 +464,7 @@ def _validate_access_request(access_request_form, user):
     return {}, access_request
 
 
-def _validate_access_labels(access_labels_json, access_type):
+def validate_access_labels(access_labels_json, access_type):
     if access_labels_json is None or access_labels_json == "":
         raise Exception("No fields were selected in the request. Please try again.")
     access_labels = json.loads(access_labels_json)

Access/group_helper.py

@@ -1,18 +1,15 @@
-from Access.models import User, GroupV2, MembershipV2
-from Access import helpers, views_helper, notifications
+from Access.models import User, GroupV2, MembershipV2, AccessV2
+from Access import helpers, views_helper, notifications, accessrequest_helper
 from django.db import transaction
 import datetime
 import logging
 from bootprocess import general
 from Access.views_helper import generateUserMappings, executeGroupAccess
-from BrowserStackAutomation.settings import (
-    MAIL_APPROVER_GROUPS,
-    PERMISSION_CONSTANTS,
-)
+from BrowserStackAutomation.settings import MAIL_APPROVER_GROUPS, PERMISSION_CONSTANTS
+from . import helpers as helper
 from Access.background_task_manager import background_task
 import json
 
-
 logger = logging.getLogger(__name__)
 
 NEW_GROUP_CREATE_SUCCESS_MESSAGE = {
@@ -60,6 +57,12 @@
 }
 
 
+class GroupAccessExistsException(Exception):
+    def __init__(self):
+        self.message = "Group Access Exists"
+        super().__init__(self.message)
+
+
 def create_group(request):
     base_datetime_prefix = datetime.datetime.utcnow().strftime("%Y%m%d%H%M%S")
     try:
@@ -134,7 +137,9 @@ def create_group(request):
 
 def get_generic_access(group_mapping):
     access_details = {}
-    access_module = helpers.get_available_access_module_from_tag(group_mapping.access.access_tag)
+    access_module = helpers.get_available_access_module_from_tag(
+        group_mapping.access.access_tag
+    )
     if not access_module:
         return {}
 
@@ -597,6 +602,150 @@ def accept_member(request, requestId, shouldRender=True):
         return context
 
 
+def get_group_access(form_data, auth_user):
+    data = dict(form_data.lists())
+    group_name = data["groupName"][0]
+    context = {}
+    context["accesses"] = []
+
+    group = GroupV2.get_active_group_by_name(group_name=group_name)
+    validation_error = validate_group_access_create_request(
+        group=group, auth_user=auth_user
+    )
+    if validation_error:
+        context["status"] = validation_error
+
+    access_module_list = data["accessList"]
+    for module_value in access_module_list:
+        module = helper.get_available_access_modules()[module_value]
+        try:
+            extra_fields = module.get_extra_fields()
+        except Exception:
+            extra_fields = []
+
+        context["genericForm"] = True
+        context["accesses"].append(
+            {
+                "formDesc": module.access_desc(),
+                "accessTag": module.tag(),
+                "accessTypes": module.access_types(),
+                "accessRequestData": module.access_request_data(
+                    form_data, is_group=True
+                ),
+                "extraFields": extra_fields,
+                "accessRequestPath": module.fetch_access_request_form_path(),
+            }
+        )
+    context["groupName"] = group_name
+    return context
+
+
+def save_group_access_request(form_data, auth_user):
+    access_request = dict(form_data.lists())
+    group_name = access_request["groupName"][0]
+    group = GroupV2.get_active_group_by_name(group_name=group_name)
+
+    context = {"status_list": []}
+    validation_error = validate_group_access_create_request(
+        group=group, auth_user=auth_user
+    )
+    if validation_error:
+        context["status"] = validation_error
+
+    for accessIndex, access_type in enumerate(access_request["accessType"]):
+        access_module = helper.get_available_access_modules()[access_type]
+        access_labels = accessrequest_helper.validate_access_labels(
+            access_labels_json=access_request["accessLabel"][accessIndex],
+            access_type=access_type,
+        )
+        extra_fields = accessrequest_helper.get_extra_fields(access_request)
+        extra_field_labels = accessrequest_helper.get_extra_field_labels(access_module)
+
+        module_access_labels = access_module.validate_request(
+            access_labels, auth_user, is_group=False
+        )
+        if extra_fields and extra_field_labels:
+            for field in extra_field_labels:
+                module_access_labels[0][field] = extra_fields[0]
+                extra_fields = extra_fields[1:]
+
+        request_id = (
+            auth_user.username
+            + "-"
+            + access_type
+            + "-"
+            + datetime.datetime.utcnow().strftime("%Y%m%d%H%M%S")
+        )
+        with transaction.atomic():
+            for labelIndex, access_label in enumerate(module_access_labels):
+                request_id = request_id + "_" + str(labelIndex)
+                try:
+                    _create_group_access_mapping(
+                        group=group,
+                        user=auth_user.user,
+                        request_id=request_id,
+                        access_type=access_type,
+                        access_label=access_label,
+                        access_reason=access_request["accessReason"],
+                    )
+                    context["status_list"].append(
+                        {
+                            "title": request_id + " Request Submitted",
+                            "msg": "Once approved you will receive the update "
+                            + json.dumps(access_label),
+                        }
+                    )
+                except GroupAccessExistsException:
+                    context["status_list"].append(
+                        {
+                            "title": request_id + " Request Submitted",
+                            "msg": "Once approved you will receive the update "
+                            + json.dumps(access_label),
+                        }
+                    )
+        email_destination = access_module.get_approvers()
+        member_list = group.get_all_approved_members()
+        notifications.send_group_access_add_email(
+            destination=email_destination,
+            group_name=group_name,
+            requester=auth_user.user.email,
+            request_id=request_id,
+            member_list=member_list,
+        )
+
+    return context
+
+
+def _create_group_access_mapping(
+    group, user, request_id, access_type, access_label, access_reason
+):
+    access = AccessV2.get(access_type=access_type, access_label=access_label)
+    if not access:
+        access = AccessV2.objects.create(
+            access_tag=access_type, access_label=access_label
+        )
+    else:
+        if group.check_access_exist(access):
+            raise GroupAccessExistsException()
+    group.add_access(
+        request_id=request_id,
+        requested_by=user,
+        request_reason=access_reason,
+        access=access,
+    )
+
+
+def validate_group_access_create_request(group, auth_user):
+    if not group:
+        logger.exception("This Group is not yet approved")
+        return {"title": "Permisison Denied", "msg": "This Group is not yet approved"}
+
+    if not (group.is_owner(auth_user.user) or auth_user.is_superuser):
+        logger.exception("Permission denied, you're not owner of this group")
+        return {"title": "Permision Denied", "msg": "You're not owner of this group"}
+    return None
+
+
 def remove_member(request):
     try:
         membership_id = request.POST.get("membershipId")