feat: restrict member removal to admins and reassign tasks

todo/constants/messages.py

@@ -54,6 +54,7 @@ class ApiErrors:
     USER_NOT_FOUND_GENERIC = "User not found."
     SEARCH_QUERY_EMPTY = "Search query cannot be empty"
     TASK_ALREADY_IN_WATCHLIST = "Task is already in the watchlist"
+    CANNOT_REMOVE_POC = "POC cannot be removed from a team"
 
 
 # Validation error messages

todo/exceptions/team_exceptions.py

@@ -0,0 +1,18 @@
+from todo.constants.messages import ApiErrors
+
+class BaseTeamException(Exception):
+    def __init__(self, message: str):
+        self.message = message
+        super().__init__(self.message)
+
+class CannotRemoveOwnerException(BaseTeamException):
+    def __init__(self, message = ApiErrors.UNAUTHORIZED_TITLE):
+        super().__init__(message)
+
+class NotTeamAdminException(BaseTeamException):
+    def __init__(self, message = ApiErrors.UNAUTHORIZED_TITLE):
+        super().__init__(message)
+
+class CannotRemoveTeamPOC(BaseTeamException):
+    def __init__(self, message = ApiErrors.CANNOT_REMOVE_POC):
+        super().__init__(message)

todo/models/audit_log.py

@@ -23,3 +23,5 @@ class AuditLogModel(Document):
     assignee_to: PyObjectId | None = None
     # For general user reference (who performed the action)
     performed_by: PyObjectId | None = None
+    # For multiple task reassignments
+    task_count: str | None = None

todo/repositories/task_assignment_repository.py

@@ -6,8 +6,7 @@
 from todo.models.task_assignment import TaskAssignmentModel
 from todo.repositories.common.mongo_repository import MongoRepository
 from todo.models.common.pyobjectid import PyObjectId
-
-
+from todo.constants.task import TaskStatus
 class TaskAssignmentRepository(MongoRepository):
     collection_name = TaskAssignmentModel.collection_name
 
@@ -224,3 +223,61 @@ def deactivate_by_task_id(cls, task_id: str, user_id: str) -> bool:
             return result.modified_count > 0
         except Exception:
             return False
+
+    @classmethod
+    def reassign_tasks_from_user_to_team(cls, user_id: str, team_id: str, removed_by_user_id: str):
+        """
+        Reassign all tasks of user to team
+        """
+        collection = cls.get_collection()
+        now = datetime.now(timezone.utc)
+        pipeline = [
+            {
+                "$match": {
+                    "team_id": team_id,
+                    "assignee_id": user_id,
+                    "user_type": "user",
+                    "is_active": True
+                    }
+            },
+            {"$addFields": {"task_id_obj": { "$toObjectId": "$task_id" }}},
+            {"$lookup": {
+                "from": "tasks",
+                "localField": "task_id_obj",
+                "foreignField": "_id",
+                "as": "task_info"
+            }},
+            { "$unwind": "$task_info" },
+            { "$match": { "task_info.status": { "$ne": TaskStatus.DONE.value } } }
+        ]
+        user_task_assignments = list(collection.aggregate(pipeline))
+        if not user_task_assignments:
+            return 0
+
+        # Deactivate current assignment (try using both ObjectId and string)
+        collection.update_many(
+            {"assignee_id": ObjectId(user_id), "team_id": ObjectId(team_id), "is_active": True },
+            {"$set": {"is_active": False, "updated_by": ObjectId(removed_by_user_id), "updated_at": now}})
+
+        collection.update_many(
+            {"assignee_id": user_id, "team_id": team_id, "is_active": True},
+            {"$set": {"is_active": False, "updated_by": ObjectId(removed_by_user_id), "updated_at": now}}
+        )
+
+        new_assignments = []
+        for task in user_task_assignments:
+            new_assignments.append(TaskAssignmentModel(
+                _id=PyObjectId(),
+                task_id=task["task_id_obj"],
+                assignee_id=PyObjectId(team_id),
+                user_type="team",
+                created_by=PyObjectId(removed_by_user_id),
+                updated_by=None,
+                team_id=PyObjectId(team_id)
+            ).model_dump(mode="json", by_alias=True, exclude_none=True))
+        if new_assignments:
+            collection.insert_many(new_assignments)
+        return len(new_assignments)
+
+
+

todo/services/task_assignment_service.py

@@ -14,7 +14,6 @@
 from todo.models.audit_log import AuditLogModel
 from todo.repositories.audit_log_repository import AuditLogRepository
 
-
 class TaskAssignmentService:
     @classmethod
     def create_task_assignment(cls, dto: CreateTaskAssignmentDTO, user_id: str) -> CreateTaskAssignmentResponse:
@@ -149,3 +148,18 @@ def delete_task_assignment(cls, task_id: str, user_id: str) -> bool:
         Delete task assignment by task ID.
         """
         return TaskAssignmentRepository.delete_assignment(task_id, user_id)
+
+    @classmethod
+    def reassign_tasks_from_user_to_team(cls, user_id: str, team_id: str, user_type: str, removed_by_user_id: str):
+        """
+        Reassign all tasks of user to team
+        """
+        assignment = TaskAssignmentRepository.reassign_tasks_from_user_to_team(user_id, team_id, removed_by_user_id)
+        if assignment != 0:
+            AuditLogRepository.create(AuditLogModel(
+                team_id=PyObjectId(team_id),
+                performed_by = PyObjectId(removed_by_user_id),
+                task_count = str(assignment),
+                action="tasks_reassigned_to_team"
+            ))
+

todo/services/team_service.py

@@ -12,7 +12,10 @@
 from todo.models.audit_log import AuditLogModel
 from todo.repositories.audit_log_repository import AuditLogRepository
 from todo.dto.responses.error_response import ApiErrorResponse, ApiErrorDetail
-
+from todo.services.user_role_service import UserRoleService
+from todo.constants.role import RoleScope, RoleName
+from todo.services.task_assignment_service import TaskAssignmentService
+from todo.exceptions.team_exceptions import (CannotRemoveOwnerException, NotTeamAdminException, CannotRemoveTeamPOC)
 DEFAULT_ROLE_ID = "1"
 
 
@@ -479,7 +482,24 @@ class TeamOrUserNotFound(Exception):
     @classmethod
     def remove_member_from_team(cls, user_id: str, team_id: str, removed_by_user_id: str = None):
         from todo.repositories.user_team_details_repository import UserTeamDetailsRepository
-
+        user_roles = UserRoleService.get_user_roles(user_id, RoleScope.TEAM.value, team_id)
+        user_role_ids = [roles["role_id"] for roles in user_roles]
+        team = TeamService.get_team_by_id(team_id)
+        # Authentication Checks
+        if user_id != removed_by_user_id:
+            if not UserRoleService.has_role(removed_by_user_id, RoleName.ADMIN.value, RoleScope.TEAM.value, team_id):
+                raise NotTeamAdminException
+            elif user_id == team.poc_id:
+                raise CannotRemoveTeamPOC
+        if user_id == team.created_by:
+            raise CannotRemoveOwnerException
+        if user_id == team.poc_id:
+            cls.update_team(team_id, UpdateTeamDTO(poc_id=team.created_by), removed_by_user_id)
+        # # Remove User Roles
+        for role_id in user_role_ids:
+            UserRoleService.remove_role_by_id(user_id, role_id, RoleScope.TEAM.value, team_id)
+        # Reassign Tasks:
+        TaskAssignmentService.reassign_tasks_from_user_to_team(user_id, team_id, 'team', removed_by_user_id)
         success = UserTeamDetailsRepository.remove_member_from_team(user_id=user_id, team_id=team_id)
         if not success:
             raise cls.TeamOrUserNotFound()
@@ -488,7 +508,7 @@ def remove_member_from_team(cls, user_id: str, team_id: str, removed_by_user_id:
         AuditLogRepository.create(
             AuditLogModel(
                 team_id=PyObjectId(team_id),
-                action="member_removed_from_team",
+                action="member_removed_from_team" if user_id != removed_by_user_id else "member_left_team",
                 performed_by=PyObjectId(removed_by_user_id) if removed_by_user_id else PyObjectId(user_id),
             )
         )

todo/views/team.py

@@ -448,6 +448,8 @@ def get(self, request: Request, team_id: str):
                 entry["status_from"] = log.status_from
             if log.status_to:
                 entry["status_to"] = log.status_to
+            if log.task_count:
+                entry["task_count"] = log.task_count
             timeline.append(entry)
         return Response({"timeline": timeline}, status=status.HTTP_200_OK)