Coverage for backend/django/core/auxiliary/viewsets/FlowsheetTemplateViewSet.py: 87%

1from django.db import transaction 

2from django.db.models import Q 

3from rest_framework import status, viewsets 

4from rest_framework.decorators import action 

5from rest_framework.response import Response 

6from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

7from core.auxiliary.models.Flowsheet import Flowsheet 

8from core.auxiliary.serializers.FlowsheetSerializer import FlowsheetSerializer 

9from core.auxiliary.serializers.FlowsheetCloneValidationSerializer import ( 

10 FlowsheetCloneValidationErrorSerializer, 

11) 

12from core.auxiliary.services.flowsheet_template_transitions import ( 

13 FlowsheetNotTemplateError, 

14 PublicTemplateRevertPermissionError, 

15 convert_flowsheet_to_template, 

16 revert_flowsheet_template, 

17) 

18from flowsheetInternals.graphicData.models.groupingModel import Grouping 

19from core.auxiliary.services.flowsheet_states import ( 

20 FlowsheetStateCloneError, 

21 clone_module_subtree, 

22 clone_validation_error_data, 

23 create_project_from_template, 

24) 

25from flowsheetInternals.graphicData.serializers.groupingSerializer import GroupingSerializer 

26from drf_spectacular.utils import extend_schema 

27from rest_framework import serializers 

28from core.managers import get_flowsheet_access 

29from core.viewset import HistoricalRevisionWriteGuardMixin 

30 

31 

32class CreateFlowsheetTemplateSerializer(serializers.Serializer): 

33 flowsheet_id = serializers.IntegerField() 

34 flowsheet_template_type = serializers.ChoiceField(choices=FlowsheetTemplateType.choices) 

35 

36 

37class CreateModuleFromTemplateSerializer(serializers.Serializer): 

38 current_group = serializers.IntegerField(required=False, allow_null=True) 

39 flowsheet = serializers.IntegerField() 

40 x = serializers.FloatField(required=False, allow_null=True) 

41 y = serializers.FloatField(required=False, allow_null=True) 

42 

43 

44class FlowsheetTemplateViewSet( 

45 HistoricalRevisionWriteGuardMixin, 

46 viewsets.ReadOnlyModelViewSet, 

47): 

48 serializer_class = FlowsheetSerializer 

49 

50 def get_queryset(self): 

51 # Public templates are visible to all users 

52 # Private templates only to the owner 

53 flowsheets = Flowsheet.objects.select_related('owner', 'project') 

54 

55 public_flowsheet_templates = flowsheets.filter( 

56 flowsheet_template_type=FlowsheetTemplateType.PublicTemplate) 

57 private_flowsheet_templates = flowsheets.filter( 

58 flowsheet_template_type=FlowsheetTemplateType.PrivateTemplate, 

59 ).filter( 

60 Q(project__owner=self.request.user) 

61 | Q(project__isnull=True, owner=self.request.user) 

62 ) 

63 return public_flowsheet_templates | private_flowsheet_templates 

64 

65 @extend_schema( 

66 request=CreateFlowsheetTemplateSerializer, 

67 responses={ 

68 200: FlowsheetSerializer, 

69 201: FlowsheetSerializer, 

70 422: FlowsheetCloneValidationErrorSerializer, 

71 }, 

72 ) 

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

74 def create_flowsheet_template(self, request) -> Response: 

75 """Create a template copy or change an existing template's visibility.""" 

76 serializer = CreateFlowsheetTemplateSerializer(data=request.data) 

77 serializer.is_valid(raise_exception=True) 

78 

79 flowsheet_id = serializer.validated_data['flowsheet_id'] 

80 flowsheet_template_type = serializer.validated_data['flowsheet_template_type'] 

81 

82 # Only allow conversion to public template if user is staff 

83 if flowsheet_template_type == FlowsheetTemplateType.PublicTemplate and not request.user.is_staff: 

84 return Response({'error': 'You do not have permission to create public templates'}, 

85 status=status.HTTP_403_FORBIDDEN) 

86 

87 try: 

88 # Project-backed flowsheets are managed by the project owner. Legacy 

89 # projectless templates still use their direct flowsheet owner. 

90 candidate_flowsheet = ( 

91 Flowsheet.objects 

92 .select_related('owner', 'project') 

93 .filter( 

94 Q(project__owner=request.user) 

95 | Q(project__isnull=True, owner=request.user) 

96 ) 

97 .get(id=flowsheet_id) 

98 ) 

99 creates_template = ( 

100 candidate_flowsheet.flowsheet_template_type 

101 == FlowsheetTemplateType.NotTemplate 

102 ) 

103 

104 with transaction.atomic(): 

105 template = convert_flowsheet_to_template( 

106 candidate=candidate_flowsheet, 

107 user_id=request.user.pk, 

108 template_type=flowsheet_template_type, 

109 ) 

110 response_data = FlowsheetSerializer( 

111 template, 

112 context={"request": request}, 

113 ).data 

114 

115 return Response( 

116 response_data, 

117 status=( 

118 status.HTTP_201_CREATED 

119 if creates_template 

120 else status.HTTP_200_OK 

121 ), 

122 ) 

