Dev To Main Sync

todo/constants/messages.py

@@ -84,6 +84,8 @@ class ValidationErrors:
     SEARCH_QUERY_EMPTY = "Search query cannot be empty"
     TASK_ID_STRING_REQUIRED = "Task ID must be a string."
     INVALID_IS_ACTIVE_VALUE = "Invalid value for is_active"
+    USER_NOT_TEAM_MEMBER = "User is not a member of the team"
+    POC_NOT_PROVIDED = "POC is required for team update"
 
 
 # Auth messages

todo/exceptions/exception_handler.py

@@ -20,10 +20,11 @@
     TokenInvalidError,
     RefreshTokenExpiredError,
     APIException,
-    UserNotFoundException,
     TokenMissingError,
 )
 
+from todo.exceptions.user_exceptions import UserNotFoundException
+
 
 def format_validation_errors(errors) -> List[ApiErrorDetail]:
     formatted_errors = []

todo/repositories/task_repository.py

@@ -130,8 +130,13 @@ def _get_assigned_task_ids_for_user(cls, user_id: str) -> List[ObjectId]:
         if team_ids:
             # Get teams where user is POC
             poc_teams = TeamRepository.get_collection().find(
-                {"_id": {"$in": [ObjectId(team_id) for team_id in team_ids]}, "is_deleted": False, "poc_id": user_id}
+                {
+                    "_id": {"$in": [ObjectId(team_id) for team_id in team_ids]},
+                    "is_deleted": False,
+                    "poc_id": {"$in": [ObjectId(user_id), user_id]},
+                }
             )
+
             poc_team_ids = [str(team["_id"]) for team in poc_teams]
 
             # Get team assignments for POC teams

todo/serializers/update_team_serializer.py

@@ -7,35 +7,12 @@
 class UpdateTeamSerializer(serializers.Serializer):
     """
     Serializer for updating team details.
-    All fields are optional for PATCH operations.
-    """
-
-    name = serializers.CharField(max_length=100, required=False, allow_blank=False)
-    description = serializers.CharField(max_length=500, required=False, allow_blank=True, allow_null=True)
-    poc_id = serializers.CharField(required=False, allow_null=True, allow_blank=False)
-    member_ids = serializers.ListField(child=serializers.CharField(), required=False, allow_empty=True, default=None)
 
-    def validate_name(self, value):
-        if value is not None and not value.strip():
-            raise serializers.ValidationError("Team name cannot be blank")
-        return value.strip() if value else None
+    """
 
-    def validate_description(self, value):
-        if value is not None:
-            return value.strip()
-        return value
+    poc_id = serializers.CharField(required=True, allow_null=False, allow_blank=False)
 
     def validate_poc_id(self, value):
-        if not value or not value.strip():
-            return None
         if not ObjectId.is_valid(value):
             raise serializers.ValidationError(ValidationErrors.INVALID_OBJECT_ID.format(value))
         return value
-
-    def validate_member_ids(self, value):
-        if value is None:
-            return value
-        for member_id in value:
-            if not ObjectId.is_valid(member_id):
-                raise serializers.ValidationError(ValidationErrors.INVALID_OBJECT_ID.format(member_id))
-        return value

todo/services/team_service.py

@@ -1,12 +1,11 @@
 from todo.dto.team_dto import CreateTeamDTO, TeamDTO
-from todo.dto.update_team_dto import UpdateTeamDTO
 from todo.dto.responses.create_team_response import CreateTeamResponse
 from todo.dto.responses.get_user_teams_response import GetUserTeamsResponse
 from todo.models.team import TeamModel, UserTeamDetailsModel
 from todo.models.common.pyobjectid import PyObjectId
 from todo.repositories.team_creation_invite_code_repository import TeamCreationInviteCodeRepository
 from todo.repositories.team_repository import TeamRepository, UserTeamDetailsRepository
-from todo.constants.messages import AppMessages
+from todo.constants.messages import AppMessages, ApiErrors, ValidationErrors
 from todo.constants.role import RoleName
 from todo.utils.invite_code_utils import generate_invite_code
 from typing import List
@@ -23,6 +22,7 @@
     CannotRemoveTeamPOCException,
 )
 
+
 DEFAULT_ROLE_ID = "1"
 
 
@@ -324,55 +324,46 @@ def join_team_by_invite_code(cls, invite_code: str, user_id: str) -> TeamDTO:
         )
 
     @classmethod
-    def update_team(cls, team_id: str, dto: UpdateTeamDTO, updated_by_user_id: str) -> TeamDTO:
+    def update_team(cls, team_id: str, poc_id: str, user_id: str) -> TeamDTO:
         """
         Update a team by its ID.
 
         Args:
             team_id: ID of the team to update
-            dto: Team update data including name, description, and POC
-            updated_by_user_id: ID of the user updating the team
+            poc_id: ID of the new POC
+            user_id: ID of the user updating the team
 
         Returns:
             TeamDTO with the updated team details
 
         Raises:
             ValueError: If team update fails or team not found
         """
-        try:
-            # Check if team exists
-            existing_team = TeamRepository.get_by_id(team_id)
-            if not existing_team:
-                raise ValueError(f"Team with id {team_id} not found")
+        existing_team = TeamRepository.get_by_id(team_id)
+        if not existing_team:
+            raise ValueError(f"Team with id {team_id} not found")
+
+        cls._validate_is_user_team_admin(team_id, user_id, existing_team)
 
-            # Prepare update data
-            update_data = {}
-            if dto.name is not None:
-                update_data["name"] = dto.name
-            if dto.description is not None:
-                update_data["description"] = dto.description
-            if dto.poc_id is not None:
-                update_data["poc_id"] = PyObjectId(dto.poc_id) if dto.poc_id and dto.poc_id.strip() else None
+        update_data = {}
 
+        validation_error = cls._validate_is_user_team_member(team_id, poc_id)
+        if validation_error:
+            return validation_error
+        update_data["poc_id"] = PyObjectId(poc_id)
+
+        try:
             # Update the team
-            updated_team = TeamRepository.update(team_id, update_data, updated_by_user_id)
+            updated_team = TeamRepository.update(team_id, update_data, user_id)
             if not updated_team:
                 raise ValueError(f"Failed to update team with id {team_id}")
 
-            # Handle member updates if provided
-            if dto.member_ids is not None:
-                from todo.repositories.team_repository import UserTeamDetailsRepository
-
-                success = UserTeamDetailsRepository.update_team_members(team_id, dto.member_ids, updated_by_user_id)
-                if not success:
-                    raise ValueError(f"Failed to update team members for team with id {team_id}")
-
             # Audit log for team update
             AuditLogRepository.create(
                 AuditLogModel(
                     team_id=PyObjectId(team_id),
                     action="team_updated",
-                    performed_by=PyObjectId(updated_by_user_id),
+                    performed_by=PyObjectId(user_id),
                 )
             )
 
@@ -388,7 +379,6 @@ def update_team(cls, team_id: str, dto: UpdateTeamDTO, updated_by_user_id: str)
                 created_at=updated_team.created_at,
                 updated_at=updated_team.updated_at,
             )
-
         except Exception as e:
             raise ValueError(f"Failed to update team: {str(e)}")
 
@@ -530,3 +520,31 @@ def remove_member_from_team(cls, user_id: str, team_id: str, removed_by_user_id:
         TaskAssignmentService.reassign_tasks_from_user_to_team(user_id, team_id, removed_by_user_id)
 
         return True
+
+    @classmethod
+    def _validate_is_user_team_admin(cls, team_id: str, user_id: str, team):
+        """
+        Validate that the user has admin role in the team.
+        """
+        if str(team.created_by) == user_id:
+            return
+
+        if UserRoleService.has_role(user_id, RoleName.ADMIN.value, RoleScope.TEAM.value, team_id):
+            return
+
+        raise PermissionError(ApiErrors.UNAUTHORIZED_TITLE)
+
+    @classmethod
+    def _validate_is_user_team_member(cls, team_id: str, poc_id: str):
+        """
+        Validate that the user is a member of the team.
+        Returns ApiErrorResponse if validation fails, None if validation passes.
+        """
+
+        if not UserRoleService.has_role(poc_id, RoleName.MEMBER.value, RoleScope.TEAM.value, team_id):
+            return ApiErrorResponse(
+                statusCode=400,
+                message=ValidationErrors.USER_NOT_TEAM_MEMBER,
+                errors=[ApiErrorDetail(detail=ValidationErrors.USER_NOT_TEAM_MEMBER)],
+            )
+        return

todo/tests/integration/test_team_update.py

@@ -0,0 +1,156 @@
+from http import HTTPStatus
+from django.urls import reverse
+from bson import ObjectId
+from datetime import datetime, timezone
+import json
+
+from todo.tests.integration.base_mongo_test import AuthenticatedMongoTestCase
+from todo.constants.messages import ApiErrors, ValidationErrors
+
+
+class TeamUpdateIntegrationTests(AuthenticatedMongoTestCase):
+    def setUp(self):
+        super().setUp()
+        self.db.teams.delete_many({})
+        self.db.user_roles.delete_many({})
+
+        self.team_id = str(ObjectId())
+        self.owner_id = str(self.user_id)
+        self.admin_id = str(ObjectId())
+        self.member_id = str(ObjectId())
+        self.non_member_id = str(ObjectId())
+
+        team_doc = {
+            "_id": ObjectId(self.team_id),
+            "name": "Test Team",
+            "description": "Test Description",
+            "poc_id": ObjectId(self.member_id),
+            "invite_code": "TEST123",
+            "created_by": ObjectId(self.owner_id),
+            "updated_by": ObjectId(self.owner_id),
+            "created_at": datetime.now(timezone.utc),
+            "updated_at": datetime.now(timezone.utc),
+            "is_deleted": False,
+        }
+        self.db.teams.insert_one(team_doc)
+
+        owner_role_doc = {
+            "_id": ObjectId(),
+            "name": "owner",
+            "scope": "TEAM",
+            "description": "Team Owner",
+            "created_at": datetime.now(timezone.utc),
+            "updated_at": datetime.now(timezone.utc),
+        }
+        self.db.roles.insert_one(owner_role_doc)
+
+        member_role_doc = {
+            "_id": ObjectId(),
+            "name": "member",
+            "scope": "TEAM",
+            "description": "Team Member",
+            "created_at": datetime.now(timezone.utc),
+            "updated_at": datetime.now(timezone.utc),
+        }
+        self.db.roles.insert_one(member_role_doc)
+
+        owner_role = {
+            "_id": ObjectId(),
+            "user_id": ObjectId(self.owner_id),
+            "role_id": owner_role_doc["_id"],
+            "role_name": "owner",
+            "scope": "TEAM",
+            "team_id": ObjectId(self.team_id),
+            "is_active": True,
+            "created_by": ObjectId(self.owner_id),
+            "created_at": datetime.now(timezone.utc),
+        }
+        self.db.user_roles.insert_one(owner_role)
+
+        authenticated_user_role = {
+            "_id": ObjectId(),
+            "user_id": str(self.user_id),
+            "role_id": owner_role_doc["_id"],
+            "role_name": "owner",
+            "scope": "TEAM",
+            "team_id": str(self.team_id),
+            "is_active": True,
+            "created_by": str(self.owner_id),
+            "created_at": datetime.now(timezone.utc),
+        }
+        self.db.user_roles.insert_one(authenticated_user_role)
+
+        member_role = {
+            "_id": ObjectId(),
+            "user_id": self.member_id,
+            "role_id": member_role_doc["_id"],
+            "role_name": "member",
+            "scope": "TEAM",
+            "team_id": self.team_id,
+            "is_active": True,
+            "created_by": self.owner_id,
+            "created_at": datetime.now(timezone.utc),
+        }
+        self.db.user_roles.insert_one(member_role)
+
+        self.db.users.insert_one(
+            {
+                "_id": ObjectId(self.member_id),
+                "google_id": "member_google_id",
+                "email_id": "member@example.com",
+                "name": "Member User",
+                "picture": "member_picture",
+                "createdAt": datetime.now(timezone.utc),
+                "updatedAt": datetime.now(timezone.utc),
+            }
+        )
+
+        self.db.users.insert_one(
+            {
+                "_id": ObjectId(self.non_member_id),
+                "google_id": "non_member_google_id",
+                "email_id": "nonmember@example.com",
+                "name": "Non Member User",
+                "picture": "non_member_picture",
+                "createdAt": datetime.now(timezone.utc),
+                "updatedAt": datetime.now(timezone.utc),
+            }
+        )
+
+        self.existing_team_id = self.team_id
+        self.non_existent_id = str(ObjectId())
+        self.invalid_team_id = "invalid-team-id"
+
+    def test_update_team_success_by_owner(self):
+        url = reverse("team_detail", args=[self.existing_team_id])
+        response = self.client.patch(
+            url,
+            data=json.dumps({"poc_id": self.member_id}),
+            content_type="application/json",
+        )
+
+        self.assertEqual(response.status_code, HTTPStatus.OK)
+        data = response.json()
+        self.assertEqual(data["poc_id"], self.member_id)
+        self.assertNotIn("invite_code", data)
+
+    def test_update_team_unauthorized_user(self):
+        other_user_id = ObjectId()
+        self._create_test_user(other_user_id)
+        self._set_auth_cookies()
+
+        url = reverse("team_detail", args=[self.existing_team_id])
+        response = self.client.patch(url, data=json.dumps({"name": "Updated Team"}), content_type="application/json")
+
+        self.assertEqual(response.status_code, HTTPStatus.FORBIDDEN)
+        data = response.json()
+        self.assertEqual(data["detail"], ApiErrors.UNAUTHORIZED_TITLE)
+
+    def test_update_team_invalid_poc_id_format(self):
+        url = reverse("team_detail", args=[self.existing_team_id])
+        response = self.client.patch(url, data=json.dumps({"poc_id": "invalid-id"}), content_type="application/json")
+
+        self.assertEqual(response.status_code, HTTPStatus.BAD_REQUEST)
+        data = response.json()
+        self.assertIn("poc_id", data["errors"])
+        self.assertIn(ValidationErrors.INVALID_OBJECT_ID.format("invalid-id"), str(data["errors"]["poc_id"]))