Coverage for backend/django/core/auxiliary/viewsets/FlowsheetViewSet.py: 97%

1from rest_framework import serializers, status, viewsets 

2from rest_framework.response import Response 

3from django.db.models import QuerySet 

4from core.auxiliary.models.Flowsheet import Flowsheet 

5from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

6from core.auxiliary.serializers.FlowsheetSerializer import FlowsheetSerializer 

7from drf_spectacular.utils import extend_schema 

8from rest_framework.decorators import action 

9from drf_spectacular.types import OpenApiTypes 

10from drf_spectacular.utils import OpenApiParameter 

11from authentication.user.AccessTable import AccessTable 

12from authentication.user.models import User 

13import core.auxiliary.enums.ViewType as ViewType 

14 

15 

16class ImportFlowsheetSerializer(serializers.Serializer): 

17 data = serializers.JSONField() 

18 

19 

20class FlowsheetSharingSerializer(serializers.Serializer): 

21 flowsheet = serializers.IntegerField() 

22 user_email = serializers.EmailField() 

23 read_only = serializers.BooleanField(default=False) 

24 

25 

26class RemoveSharedUserSerializer(serializers.Serializer): 

27 flowsheet = serializers.IntegerField() 

28 user_email = serializers.EmailField() 

29 

30 

31class UpdateSharedUserAccessSerializer(serializers.Serializer): 

32 flowsheet = serializers.IntegerField() 

33 user_email = serializers.EmailField() 

34 read_only = serializers.BooleanField() 

35 

36 

37class SharedUserSerializer(serializers.Serializer): 

38 email = serializers.EmailField() 

39 read_only = serializers.BooleanField() 

40 

41 

42class ListSharedUsersRequestSerializer(serializers.Serializer): 

43 flowsheet = serializers.IntegerField() 

44 

45 

46class ListSharedUsersSerializer(serializers.Serializer): 

47 users = SharedUserSerializer(many=True) 

48 

49 

50class FlowsheetViewSet(viewsets.ModelViewSet): 

51 """ 

52 Viewset for the Flowsheet model itself. 

53 

54 Flowsheet does not use AccessControlManager, so its access rules stay here 

55 and in the serializer rather than in the generic flowsheet-scoped manager. 

56 """ 

57 serializer_class = FlowsheetSerializer 

58 

59 @staticmethod 

60 def _with_current_user_access(queryset: QuerySet[Flowsheet], user) -> QuerySet[Flowsheet]: 

61 return Flowsheet.with_user_access(queryset, user) 

62 

63 def get_queryset(self) -> QuerySet[Flowsheet]: 

64 current_user = self.request.user 

65 queryset = (Flowsheet.objects 

66 .filter(owner=current_user, flowsheet_template_type=FlowsheetTemplateType.NotTemplate) 

67 .select_related('owner')) 

68 return self._with_current_user_access(queryset, current_user) 

69 

70 @extend_schema( 

71 parameters=[ 

72 OpenApiParameter(name="type", required=True, type=OpenApiTypes.STR), 

73 ] 

74 ) 

75 def list(self, request, *args, **kwargs): 

76 type = request.query_params.get('type', ViewType.OWNED) 

77 queryset = Flowsheet.get_flowsheets_by_view_type(request.user, type) 

78 queryset = (queryset 

79 .filter(flowsheet_template_type=FlowsheetTemplateType.NotTemplate) 

80 .select_related('owner')) 

81 queryset = self._with_current_user_access(queryset, request.user) 

82 serializer = self.get_serializer(queryset, many=True) 

83 return Response(serializer.data) 

84 

85 def retrieve(self, request, *args, **kwargs): 

86 try: 

87 instance = ( 

88 self._with_current_user_access( 

89 Flowsheet.objects.select_related("owner"), 

90 request.user, 

91 ) 

92 .get(id=kwargs["pk"]) 

93 ) 

94 except Flowsheet.DoesNotExist: 

95 return Response({"error": "Flowsheet not found"}, status=status.HTTP_404_NOT_FOUND) 

96 

97 user_access_entries = getattr(instance, "current_user_access_entries", []) 

98 has_direct_access = request.user.id == instance.owner_id or bool(user_access_entries) 

99 

100 if has_direct_access: 

101 # Update the savedDate field 

102 instance.set_saved_date() 

103 serializer = self.get_serializer(instance) 

104 

105 return Response(serializer.data) 

106 elif instance.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate and request.user.is_staff: 

107 # Allow admins to view contents of public templates 

108 serializer = self.get_serializer(instance) 

109 

110 return Response(serializer.data) 

111 else: 

112 return Response({"error": "You do not have access to this flowsheet"}, status=status.HTTP_403_FORBIDDEN) 

113 

114 def create(self, request, *args, **kwargs): 

115 flowsheet_serializer = FlowsheetSerializer( 

116 data=request.data, context={"request": request}) 

117 flowsheet_serializer.is_valid(raise_exception=True) 

118 flowsheet_serializer.save() 

119 

120 return Response(flowsheet_serializer.data, status=status.HTTP_201_CREATED) 

121 

122 def _is_read_only_sharee(self, user, flowsheet_id: int) -> bool: 

123 """ 

124 Lightweight helper for the flowsheet record itself, which is not filtered 

125 through AccessControlManager like flowsheet-owned child models are. 

126 """ 