123 

124 except Flowsheet.DoesNotExist: 

125 return Response({'error': 'Flowsheet not found or you do not have permission'}, 

126 status=status.HTTP_404_NOT_FOUND) 

127 except FlowsheetStateCloneError as exc: 

128 if not exc.is_user_repairable: 

129 raise 

130 return Response( 

131 clone_validation_error_data(exc, flowsheet=candidate_flowsheet), 

132 status=status.HTTP_422_UNPROCESSABLE_ENTITY, 

133 ) 

134 

135 @extend_schema(responses=None) 

136 @action(detail=True, methods=['post'], url_path='create-from-flowsheet-template') 

137 def create_from_flowsheet_template(self, request, pk=None) -> Response: 

138 """ 

139 Create a new flowsheet from a template 

140 """ 

141 try: 

142 flowsheet_template = self.get_object() 

143 

144 # copy the template to create a new flowsheet 

145 project, new_flowsheet = create_project_from_template( 

146 source_template=flowsheet_template, 

147 user=request.user, 

148 ) 

149 

150 return Response( 

151 FlowsheetSerializer(new_flowsheet, context={"request": request}).data, 

152 status=status.HTTP_201_CREATED, 

153 ) 

154 

155 except Flowsheet.DoesNotExist: 

156 return Response({'error': 'Template not found'}, status=status.HTTP_404_NOT_FOUND) 

157 except Exception as e: 

158 return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST) 

159 

160 @extend_schema(request=CreateModuleFromTemplateSerializer, responses=None) 

161 @action(detail=True, methods=['post'], url_path='create-module-from-template') 

162 def create_module_from_template(self, request, pk=None) -> Response: 

163 """ 

164 Create a new module (grouping) from a template within the current flowsheet 

165 """ 

166 data = request.data.get('createModuleFromTemplate', request.data) 

167 

168 serializer = CreateModuleFromTemplateSerializer(data=data) 

169 serializer.is_valid(raise_exception=True) 

170 

171 current_flowsheet_id = serializer.validated_data['flowsheet'] 

172 current_group_id = serializer.validated_data.get('current_group') 

173 x = serializer.validated_data.get("x") 

174 y = serializer.validated_data.get("y") 

175 

176 try: 

177 with transaction.atomic(): 

178 flowsheet_template = self.get_object() 

179 access_state = get_flowsheet_access(request.user, current_flowsheet_id) 

180 if not access_state.has_read_access: 

181 return Response({'error': 'Template or flowsheet not found'}, status=status.HTTP_404_NOT_FOUND) 

182 if not access_state.has_write_access: 

183 return Response( 

184 {'error': 'You do not have write access to this flowsheet'}, 

185 status=status.HTTP_403_FORBIDDEN, 

186 ) 

187 

188 current_flowsheet = Flowsheet.objects.get(id=current_flowsheet_id) 

189 

190 current_group: Grouping | None = None 

191 if current_group_id: 191 ↛ 195line 191 didn't jump to line 195 because the condition on line 191 was always true

192 current_group = Grouping.objects.get(id=current_group_id) 

193 

194 # Create the module from template 

195 new_module = clone_module_subtree( 

196 template_flowsheet=flowsheet_template, 

197 target_flowsheet=current_flowsheet, 

198 current_group=current_group, 

199 x=x, 

200 y=y, 

201 ) 

202 

203 return Response(GroupingSerializer(new_module).data, status=status.HTTP_201_CREATED) 

204 

205 except Flowsheet.DoesNotExist: 

206 return Response({'error': 'Template or flowsheet not found'}, status=status.HTTP_404_NOT_FOUND) 

207 except Exception as e: 

208 return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST) 

209 

210 @extend_schema(request=None, responses=FlowsheetSerializer) 

211 @action(detail=True, methods=['post'], url_path='revert-flowsheet-template') 

212 def revert_flowsheet_template(self, request, pk=None) -> Response: 

213 """ 

214 Revert a template flowsheet back to a regular (not template) flowsheet. 

215 """ 

216 try: 

217 candidate_flowsheet: Flowsheet = self.get_object() 

218 

219 with transaction.atomic(): 

220 flowsheet_template = revert_flowsheet_template( 

221 candidate=candidate_flowsheet, 

222 user_id=request.user.pk, 

223 is_staff=request.user.is_staff, 

224 ) 

225 response_data = FlowsheetSerializer( 

226 flowsheet_template, 

227 context={"request": request}, 

228 ).data 

229 

230 return Response( 

231 response_data, 

232 status=status.HTTP_200_OK, 

233 ) 

234 

235 except Flowsheet.DoesNotExist: 

236 return Response({'error': 'Template not found or you do not have permission'}, 

237 status=status.HTTP_404_NOT_FOUND) 

238 except FlowsheetNotTemplateError as exc: 

239 return Response({'error': str(exc)}, status=status.HTTP_400_BAD_REQUEST) 

240 except PublicTemplateRevertPermissionError as exc: 

241 return Response({'error': str(exc)}, status=status.HTTP_403_FORBIDDEN) 

242 except Exception as e: 

243 return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST)