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

1from rest_framework import serializers, viewsets 

2from rest_framework import status 

3from rest_framework.response import Response 

4from django.db.models import QuerySet 

5from core.auxiliary.models.Flowsheet import Flowsheet 

6from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

7from core.auxiliary.serializers.FlowsheetSerializer import FlowsheetSerializer 

8from drf_spectacular.utils import extend_schema 

9from rest_framework.decorators import action 

10from drf_spectacular.types import OpenApiTypes 

11from drf_spectacular.utils import OpenApiParameter 

12from authentication.user.AccessTable import AccessTable 

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 

24 

25class ListSharedUsersSerializer(serializers.Serializer): 

26 users = serializers.ListField(child=serializers.EmailField()) 

27 

28 

29class FlowsheetViewSet(viewsets.ModelViewSet): 

30 serializer_class = FlowsheetSerializer 

31 

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

33 current_user = self.request.user 

34 return (Flowsheet.objects 

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

36 .select_related('owner')) 

37 

38 @extend_schema( 

39 parameters=[ 

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

41 ] 

42 ) 

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

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

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

46 queryset = (queryset 

47 .filter(flowsheet_template_type=FlowsheetTemplateType.NotTemplate) 

48 .select_related('owner')) 

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

50 return Response(serializer.data) 

51 

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

53 # Update the savedDate field 

54 instance = Flowsheet.objects.select_related('owner').get(id=kwargs['pk']) 

55 if AccessTable.objects.filter( 

56 user=request.user, 

57 flowsheet=instance 

58 ).exists(): 

59 instance.set_saved_date() 

60 serializer = self.get_serializer(instance) 

61 

62 return Response(serializer.data) 

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

64 # Allow admins to view contents of public templates 

65 serializer = self.get_serializer(instance) 

66 

67 return Response(serializer.data) 

68 else: 

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

70 

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

72 flowsheet_serializer = FlowsheetSerializer( 

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

74 flowsheet_serializer.is_valid(raise_exception=True) 

75 flowsheet_serializer.save() 

76 

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

78 

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

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

81 def share_flowsheet(self, request): 

82 serializer = FlowsheetSharingSerializer(data=request.data) 

83 serializer.is_valid(raise_exception=True) 

84 flowsheet = serializer.validated_data['flowsheet'] 

85 user_email = serializer.validated_data['user_email'] 

86 

87 try: 

88 Flowsheet.share_flowsheet(request.user, flowsheet, user_email) 

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

90 except: 

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

92 

93 @extend_schema(responses=ListSharedUsersSerializer) 

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

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

96 flowsheet_id = request.query_params.get("flowsheet") 

97 if not flowsheet_id: 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true

98 return Response({"error": "Flowsheet ID is required"}, status=status.HTTP_400_BAD_REQUEST) 

99 

100 try: 

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

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

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

104 except: 

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

106 

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

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

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

110 flowsheet_id = request.data.get("flowsheet") 

111 user_email = request.data.get("user_email") 

112 

113 if not flowsheet_id or not user_email: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true

114 return Response({"error": "Flowsheet ID and user email are required"}, status=status.HTTP_400_BAD_REQUEST) 

115 

116 try: 

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

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

119 except: 

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