fix user assigning engagement to themselves (#2043)

met-api/src/met_api/services/membership_service.py

@@ -1,18 +1,19 @@
 """Service for membership."""
-from http import HTTPStatus
 from datetime import datetime
+from http import HTTPStatus
 
 from met_api.constants.membership_type import MembershipType
+from met_api.exceptions.business_exception import BusinessException
 from met_api.models import StaffUser as StaffUserModel
-from met_api.schemas.staff_user import StaffUserSchema
 from met_api.models.engagement import Engagement as EngagementModel
 from met_api.models.membership import Membership as MembershipModel
+from met_api.schemas.staff_user import StaffUserSchema
+from met_api.services import authorization
 from met_api.services.staff_user_service import KEYCLOAK_SERVICE, StaffUserService
-from met_api.utils.enums import KeycloakGroups, MembershipStatus
 from met_api.utils.constants import Groups
-from met_api.services import authorization
-from met_api.exceptions.business_exception import BusinessException
+from met_api.utils.enums import KeycloakGroups, MembershipStatus
 from met_api.utils.roles import Role
+from met_api.utils.token_info import TokenInfo
 
 
 class MembershipService:
@@ -37,15 +38,45 @@ def create_membership(engagement_id, request_json: dict):
         user_details = StaffUserSchema().dump(user)
         # attach and map groups
         StaffUserService.attach_groups([user_details])
-        # this makes sure duplicate membership doesnt happen.
-        # Can remove when user can have multiple roles with in same engagement.
-        # MembershipService._validate_member(engagement_id, user_details)
+        MembershipService._validate_create_membership(engagement_id, user_details)
         group_name, membership_type = MembershipService._get_membership_details(user_details)
         MembershipService._add_user_group(user_details, group_name)
         membership = MembershipService._create_membership_model(engagement_id, user_details, membership_type)
         membership.commit()
         return membership
 
+    @staticmethod
+    def _validate_create_membership(engagement_id, user_details):
+        """Validate create membership."""
+        request_user = TokenInfo.get_user_data()
+        if request_user.get('external_id') == user_details.get('external_id'):
+            raise BusinessException(
+                error='You cannot add yourself to an engagement.',
+                status_code=HTTPStatus.FORBIDDEN.value)
+
+        groups = user_details.get('groups')
+        if KeycloakGroups.EAO_IT_ADMIN.value in groups:
+            raise BusinessException(
+                error='This user is already a Superuser.',
+                status_code=HTTPStatus.CONFLICT.value)
+
+        existing_membership = MembershipModel.find_by_engagement_and_user_id(
+            engagement_id,
+            user_details.get('id'),
+            status=MembershipStatus.ACTIVE.value
+        )
+
+        if existing_membership:
+            raise BusinessException(
+                error=f'This {user_details.get("main_group", "user")} is already assigned to this engagement.',
+                status_code=HTTPStatus.CONFLICT.value)
+
+        request_user = TokenInfo.get_user_data()
+        if request_user.get('external_id') == user_details.get('external_id'):
+            raise BusinessException(
+                error='You cannot add yourself to an engagement.',
+                status_code=HTTPStatus.FORBIDDEN.value)
+
     @staticmethod
     def _get_membership_details(user_details):
         """Get the group name and membership type for the user based on their assigned groups."""
@@ -84,24 +115,6 @@ def _add_user_group(user: StaffUserModel, group_name=Groups.EAO_TEAM_MEMBER.name
             group_name=group_name
         )
 
-    @staticmethod
-    def _validate_member(engagement_id, user_details):
-        groups = user_details.get('groups')
-        if KeycloakGroups.EAO_IT_ADMIN.value in groups:
-            raise BusinessException(
-                error='This user is already a Superuser.',
-                status_code=HTTPStatus.CONFLICT.value)
-
-        existing_membership = MembershipModel.find_by_engagement_and_user_id(
-            engagement_id,
-            user_details.get('id'),
-            status=MembershipStatus.ACTIVE.value
-        )
-        if existing_membership:
-            raise BusinessException(
-                error=f'This {user_details.get("main_group", "user")} is already assigned to this engagement.',
-                status_code=HTTPStatus.CONFLICT.value)
-
     @staticmethod
     def _create_membership_model(engagement_id, user_details, membership_type=MembershipType.TEAM_MEMBER):
         if membership_type not in MembershipType.__members__.values():