Coverage for backend/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 

15class ImportFlowsheetSerializer(serializers.Serializer): 

16 data = serializers.JSONField() 

17 

18class FlowsheetSharingSerializer(serializers.Serializer): 

19 flowsheet = serializers.IntegerField() 

20 user_email = serializers.EmailField() 

21 

22class ListSharedUsersSerializer(serializers.Serializer): 

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

24 

25class FlowsheetViewSet(viewsets.ModelViewSet): 

26 

27 serializer_class = FlowsheetSerializer 

28 

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

30 current_user = self.request.user 

31 return Flowsheet.objects.filter(owner=current_user, flowsheet_template_type=FlowsheetTemplateType.NotTemplate) 

32 

33 @extend_schema( 

34 parameters=[ 

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

36 ] 

37 ) 

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

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

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

41 queryset = queryset.filter(flowsheet_template_type=FlowsheetTemplateType.NotTemplate) 

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

43 return Response(serializer.data) 

44 

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

46 # Update the savedDate field 

47 instance = Flowsheet.objects.get(id=kwargs['pk']) 

48 if AccessTable.objects.filter( 

49 user=request.user, 

50 flowsheet=instance 

51 ).exists(): 

52 instance.set_saved_date() 

53 serializer = self.get_serializer(instance) 

54 

55 return Response(serializer.data) 

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

57 # Allow admins to view contents of public templates 

58 serializer = self.get_serializer(instance) 

59 

60 return Response(serializer.data) 

61 else: 

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

63 

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

65 flowsheet_serializer = FlowsheetSerializer( 

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

67 flowsheet_serializer.is_valid(raise_exception=True) 

68 flowsheet_serializer.save() 

69 

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

71 

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

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

74 def share_flowsheet(self, request): 

75 serializer = FlowsheetSharingSerializer(data=request.data) 

76 serializer.is_valid(raise_exception=True) 

77 flowsheet = serializer.validated_data['flowsheet'] 

78 user_email = serializer.validated_data['user_email'] 

79 

80 try: 

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

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

83 except: 

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

85 

86 @extend_schema(responses=ListSharedUsersSerializer) 

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

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

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

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

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

92 

93 try: 

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

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

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

97 except: 

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

99 

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

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

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

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

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

105 

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

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

108 

109 try: 

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

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

112 except: 

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