Coverage for backend/django/core/auxiliary/viewsets/ScenarioViewSet.py: 92%

1from core.viewset import ModelViewSet 

2 

3from django.db import transaction 

4from drf_spectacular.utils import extend_schema, OpenApiParameter, OpenApiTypes 

5from rest_framework import status 

6from rest_framework.decorators import action 

7from rest_framework.response import Response 

8from ..serializers.ScenarioSerializer import ScenarioSerializer, OptimizationDegreesOfFreedomSerializer 

9from ..models.Scenario import ( 

10 Scenario, 

11 OptimizationDegreesOfFreedom, 

12 ScenarioInputModeEnum, 

13) 

14from ..services import parameter_sweep 

15from core.managers import get_flowsheet_access 

16 

17 

18class ScenarioViewSet(ModelViewSet): 

19 serializer_class = ScenarioSerializer 

20 

21 

22 def get_queryset(self): 

23 queryset = Scenario.objects.all() 

24 simulationObjectId = self.request.query_params.get("simulationObjectId") 

25 if simulationObjectId is not None: 

26 queryset = queryset.filter(simulationObject=simulationObjectId) 

27 return queryset 

28 

29 @extend_schema( 

30 parameters=[ 

31 OpenApiParameter(name="simulationObjectId", required=True, type=OpenApiTypes.INT), 

32 ] 

33 ) 

34 def list(self, request): 

35 return super().list(request) 

36 

37 def _validate_sweep_mode_data(self, instance: Scenario, validated_data: dict): 

38 """Reject MSS mode changes that general scenario validation cannot assess. 

39 

40 The serializer only validates the submitted fields. Parameter sweep mode 

41 changes also depend on existing sweep definitions and optimisation state, 

42 so this guard runs before update side effects can clear CSV or sweep rows. 

43 """ 

44 requested_mode = validated_data.get("mss_input_mode", instance.mss_input_mode) 

45 requested_optimization = validated_data.get( 

46 "enable_optimization", 

47 instance.enable_optimization, 

48 ) 

49 sweep_definition_exists = hasattr(instance, "parameterSweepDefinition") 

50 clearing_sweep_mode = ( 

51 requested_mode == ScenarioInputModeEnum.Csv 

52 and ( 

53 instance.mss_input_mode == ScenarioInputModeEnum.ParameterSweep 

54 or sweep_definition_exists 

55 ) 

56 ) 

57 

58 if ( 

59 requested_optimization is True 

60 and ( 

61 requested_mode == ScenarioInputModeEnum.ParameterSweep 

62 or (sweep_definition_exists and not clearing_sweep_mode) 

63 ) 

64 ): 

65 return Response( 

66 { 

67 "enable_optimization": ( 

68 "Optimisation cannot be enabled while parameter sweep data " 

69 "or a sweep definition is active." 

70 ) 

71 }, 

72 status=status.HTTP_400_BAD_REQUEST, 

73 ) 

74 

75 if ( 75 ↛ 79line 75 didn't jump to line 79 because the condition on line 75 was never true

76 requested_mode == ScenarioInputModeEnum.ParameterSweep 

77 and requested_optimization is True 

78 ): 

79 return Response( 

80 { 

81 "mss_input_mode": ( 

82 "Parameter sweep mode cannot be enabled while optimisation is active." 

83 ) 

84 }, 

85 status=status.HTTP_400_BAD_REQUEST, 

86 ) 

87 

88 return None 

89 

90 def _update_scenario(self, request, *, partial: bool): 

91 instance = self.get_object() 

92 serializer = self.get_serializer(instance, data=request.data, partial=partial) 

93 serializer.is_valid(raise_exception=True) 

94 requested_mode = serializer.validated_data.get("mss_input_mode") 

95 previous_mode = instance.mss_input_mode 

96 had_parameter_sweep_definition = hasattr(instance, "parameterSweepDefinition") 

97 

98 validation_response = self._validate_sweep_mode_data( 

99 instance, 

100 serializer.validated_data, 

101 ) 

102 if validation_response is not None: 

103 return validation_response 

104 

105 with transaction.atomic(): 

106 self.perform_update(serializer) 

107 if requested_mode is not None: 

108 parameter_sweep.clear_mss_input_data_after_mode_switch( 

109 serializer.instance, 

110 previous_mode=previous_mode, 

111 requested_mode=requested_mode, 

112 had_parameter_sweep_definition=had_parameter_sweep_definition, 

113 ) 

114 

115 if getattr(instance, "_prefetched_objects_cache", None): 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 instance._prefetched_objects_cache = {} 

117 

118 instance.refresh_from_db() 

119 return Response(self.get_serializer(instance).data) 

120 

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

122 return self._update_scenario(request, partial=True) 

123 

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

125 return self._update_scenario(request, partial=False) 

126 

127 @extend_schema( 

128 operation_id="core_scenario_eligible_parameter_sweep_targets_list", 

129 responses=parameter_sweep.EligibleParameterSweepTargetsResponse, 

130 ) 

131 @action(detail=True, methods=["get"], url_path="eligible_parameter_sweep_targets") 

132 def eligible_parameter_sweep_targets(self, request, pk=None): 

133 scenario = self.get_object() 

134 targets = parameter_sweep.eligible_parameter_sweep_targets(scenario.flowsheet_id) 

135 return Response( 

136 parameter_sweep.EligibleParameterSweepTargetsResponse(targets).model_dump( 

137 mode="json", 

138 ) 

139 ) 

140 

141 @extend_schema( 

142 methods=["POST"], 

143 request=parameter_sweep.ParameterSweepRequest, 

144 responses=parameter_sweep.ParameterSweepPreviewResponse, 

145 ) 

146 @action(detail=True, methods=["post"], url_path="parameter_sweep_preview") 

147 def parameter_sweep_preview(self, request, pk=None): 

148 scenario = self.get_object() 

149 preview = parameter_sweep.preview_parameter_sweep(scenario, request.data) 

150 return Response(preview.model_dump(mode="json")) 

151 

152 @extend_schema( 

153 request=parameter_sweep.ParameterSweepRequest, 

154 responses=parameter_sweep.ParameterSweepGenerateResponse, 

155 ) 

156 @action(detail=True, methods=["post"], url_path="generate_parameter_sweep") 

157 def generate_parameter_sweep(self, request, pk=None): 

158 scenario = self.get_object() 

159 access_state = get_flowsheet_access(request.user, scenario.flowsheet_id) 

160 if access_state.has_read_access and not access_state.has_write_access: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true

161 return Response( 

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

163 status=status.HTTP_403_FORBIDDEN, 

164 ) 

165 if scenario.enable_optimization: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true

166 return Response( 

167 { 

168 "enable_optimization": ( 

169 "Parameter sweep generation is not available while optimisation is enabled." 

170 ) 

171 }, 

172 status=status.HTTP_400_BAD_REQUEST, 

173 ) 

174 

175 result = parameter_sweep.generate_parameter_sweep( 

176 scenario, 

177 request.data, 

178 ) 

179 return Response(result.model_dump(mode="json")) 

180 

181 

182class OptimizationDegreesOfFreedomViewSet(ModelViewSet): 

183 serializer_class = OptimizationDegreesOfFreedomSerializer 

184 

185 def get_queryset(self): 

186 return OptimizationDegreesOfFreedom.objects.all()