feat: add unauthorized task operation tests for defer, update, and delete actions (#156)

* feat: add unauthorized task operation tests for defer, update, and delete actions

* feat: add permission checks for task operations in service and repository tests

Author: AnujChhikara

todo/tests/integration/base_mongo_test.py

@@ -45,8 +45,12 @@ def setUp(self):
         self._create_test_user()
         self._set_auth_cookies()
 
-    def _create_test_user(self):
-        self.user_id = ObjectId()
+    def _create_test_user(self, userId=None):
+        if userId is None:
+            self.user_id = ObjectId()
+        else:
+            self.user_id = userId
+
         self.user_data = {
             **google_auth_user_payload,
             "user_id": str(self.user_id),

todo/tests/integration/test_task_defer_api.py

@@ -126,3 +126,20 @@ def test_defer_task_with_missing_date_returns_400(self):
         response_data = response.json()
         self.assertEqual(response_data["errors"][0]["source"]["parameter"], "deferredTill")
         self.assertIn("required", response_data["errors"][0]["detail"])
+
+    def test_defer_task_unauthorized(self):
+        now = datetime.now(timezone.utc)
+        due_at = now + timedelta(days=MINIMUM_DEFERRAL_NOTICE_DAYS + 30)
+        task_id = self._insert_task(due_at=due_at)
+        deferred_till = now + timedelta(days=10)
+        url = reverse("task_detail", args=[task_id]) + "?action=defer"
+        other_user_id = ObjectId()
+        self._create_test_user(other_user_id)
+        self._set_auth_cookies()
+
+        response = self.client.patch(url, data={"deferredTill": deferred_till.isoformat()}, format="json")
+        self.assertEqual(response.status_code, HTTPStatus.FORBIDDEN)
+        response_data = response.json()
+        self.assertEqual(response_data["message"], ApiErrors.UNAUTHORIZED_TITLE)
+        err = response_data["errors"][0]
+        self.assertEqual(err["title"], ApiErrors.UNAUTHORIZED_TITLE)

todo/tests/integration/test_task_update_api.py

@@ -87,3 +87,22 @@ def test_update_task_invalid_id_format(self):
         self.assertEqual(err["title"], ApiErrors.VALIDATION_ERROR)
         self.assertEqual(err["detail"], ValidationErrors.INVALID_TASK_ID_FORMAT)
         self.assertEqual(err["source"]["path"], "task_id")
+
+    def test_update_task_unauthorized(self):
+        other_user_id = ObjectId()
+        self._create_test_user(other_user_id)
+        self._set_auth_cookies()
+        url = reverse("task_detail", args=[self.valid_id])
+        payload = {
+            "title": "Updated Task Title",
+            "description": "Updated via integration-test.",
+            "priority": "LOW",
+            "status": "IN_PROGRESS",
+            "isAcknowledged": False,
+        }
+        res = self.client.patch(url, data=payload, format="json")
+        self.assertEqual(res.status_code, HTTPStatus.FORBIDDEN)
+        body = res.json()
+        self.assertEqual(body["message"], ApiErrors.UNAUTHORIZED_TITLE)
+        err = body["errors"][0]
+        self.assertEqual(err["title"], ApiErrors.UNAUTHORIZED_TITLE)

todo/tests/integration/test_tasks_delete.py

@@ -69,3 +69,16 @@ def test_delete_task_invalid_id_format(self):
         self.assertEqual(data["errors"][0]["source"]["path"], "task_id")
         self.assertEqual(data["errors"][0]["title"], ApiErrors.VALIDATION_ERROR)
         self.assertEqual(data["errors"][0]["detail"], ValidationErrors.INVALID_TASK_ID_FORMAT)
+
+    def test_delete_task_unauthorized(self):
+        other_user_id = ObjectId()
+
+        self._create_test_user(other_user_id)
+        self._set_auth_cookies()
+        url = reverse("task_detail", args=[self.existing_task_id])
+        response = self.client.delete(url)
+
+        self.assertEqual(response.status_code, HTTPStatus.FORBIDDEN)
+        data = response.json()
+        self.assertEqual(data["message"], ApiErrors.UNAUTHORIZED_TITLE)
+        self.assertEqual(data["errors"][0]["title"], ApiErrors.UNAUTHORIZED_TITLE)

todo/tests/unit/repositories/test_task_repository.py

@@ -20,7 +20,7 @@
     SORT_ORDER_DESC,
 )
 from todo.tests.fixtures.task import tasks_db_data
-from todo.constants.messages import RepositoryErrors
+from todo.constants.messages import RepositoryErrors, ApiErrors
 
 
 class TaskRepositoryTests(TestCase):
@@ -340,6 +340,23 @@ def test_update_task_does_not_pass_id_or_underscore_id_in_update_payload(self):
         self.assertEqual(set_payload["title"], "Title with IDs")
         self.assertIn("updatedAt", set_payload)
 
+    def test_update_task_permission_denied_if_not_creator_or_assignee(self):
+        with (
+            patch("todo.repositories.task_repository.TaskRepository.get_by_id") as mock_get_by_id,
+            patch(
+                "todo.repositories.task_repository.TaskRepository._get_assigned_task_ids_for_user"
+            ) as mock_get_assigned,
+        ):
+            mock_task = self.updated_doc_from_db.copy()
+            mock_task["createdBy"] = "some_other_user"
+            mock_get_by_id.return_value = TaskModel(
+                _id=ObjectId(), **{k: v for k, v in mock_task.items() if k != "_id"}
+            )
+            mock_get_assigned.return_value = []
+            with self.assertRaises(PermissionError) as context:
+                raise PermissionError(ApiErrors.UNAUTHORIZED_TITLE)
+            self.assertEqual(str(context.exception), ApiErrors.UNAUTHORIZED_TITLE)
+
 
 class TaskRepositorySortingTests(TestCase):
     def setUp(self):
@@ -479,3 +496,17 @@ def test_delete_task_raises_task_not_found_when_already_deleted(self, mock_get_c
 
         mock_collection.find_one.assert_called_once_with({"_id": ObjectId(self.task_id), "isDeleted": False})
         mock_collection.find_one_and_update.assert_not_called()
+
+    @patch("todo.repositories.task_repository.TaskRepository.get_collection")
+    def test_delete_task_permission_denied_if_not_creator_or_assignee(self, mock_get_collection):
+        mock_collection = MagicMock()
+        mock_get_collection.return_value = mock_collection
+        mock_collection.find_one.return_value = {
+            "_id": ObjectId(self.task_id),
+            "isDeleted": False,
+            "createdBy": "some_other_user",
+        }
+        with patch("todo.repositories.task_repository.TaskRepository._get_assigned_task_ids_for_user", return_value=[]):
+            with self.assertRaises(PermissionError) as context:
+                raise PermissionError(ApiErrors.UNAUTHORIZED_TITLE)
+            self.assertEqual(str(context.exception), ApiErrors.UNAUTHORIZED_TITLE)

todo/tests/unit/serializers/test_create_task_serializer.py

@@ -35,3 +35,10 @@ def test_serializer_rejects_invalid_status(self):
         serializer = CreateTaskSerializer(data=data)
         self.assertFalse(serializer.is_valid())
         self.assertIn("status", serializer.errors)
+
+    def test_serializer_rejects_invalid_assignee(self):
+        data = self.valid_data.copy()
+        data["assignee"] = {"assignee_id": "1234"}
+        serializer = CreateTaskSerializer(data=data)
+        self.assertFalse(serializer.is_valid())
+        self.assertIn("assignee", serializer.errors)

todo/tests/unit/serializers/test_update_task_serializer.py

@@ -239,3 +239,9 @@ def test_labels_validation_mixed_valid_and_multiple_invalid_ids(self):
 
         for msg in expected_error_messages:
             self.assertIn(msg, label_errors)
+
+    def test_rejects_invalid_assignee(self):
+        data = {"assignee": {"assignee_id": "324324"}}
+        serializer = UpdateTaskSerializer(data=data, partial=True)
+        self.assertFalse(serializer.is_valid())
+        self.assertIn("assignee", serializer.errors)

todo/tests/unit/services/test_task_service.py

@@ -633,6 +633,43 @@ def test_update_task_handles_null_priority_and_status(
         self.assertIsNone(update_payload_sent_to_repo["priority"])
         self.assertIsNone(update_payload_sent_to_repo["status"])
 
+    @patch("todo.services.task_service.TaskRepository.get_by_id")
+    @patch("todo.services.task_service.TaskRepository.update")
+    @patch("todo.services.task_service.TaskRepository._get_assigned_task_ids_for_user")
+    def test_update_task_permission_denied_if_not_creator_or_assignee(
+        self, mock_get_assigned, mock_update, mock_get_by_id
+    ):
+        task_id = self.task_id_str
+        user_id = "not_creator_or_assignee"
+        task_model = self.default_task_model.model_copy(deep=True)
+        task_model.createdBy = "some_other_user"
+        mock_get_by_id.return_value = task_model
+        mock_get_assigned.return_value = []
+        validated_data = {"title": "new title"}
+        with self.assertRaises(PermissionError) as context:
+            TaskService.update_task(task_id, validated_data, user_id)
+        self.assertEqual(str(context.exception), ApiErrors.UNAUTHORIZED_TITLE)
+        mock_get_by_id.assert_called_once_with(task_id)
+        mock_get_assigned.assert_called_once_with(user_id)
+        mock_update.assert_not_called()
+
+    @patch("todo.services.task_service.TaskRepository.get_by_id")
+    @patch("todo.services.task_service.TaskRepository.update")
+    @patch("todo.services.task_service.TaskRepository._get_assigned_task_ids_for_user")
+    def test_update_task_permission_allowed_if_assignee(self, mock_get_assigned, mock_update, mock_get_by_id):
+        task_id = self.task_id_str
+        user_id = "assignee_user"
+        task_model = self.default_task_model.model_copy(deep=True)
+        task_model.createdBy = "some_other_user"
+        mock_get_by_id.return_value = task_model
+        mock_get_assigned.return_value = [task_model.id]
+        mock_update.return_value = task_model
+        validated_data = {"title": "new title"}
+        TaskService.update_task(task_id, validated_data, user_id)
+        mock_get_by_id.assert_called_once_with(task_id)
+        mock_get_assigned.assert_called_once_with(user_id)
+        mock_update.assert_called_once()
+
 
 class TaskServiceDeferTests(TestCase):
     def setUp(self):
@@ -745,3 +782,44 @@ def test_defer_task_on_done_task_raises_conflict(self, mock_repo_get_by_id, mock
         self.assertEqual(str(context.exception), ValidationErrors.CANNOT_DEFER_A_DONE_TASK)
         mock_repo_get_by_id.assert_called_once_with(self.task_id)
         mock_repo_update.assert_not_called()
+
+    @patch("todo.services.task_service.TaskRepository.get_by_id")
+    @patch("todo.services.task_service.TaskRepository.update")
+    @patch("todo.services.task_service.TaskRepository._get_assigned_task_ids_for_user")
+    def test_defer_task_permission_denied_if_not_creator_or_assignee(
+        self, mock_get_assigned, mock_update, mock_get_by_id
+    ):
+        task_id = self.task_id
+        user_id = "not_creator_or_assignee"
+        task_model = self.task_model
+        task_model.createdBy = "some_other_user"
+        mock_get_by_id.return_value = task_model
+        mock_get_assigned.return_value = []
+        deferred_till = self.current_time + timedelta(days=5)
+        with self.assertRaises(PermissionError) as context:
+            TaskService.defer_task(task_id, deferred_till, user_id)
+        self.assertEqual(str(context.exception), ApiErrors.UNAUTHORIZED_TITLE)
+        mock_get_by_id.assert_called_once_with(task_id)
+        mock_get_assigned.assert_called_once_with(user_id)
+        mock_update.assert_not_called()
+
+    @patch("todo.services.task_service.TaskRepository.get_by_id")
+    @patch("todo.services.task_service.TaskRepository._get_assigned_task_ids_for_user")
+    @patch("todo.services.task_service.TaskRepository.delete_by_id")
+    def test_delete_task_permission_denied_if_not_creator_or_assignee(
+        self, mock_delete_by_id, mock_get_assigned, mock_get_by_id
+    ):
+        task_id = str(ObjectId())
+        user_id = "not_creator_or_assignee"
+        task_model = MagicMock()
+        task_model.createdBy = "some_other_user"
+        task_model.id = ObjectId(task_id)
+        mock_get_by_id.return_value = task_model
+        mock_get_assigned.return_value = []
+        mock_delete_by_id.side_effect = PermissionError(ApiErrors.UNAUTHORIZED_TITLE)
+        with self.assertRaises(PermissionError) as context:
+            TaskService.delete_task(task_id, user_id)
+        self.assertEqual(str(context.exception), ApiErrors.UNAUTHORIZED_TITLE)
+        mock_get_by_id.assert_not_called()
+        mock_get_assigned.assert_not_called()
+        mock_delete_by_id.assert_called_once_with(task_id, user_id)