4a0746b45e
* fix: scope workspace user preference filter to current user Without user=request.user on the PATCH filter, the ORM could match another user's preference record in the same workspace, causing pin/unpin state to leak across users or silently fail to persist. Fixes #9260 Signed-off-by: okxint <cashmein.eth@gmail.com> * test: add regression coverage for workspace user preference scoping (#9260) Adds contract tests for the sidebar preference PATCH endpoint: - test_patch_only_updates_requesting_users_preference: in a multi-member workspace, a member's PATCH must update only their own preference row, never another member's. Fails against the pre-fix code (the unscoped .first() mutates the most-recently-created row regardless of user). - test_patch_updates_own_preference: baseline that a member's PATCH persists to their own row. Verified RED on the unpatched view and GREEN with the user=request.user filter from #9261. * fix(api): wrap long line to satisfy ruff E501 in user preference view --------- Signed-off-by: okxint <cashmein.eth@gmail.com> Co-authored-by: okxint <cashmein.eth@gmail.com>
103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
# Copyright (c) 2023-present Plane Software, Inc. and contributors
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# See the LICENSE file for details.
|
|
|
|
# Module imports
|
|
from ..base import BaseAPIView
|
|
from plane.db.models.workspace import WorkspaceUserPreference
|
|
from plane.app.serializers.workspace import WorkspaceUserPreferenceSerializer
|
|
from plane.app.permissions import allow_permission, ROLE
|
|
from plane.db.models import Workspace
|
|
|
|
|
|
# Third party imports
|
|
from rest_framework.response import Response
|
|
from rest_framework import status
|
|
|
|
|
|
class WorkspaceUserPreferenceViewSet(BaseAPIView):
|
|
model = WorkspaceUserPreference
|
|
use_read_replica = True
|
|
|
|
def get_serializer_class(self):
|
|
return WorkspaceUserPreferenceSerializer
|
|
|
|
@allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST], level="WORKSPACE")
|
|
def get(self, request, slug):
|
|
workspace = Workspace.objects.get(slug=slug)
|
|
|
|
get_preference = WorkspaceUserPreference.objects.filter(user=request.user, workspace_id=workspace.id)
|
|
|
|
create_preference_keys = []
|
|
|
|
keys = [key for key, _ in WorkspaceUserPreference.UserPreferenceKeys.choices]
|
|
|
|
for preference in keys:
|
|
if preference not in get_preference.values_list("key", flat=True):
|
|
create_preference_keys.append(preference)
|
|
|
|
preference = WorkspaceUserPreference.objects.bulk_create(
|
|
[
|
|
WorkspaceUserPreference(
|
|
key=key,
|
|
user=request.user,
|
|
workspace=workspace,
|
|
sort_order=(65535 + (i * 10000)),
|
|
is_pinned=(
|
|
True
|
|
if key
|
|
in [
|
|
WorkspaceUserPreference.UserPreferenceKeys.DRAFTS,
|
|
WorkspaceUserPreference.UserPreferenceKeys.YOUR_WORK,
|
|
WorkspaceUserPreference.UserPreferenceKeys.STICKIES,
|
|
]
|
|
else False
|
|
),
|
|
)
|
|
for i, key in enumerate(create_preference_keys)
|
|
],
|
|
batch_size=10,
|
|
ignore_conflicts=True,
|
|
)
|
|
|
|
preferences = (
|
|
WorkspaceUserPreference.objects.filter(user=request.user, workspace_id=workspace.id)
|
|
.order_by("sort_order")
|
|
.values("key", "is_pinned", "sort_order")
|
|
)
|
|
|
|
user_preferences = {}
|
|
|
|
for preference in preferences:
|
|
user_preferences[(str(preference["key"]))] = {
|
|
"is_pinned": preference["is_pinned"],
|
|
"sort_order": preference["sort_order"],
|
|
}
|
|
return Response(
|
|
user_preferences,
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
|
|
@allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST], level="WORKSPACE")
|
|
def patch(self, request, slug):
|
|
for data in request.data:
|
|
key = data.pop("key", None)
|
|
if not key:
|
|
continue
|
|
|
|
preference = WorkspaceUserPreference.objects.filter(
|
|
key=key, workspace__slug=slug, user=request.user
|
|
).first()
|
|
|
|
if not preference:
|
|
continue
|
|
|
|
if "is_pinned" in data:
|
|
preference.is_pinned = data["is_pinned"]
|
|
|
|
if "sort_order" in data:
|
|
preference.sort_order = data["sort_order"]
|
|
|
|
preference.save(update_fields=["is_pinned", "sort_order"])
|
|
|
|
return Response({"message": "Successfully updated"}, status=status.HTTP_200_OK)
|