15166388190

Committed 21 May 2025 03:32PM UTC coverage: 92.47% (-0.3%) from 92.77%

Build # 15166388190

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #27 from Brunowar12/bugfix-and-pep

Bug fixes and code style cleanup

Run Details

159 of 183 new or added lines in 24 files covered. (86.89%)

10 existing lines in 6 files now uncovered.

1621 of 1753 relevant lines covered (92.47%)

5.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.11

/projects/views.py

import logging
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.db import transaction
from django.db.models import Q, Count
from django.shortcuts import get_object_or_404
from django.utils import timezone
from rest_framework import viewsets, status
from rest_framework.decorators import action, api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response

from api.mixins import UserQuerysetMixin
from api.utils import error_response, status_response

from .models import Project, ProjectMembership, Role, ProjectShareLink
from .serializers import (
    KickUserSerializer, ProjectSerializer, ProjectShareLinkSerializer,
    RoleSerializer, ProjectMembershipSerializer, ShareLinkCreateSerializer
)
from .services import (
    ProjectService, ProjectMembershipService, ProjectShareLinkService,
)
from .permissions import IsProjectAdmin, IsProjectMinRole

logger = logging.getLogger(__name__)
User = get_user_model()


class ProjectViewSet(UserQuerysetMixin, viewsets.ModelViewSet):
    """
    ViewSet for operations with projects
    
    Allows you to view, create, edit, and delete projects
    Includes an additional method for getting tasks in a project
    """
    serializer_class = ProjectSerializer
    queryset = Project.objects.all().prefetch_related('tasks')
    permission_classes = [IsAuthenticated]

    ACTION_PERMISSIONS = {
        "list": ["Viewer"],
        "retrieve": ["Viewer"],
        "update": ["Member"],
        "partial_update": ["Member"],
        "assign_role": ["Moderator"],
        "generate_share_link": ["Moderator"],
        "delete_share_link": ["Moderator"],
        "kick": ["Admin"],
        "destroy": ["Admin"],
        "leave_project": ["Viewer"],
    }

    def get_permissions(self):
        perms = [IsAuthenticated()]
        min_role = self.ACTION_PERMISSIONS.get(self.action)
        if min_role:
            perms.append(
                IsProjectAdmin()
                if min_role == "Admin"
                else IsProjectMinRole(min_role)
            )
        return perms

    def get_queryset(self):
        user = self.request.user
        return (
            self.queryset
            .filter(Q(owner=user) | Q(memberships__user=user))
            .annotate(tasks_count=Count("tasks"))
            .distinct()
            .order_by('id')
        )

    def perform_create(self, serializer):
        project = ProjectService.create_project(
            owner=self.request.user, **serializer.validated_data
        )
        serializer.instance = project

    #
    # === HELPERS ===
    #

    def _get_project(self, pk=None):
        return ProjectService.get_project_or_404(
            pk or self.kwargs["pk"], self.request.user
        )

    def _get_membership(self, project, user):
        return ProjectMembership.objects.filter(
            project=project, user=user
        ).first()

    def _forbidden(self, msg):
        return error_response(msg, status.HTTP_403_FORBIDDEN)

    #
    # === ACTIONS ===
    #

    @action(detail=True, methods=["post"])
    def assign_role(self, request, pk=None):
        """        
        Assign a role to a user in a project.
        Prevents self-assignment, assigning to owner,
        or assigning role ≥ assigner's role (except Admin/Owner).
        """
        project = self._get_project(pk)
        target = get_object_or_404(User, id=request.data.get("user_id"))
        new_role = get_object_or_404(Role, id=request.data.get("role_id"))

        if target == project.owner:
            return error_response("Cannot assign role to the project owner")
        if target == request.user:
            return self._forbidden("You cannot change your own role")

        if not self._get_membership(project, target):
            return error_response("User must join via ShareLink")

        assigner = self._get_membership(project, request.user)
        assigner_role = assigner.role.name if assigner else "Owner"    

        hierarchy = settings.ROLE_ORDER        
        if assigner_role not in ("Admin", "Owner"):
            if hierarchy.index(new_role.name) >= hierarchy.index(assigner_role):
                return error_response(
                    f"Cannot assign role '{new_role.name}'",
                    status.HTTP_403_FORBIDDEN,
                )

        try:
            ProjectMembershipService.assign_role(project, target, new_role)
        except ValidationError as e:
            return error_response(e.messages[0], exc=e)
        except Exception as e:
            message = getattr(e, 'detail', str(e))
            return error_response(message, exc=e)

        return status_response("Role assigned")

    @action(
        detail=True, methods=["post"], url_path="kick",
        serializer_class=KickUserSerializer
    )
    def kick(self, request, pk=None):
        """        
        Kick user from project (not owner)
        """
        project = self._get_project(pk)
        user_id = request.data.get("user_id")
        membership = get_object_or_404(
            ProjectMembership, project=project, user__id=user_id
        )
        if membership.user == project.owner:
            return error_response("Cannot kick project owner")

        membership.delete()
        return status_response("Member excluded")

    @action(detail=True, methods=["post"], url_path="leave")
    def leave_project(self, request, pk=None):
        """
        Leave project (not owner)
        """
        project = self._get_project(pk)
        if project.owner == request.user:
            return self._forbidden("Owner cannot leave project")

        membership = self._get_membership(project, request.user)
        if not membership:
            return error_response("Not a member of this project")

        membership.delete()
        return status_response("You left the project")


class RoleViewSet(viewsets.ReadOnlyModelViewSet):
    """
    Read‑only endpoints for project roles.
    
    All roles are created and managed via migrations/signals,
    not via API. This ViewSet only allows listing and retrieving.
    """
    queryset = Role.objects.all()
    serializer_class = RoleSerializer
    permission_classes = [IsAuthenticated]


class ProjectMembershipViewSet(viewsets.ReadOnlyModelViewSet):
    """
    Read-only viewset for viewing project members
    """
    serializer_class = ProjectMembershipSerializer
    permission_classes = [IsAuthenticated]

    def get_queryset(self):
        user = self.request.user
        projects = Project.objects.filter(
            Q(owner=user) | Q(memberships__user=user)
        ).distinct()

        return (
            ProjectMembership.objects.filter(project__in=projects)
            .select_related("user", "role")
            .order_by("id")
        )


class ProjectShareLinkViewSet(viewsets.ModelViewSet):
    """
    Read-only viewset for viewing project share links
    """

    lookup_field = "id"
    permission_classes = [IsAuthenticated]

    def get_permissions(self):
        perms = super().get_permissions()
        if self.action in ("list", "retrieve", "create", "destroy"):
            perms.append(IsProjectMinRole("Moderator"))
        return perms

    def get_queryset(self):
        project = ProjectService.get_project_or_404(
            self.kwargs["project_pk"], self.request.user
        )
        return ProjectShareLink.objects.filter(project=project)

    def get_serializer_class(self):
        if self.action == "create":
            return ShareLinkCreateSerializer
        return ProjectShareLinkSerializer

    def create(self, request, *args, **kwargs):
        project = ProjectService.get_project_or_404(
            self.kwargs["project_pk"], request.user
        )

        if ProjectShareLink.objects.filter(
            project=project, is_active=True, expires_at__gt=timezone.now()
        ).exists():
            return error_response(
                "An active share link already exists for this project"
            )

        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        data = serializer.validated_data

        share_link = ProjectShareLinkService.create_share_link(
            project=project,
            role_id=data["role_id"],
            user=request.user,
            max_uses=data.get("max_uses"),
            expires_in=data["expires_in"],
        )
        output = ProjectShareLinkSerializer(
            share_link, context={"request": request}
        )
        return Response(output.data, status=status.HTTP_201_CREATED)

    def destroy(self, request, *args, **kwargs):
        project = ProjectService.get_project_or_404(
            self.kwargs["project_pk"], request.user
        )
        share_link = get_object_or_404(
            ProjectShareLink, id=self.kwargs["id"], project=project
        )
        share_link.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)