127 access = AccessTable.objects.filter( 

128 user=user, 

129 flowsheet_id=flowsheet_id, 

130 ).values("read_only").first() 

131 return bool(access and access["read_only"]) 

132 

133 def update(self, request, *args, **kwargs): 

134 if self._is_read_only_sharee(request.user, kwargs["pk"]): 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 return Response( 

136 {"error": "This flowsheet is shared with read-only access."}, 

137 status=status.HTTP_403_FORBIDDEN, 

138 ) 

139 return super().update(request, *args, **kwargs) 

140 

141 def partial_update(self, request, *args, **kwargs): 

142 if self._is_read_only_sharee(request.user, kwargs["pk"]): 

143 return Response( 

144 {"error": "This flowsheet is shared with read-only access."}, 

145 status=status.HTTP_403_FORBIDDEN, 

146 ) 

147 return super().partial_update(request, *args, **kwargs) 

148 

149 def destroy(self, request, *args, **kwargs): 

150 if self._is_read_only_sharee(request.user, kwargs["pk"]): 

151 return Response( 

152 {"error": "This flowsheet is shared with read-only access."}, 

153 status=status.HTTP_403_FORBIDDEN, 

154 ) 

155 return super().destroy(request, *args, **kwargs) 

156 

157 @extend_schema(request=FlowsheetSharingSerializer, responses=None) 

158 @action(detail=False, methods=['post'], url_path='share-flowsheet') 

159 def share_flowsheet(self, request): 

160 serializer = FlowsheetSharingSerializer(data=request.data) 

161 serializer.is_valid(raise_exception=True) 

162 flowsheet = serializer.validated_data['flowsheet'] 

163 user_email = serializer.validated_data['user_email'] 

164 read_only = serializer.validated_data['read_only'] 

165 

166 try: 

167 Flowsheet.share_flowsheet(request.user, flowsheet, user_email, read_only=read_only) 

168 return Response({"message": "Flowsheet shared successfully"}, status=status.HTTP_200_OK) 

169 except ValueError as exc: 

170 return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) 

171 except PermissionError as exc: 

172 return Response({"error": str(exc)}, status=status.HTTP_403_FORBIDDEN) 

173 except (Flowsheet.DoesNotExist, User.DoesNotExist): 

174 return Response({"error": "Failed to share flowsheet"}, status=status.HTTP_404_NOT_FOUND) 

175 

176 @extend_schema(responses=ListSharedUsersSerializer) 

177 @action(detail=False, methods=['get'], url_path='list-shared-users') 

178 def list_shared_users(self, request, *args, **kwargs): 

179 query_serializer = ListSharedUsersRequestSerializer(data=request.query_params) 

180 query_serializer.is_valid(raise_exception=True) 

181 flowsheet_id = query_serializer.validated_data["flowsheet"] 

182 

183 try: 

184 users = Flowsheet.get_shared_users(request.user, flowsheet_id) 

185 serializer = ListSharedUsersSerializer({"users": users}) 

186 return Response(serializer.data, status=status.HTTP_200_OK) 

187 except Flowsheet.DoesNotExist: 

188 return Response({"error": "Failed to retrieve list of user"}, status=status.HTTP_404_NOT_FOUND) 

189 

190 @extend_schema(request=RemoveSharedUserSerializer, responses=None) 

191 @action(detail=False, methods=['post'], url_path='remove-user') 

192 def remove_user(self, request, *args, **kwargs): 

193 serializer = RemoveSharedUserSerializer(data=request.data) 

194 serializer.is_valid(raise_exception=True) 

195 flowsheet_id = serializer.validated_data["flowsheet"] 

196 user_email = serializer.validated_data["user_email"] 

197 

198 try: 

199 Flowsheet.remove_user(request.user, flowsheet_id, user_email) 

200 return Response({"message": "User removed successfully"}, status=status.HTTP_200_OK) 

201 except ValueError as exc: 

202 return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) 

203 except (Flowsheet.DoesNotExist, User.DoesNotExist, AccessTable.DoesNotExist): 

204 return Response({"error": "Failed to remove user"}, status=status.HTTP_404_NOT_FOUND) 

205 

206 @extend_schema(request=UpdateSharedUserAccessSerializer, responses=None) 

207 @action(detail=False, methods=['post'], url_path='update-shared-user-access') 

208 def update_shared_user_access(self, request, *args, **kwargs): 

209 serializer = UpdateSharedUserAccessSerializer(data=request.data) 

210 serializer.is_valid(raise_exception=True) 

211 

212 flowsheet_id = serializer.validated_data["flowsheet"] 

213 user_email = serializer.validated_data["user_email"] 

214 read_only = serializer.validated_data["read_only"] 

215 

216 try: 

217 Flowsheet.update_shared_user_access( 

218 owner=request.user, 

219 flowsheet_id=flowsheet_id, 

220 user_email=user_email, 

221 read_only=read_only, 

222 ) 

223 return Response({"message": "Shared user access updated successfully"}, status=status.HTTP_200_OK) 

224 except ValueError as exc: 

225 return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) 

226 except (Flowsheet.DoesNotExist, User.DoesNotExist, AccessTable.DoesNotExist): 

227 return Response({"error": "Failed to update shared user access"}, status=status.HTTP_404_NOT_FOUND)