Coverage for backend/django/core/auxiliary/viewsets/PropertyInfoViewSet.py: 85%

1from core.viewset import ModelViewSet, ReadOnlyModelViewSet 

2from rest_framework import serializers 

3from ..serializers.PropertyInfoSerializer import PropertyInfoSerializer, PropertyHistorySerializer 

4from ..models import PropertyInfo, HistoricalValue 

5from drf_spectacular.utils import extend_schema, OpenApiParameter, OpenApiTypes 

6from rest_framework.decorators import action 

7from rest_framework.response import Response 

8from core.auxiliary.property_events import ( 

9 property_info_deleted, 

10 property_info_deleting, 

11 property_info_saved, 

12) 

13from core.auxiliary.property_state import reject_property_update 

14from flowsheetInternals.unitops.models.FlowsheetEditOperation import ( 

15 FlowsheetEditOperation, 

16) 

17from flowsheetInternals.unitops.services.edit_operations.mutation import flowsheet_edit 

18 

19class CompoundUpdateSerializer(serializers.Serializer): 

20 basis = serializers.CharField() 

21 flowType = serializers.CharField() 

22 value = serializers.CharField(allow_blank=True) 

23 

24class PropertyInfoViewSet(ModelViewSet): 

25 serializer_class = PropertyInfoSerializer 

26 

27 def get_queryset(self): 

28 return PropertyInfo.objects.all() 

29 

30 def perform_update(self, serializer): 

31 property_info = serializer.save() 

32 property_info_saved(property_info) 

33 

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

35 instance = self.get_object() 

36 property_set = instance.set 

37 if property_set is None: 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true

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

39 with flowsheet_edit( 

40 flowsheet=instance.flowsheet_state.flowsheet, 

41 user=request.user, 

42 kind=FlowsheetEditOperation.Kind.FieldPatch, 

43 label_key="Edit property", 

44 ) as mutation: 

45 response = super().update(request, *args, **kwargs) 

46 return mutation.add_to_response(response, header_only=True) 

47 

48 def perform_destroy(self, instance): 

49 reject_property_update(instance, {}, action="delete") 

50 property_info_deleting(instance) 

51 response = super().perform_destroy(instance) 

52 property_info_deleted(instance) 

53 return response 

54 

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

56 """Delete a custom property as one reversible operation.""" 

57 instance = self.get_object() 

58 property_set = instance.set 

59 if property_set is None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

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

61 with flowsheet_edit( 

62 flowsheet=instance.flowsheet_state.flowsheet, 

63 user=request.user, 

64 kind=FlowsheetEditOperation.Kind.FieldPatch, 

65 label_key="Delete custom property", 

66 ) as mutation: 

67 response = super().destroy(request, *args, **kwargs) 

68 return mutation.add_to_response(response, header_only=True) 

69 

70class HistoryViewSet(ReadOnlyModelViewSet): 

71 serializer_class = PropertyHistorySerializer 

72 

73 @extend_schema( 

74 parameters=[ 

75 OpenApiParameter(name="property", required=True, type=OpenApiTypes.INT), 

76 ] 

77 ) 

78 def get_queryset(self): 

79 # Filter to only the history of a specific property, e.g material stream flowrate 

80 queryset = HistoricalValue.objects.all() 

81 property = self.request.query_params.get("property") 

82 if property is not None: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 queryset = queryset.filter(property=property) 

84 return queryset 

85 

86 

87 @extend_schema( 

88 parameters=[ 

89 OpenApiParameter(name="flowsheet", required=True, type=OpenApiTypes.INT), 

90 ] 

91 ) 

92 @action(detail=False, methods=['delete']) 

93 def delete(self,request): 

94 # Delete all  

95 id = self.request.query_params.get("flowsheet") 

96 HistoricalValue.objects.filter( 

97 property__set__setOwner__flowsheet_state__flowsheet_id=id 

98 ).delete() 

99 HistoricalValue.objects.filter( 

100 property__set__msSetOwner__flowsheet_state__flowsheet_id=id 

101 ).delete() 

102 return Response(status=204)