@api_view(["POST"])
@permission_classes([IsAuthenticated])
def join_project(request, token):
    with transaction.atomic():
        link = get_object_or_404(
            ProjectShareLink.objects.select_for_update(), token=token
        )
        ProjectShareLinkService.validate_share_link(link)

        already_member = ProjectMembership.objects.filter(
            project=link.project, user=request.user
        ).exists()

        if already_member:
            return status_response(
                "Already a member of this project", status.HTTP_200_OK
            )

        ProjectMembership.objects.create(
            user=request.user, project=link.project, role=link.role
        )
        link.used_count += 1
        link.save(update_fields=["used_count"])

        return status_response(
            "Successfully joined the project", status.HTTP_200_OK
        )

1	import logging	6✔
2	from django.conf import settings	6✔
3	from django.contrib.auth import get_user_model	6✔
4	from django.core.exceptions import ValidationError	6✔
5	from django.db import transaction	6✔
6	from django.db.models import Q, Count	6✔
7	from django.shortcuts import get_object_or_404	6✔
8	from django.utils import timezone	6✔
9	from rest_framework import viewsets, status	6✔
10	from rest_framework.decorators import action, api_view, permission_classes	6✔
11	from rest_framework.permissions import IsAuthenticated	6✔
12	from rest_framework.response import Response	6✔
13
14	from api.mixins import UserQuerysetMixin	6✔
15	from api.utils import error_response, status_response	6✔
16
17	from .models import Project, ProjectMembership, Role, ProjectShareLink	6✔
18	from .serializers import (	6✔
19	KickUserSerializer, ProjectSerializer, ProjectShareLinkSerializer,
20	RoleSerializer, ProjectMembershipSerializer, ShareLinkCreateSerializer
21	)
22	from .services import (	6✔
23	ProjectService, ProjectMembershipService, ProjectShareLinkService,
24	)
25	from .permissions import IsProjectAdmin, IsProjectMinRole	6✔
26
27	logger = logging.getLogger(__name__)	6✔
28	User = get_user_model()	6✔
29
30
31	class ProjectViewSet(UserQuerysetMixin, viewsets.ModelViewSet):	6✔
32	"""
33	ViewSet for operations with projects
34
35	Allows you to view, create, edit, and delete projects
36	Includes an additional method for getting tasks in a project
37	"""
38	serializer_class = ProjectSerializer	6✔
39	queryset = Project.objects.all().prefetch_related('tasks')	6✔
40	permission_classes = [IsAuthenticated]	6✔
41
42	ACTION_PERMISSIONS = {	6✔
43	"list": ["Viewer"],
44	"retrieve": ["Viewer"],
45	"update": ["Member"],
46	"partial_update": ["Member"],
47	"assign_role": ["Moderator"],
48	"generate_share_link": ["Moderator"],
49	"delete_share_link": ["Moderator"],
50	"kick": ["Admin"],
51	"destroy": ["Admin"],
52	"leave_project": ["Viewer"],
53	}
54
55	def get_permissions(self):	6✔
56	perms = [IsAuthenticated()]	6✔
57	min_role = self.ACTION_PERMISSIONS.get(self.action)	6✔
58	if min_role:	6✔
59	perms.append(	6✔
60	IsProjectAdmin()
61	if min_role == "Admin"
62	else IsProjectMinRole(min_role)
63	)
64	return perms	6✔
65
66	def get_queryset(self):	6✔
67	user = self.request.user	6✔
68	return (	6✔
69	self.queryset
70	.filter(Q(owner=user) \| Q(memberships__user=user))
71	.annotate(tasks_count=Count("tasks"))
72	.distinct()
73	.order_by('id')
74	)
75
76	def perform_create(self, serializer):	6✔
77	project = ProjectService.create_project(	6✔
78	owner=self.request.user, **serializer.validated_data
79	)
80	serializer.instance = project	6✔
81
82	#
83	# === HELPERS ===
84	#
85
86	def _get_project(self, pk=None):	6✔
87	return ProjectService.get_project_or_404(	6✔
88	pk or self.kwargs["pk"], self.request.user
89	)
90
91	def _get_membership(self, project, user):	6✔
92	return ProjectMembership.objects.filter(	6✔
93	project=project, user=user
94	).first()
95
96	def _forbidden(self, msg):	6✔
97	return error_response(msg, status.HTTP_403_FORBIDDEN)	×
98
99	#
100	# === ACTIONS ===
101	#
102
103	@action(detail=True, methods=["post"])	6✔
104	def assign_role(self, request, pk=None):	6✔
105	"""
106	Assign a role to a user in a project.
107	Prevents self-assignment, assigning to owner,
108	or assigning role ≥ assigner's role (except Admin/Owner).
109	"""
110	project = self._get_project(pk)	6✔
111	target = get_object_or_404(User, id=request.data.get("user_id"))	6✔
112	new_role = get_object_or_404(Role, id=request.data.get("role_id"))	6✔
113
114	if target == project.owner:	6✔
115	return error_response("Cannot assign role to the project owner")	6✔
116	if target == request.user:	6✔
UNCOV 117	return self._forbidden("You cannot change your own role")	×
118
119	if not self._get_membership(project, target):	6✔
120	return error_response("User must join via ShareLink")	×
121
122	assigner = self._get_membership(project, request.user)	6✔
123	assigner_role = assigner.role.name if assigner else "Owner"	6✔
124
125	hierarchy = settings.ROLE_ORDER	6✔
126	if assigner_role not in ("Admin", "Owner"):	6✔
NEW 127	if hierarchy.index(new_role.name) >= hierarchy.index(assigner_role):	×
NEW 128	return error_response(	×
129	f"Cannot assign role '{new_role.name}'",
130	status.HTTP_403_FORBIDDEN,
131	)
132
133	try:	6✔
134	ProjectMembershipService.assign_role(project, target, new_role)	6✔
135	except ValidationError as e:	6✔
136	return error_response(e.messages[0], exc=e)	6✔
UNCOV 137	except Exception as e:	×
UNCOV 138	message = getattr(e, 'detail', str(e))	×
UNCOV 139	return error_response(message, exc=e)	×
140
141	return status_response("Role assigned")	6✔
142
143	@action(	6✔
144	detail=True, methods=["post"], url_path="kick",
145	serializer_class=KickUserSerializer
146	)
147	def kick(self, request, pk=None):	6✔
148	"""
149	Kick user from project (not owner)
150	"""
151	project = self._get_project(pk)	6✔
152	user_id = request.data.get("user_id")	6✔
153	membership = get_object_or_404(	6✔
154	ProjectMembership, project=project, user__id=user_id
155	)
156	if membership.user == project.owner:	6✔
157	return error_response("Cannot kick project owner")	×
158
159	membership.delete()	6✔
160	return status_response("Member excluded")	6✔
161
162	@action(detail=True, methods=["post"], url_path="leave")	6✔
163	def leave_project(self, request, pk=None):	6✔
164	"""
165	Leave project (not owner)
166	"""
167	project = self._get_project(pk)	×
168	if project.owner == request.user:	×
169	return self._forbidden("Owner cannot leave project")	×
170
171	membership = self._get_membership(project, request.user)	×
172	if not membership:	×
173	return error_response("Not a member of this project")	×
174
175	membership.delete()	×
176	return status_response("You left the project")	×
177
178
179	class RoleViewSet(viewsets.ReadOnlyModelViewSet):	6✔
180	"""
181	Read‑only endpoints for project roles.
182
183	All roles are created and managed via migrations/signals,
184	not via API. This ViewSet only allows listing and retrieving.
185	"""
186	queryset = Role.objects.all()	6✔
187	serializer_class = RoleSerializer	6✔
188	permission_classes = [IsAuthenticated]	6✔
189
190
191	class ProjectMembershipViewSet(viewsets.ReadOnlyModelViewSet):	6✔
192	"""
193	Read-only viewset for viewing project members
194	"""
195	serializer_class = ProjectMembershipSerializer	6✔
196	permission_classes = [IsAuthenticated]	6✔
197
198	def get_queryset(self):	6✔
199	user = self.request.user	6✔
200	projects = Project.objects.filter(	6✔
201	Q(owner=user) \| Q(memberships__user=user)
202	).distinct()
203
204	return (	6✔
205	ProjectMembership.objects.filter(project__in=projects)
206	.select_related("user", "role")
207	.order_by("id")
208	)
209
210
211	class ProjectShareLinkViewSet(viewsets.ModelViewSet):	6✔
212	"""
213	Read-only viewset for viewing project share links
214	"""
215
216	lookup_field = "id"	6✔
217	permission_classes = [IsAuthenticated]	6✔
218
219	def get_permissions(self):	6✔
220	perms = super().get_permissions()	6✔
221	if self.action in ("list", "retrieve", "create", "destroy"):	6✔
222	perms.append(IsProjectMinRole("Moderator"))	6✔
223	return perms	6✔
224
225	def get_queryset(self):	6✔
226	project = ProjectService.get_project_or_404(	×
227	self.kwargs["project_pk"], self.request.user
228	)
229	return ProjectShareLink.objects.filter(project=project)	×
230
231	def get_serializer_class(self):	6✔
232	if self.action == "create":	6✔
233	return ShareLinkCreateSerializer	6✔
234	return ProjectShareLinkSerializer	×
235
236	def create(self, request, args, *kwargs):	6✔
237	project = ProjectService.get_project_or_404(	6✔
238	self.kwargs["project_pk"], request.user
239	)
240
241	if ProjectShareLink.objects.filter(	6✔
242	project=project, is_active=True, expires_at__gt=timezone.now()
243	).exists():
244	return error_response(	×
245	"An active share link already exists for this project"
246	)
247
248	serializer = self.get_serializer(data=request.data)	6✔
249	serializer.is_valid(raise_exception=True)	6✔
250	data = serializer.validated_data	6✔
251
252	share_link = ProjectShareLinkService.create_share_link(	6✔
253	project=project,
254	role_id=data["role_id"],
255	user=request.user,
256	max_uses=data.get("max_uses"),
257	expires_in=data["expires_in"],
258	)
259	output = ProjectShareLinkSerializer(	6✔
260	share_link, context={"request": request}
261	)
262	return Response(output.data, status=status.HTTP_201_CREATED)	6✔
263
264	def destroy(self, request, args, *kwargs):	6✔
265	project = ProjectService.get_project_or_404(	6✔
266	self.kwargs["project_pk"], request.user
267	)
268	share_link = get_object_or_404(	6✔
269	ProjectShareLink, id=self.kwargs["id"], project=project
270	)
271	share_link.delete()	6✔
272	return Response(status=status.HTTP_204_NO_CONTENT)	6✔
273
274
275	@api_view(["POST"])	6✔
276	@permission_classes([IsAuthenticated])	6✔
277	def join_project(request, token):	6✔
278	with transaction.atomic():	6✔
279	link = get_object_or_404(	6✔
280	ProjectShareLink.objects.select_for_update(), token=token
281	)
282	ProjectShareLinkService.validate_share_link(link)	6✔
283
284	already_member = ProjectMembership.objects.filter(	6✔
285	project=link.project, user=request.user
286	).exists()
287
288	if already_member:	6✔
289	return status_response(	6✔
290	"Already a member of this project", status.HTTP_200_OK
291	)
292
293	ProjectMembership.objects.create(	6✔
294	user=request.user, project=link.project, role=link.role
295	)
296	link.used_count += 1	6✔
297	link.save(update_fields=["used_count"])	6✔
298
299	return status_response(	6✔
300	"Successfully joined the project", status.HTTP_200_OK
301	)

Brunowar12 / TaskManagerSystem / 15166388190

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous