Coverage for backend/django/core/auxiliary/viewsets/PropertyValueViewSet.py: 84%

1from core.viewset import ModelViewSet 

2from rest_framework.response import Response 

3from drf_spectacular.types import OpenApiTypes 

4from drf_spectacular.utils import OpenApiParameter, extend_schema 

5from rest_framework.decorators import action 

6from rest_framework.serializers import DictField 

7from django.db.models import QuerySet 

8from core.auxiliary.models.PropertyValue import PropertyValue 

9from core.auxiliary.property_events import property_value_deleted, property_value_saved 

10from core.auxiliary.property_state import reject_property_update 

11from ..serializers.PropertyValueSerializer import PropertyValueSerializer 

12from flowsheetInternals.unitops.models.FlowsheetEditOperation import ( 

13 FlowsheetEditOperation, 

14) 

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

16 

17from idaes_factory.endpoints import BuildStateSolveError 

18 

19class PropertyValueViewSet(ModelViewSet): 

20 serializer_class = PropertyValueSerializer 

21 

22 def get_queryset(self): 

23 return PropertyValue.objects.select_related("property") 

24 

25 def perform_update(self, serializer): 

26 instance = serializer.save() 

27 if instance.property_id is not None: 27 ↛ exitline 27 didn't return from function 'perform_update' because the condition on line 27 was always true

28 property_value_saved(instance) 

29 

30 def perform_destroy(self, instance): 

31 property_info = instance.property 

32 reject_property_update(property_info, {}, action="delete") 

33 response = super().perform_destroy(instance) 

34 if property_info is not None: 34 ↛ 36line 34 didn't jump to line 36 because the condition on line 34 was always true

35 property_value_deleted(instance) 

36 return response 

37 

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

39 """Delete a property value as one reversible operation.""" 

40 instance = self.get_object() 

41 property_set = instance.property.set if instance.property_id else None 

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

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

44 with flowsheet_edit( 

45 flowsheet=instance.flowsheet_state.flowsheet, 

46 user=request.user, 

47 kind=FlowsheetEditOperation.Kind.FieldPatch, 

48 label_key="Delete property value", 

49 ) as mutation: 

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

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

52 

53 

54 def update(self, request, *args, **kwargs) -> Response: 

55 instance = self.get_object() 

56 property_set = instance.property.set if instance.property_id else None 

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

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

59 try: 

60 with flowsheet_edit( 

61 flowsheet=instance.flowsheet_state.flowsheet, 

62 user=request.user, 

63 kind=FlowsheetEditOperation.Kind.FieldPatch, 

64 label_key="Edit property", 

65 ) as mutation: 

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

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

68 except BuildStateSolveError as e: 

69 return Response({"error": str(e)}, status=400, content_type="application/json") 

70 except Exception as e: 

71 import traceback 

72 traceback.print_exc() 

73 return Response({"error": str(e)}, status=400, content_type="application/json") 

74 

75 

76 @extend_schema(request=None, responses=PropertyValueSerializer) 

77 @action(detail=True, methods=["post"], url_path="auto_replace") 

78 def auto_replace(self, request, *args, **kwargs) -> Response: 

79 instance = self.get_object() 

80 reject_property_update(instance.property, {}, action="auto_replace") 

81 with flowsheet_edit( 

82 flowsheet=instance.flowsheet_state.flowsheet, 

83 user=request.user, 

84 kind=FlowsheetEditOperation.Kind.FieldPatch, 

85 label_key="Replace parameter", 

86 ) as mutation: 

87 instance.auto_replace() 

88 if instance.property_id is not None: 88 ↛ 90line 88 didn't jump to line 90

89 property_value_saved(instance) 

90 return Response( 

91 mutation.add_to_data(dict(PropertyValueSerializer(instance).data)) 

92 ) 

93 

94 @extend_schema( 

95 responses=DictField(), 

96 parameters=[ 

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

98 ], 

99 ) 

100 @action(detail=False, methods=["get"], url_path="download_tag_mappings") 

101 def download_tag_mappings(self, request, *args, **kwargs) -> Response: 

102 return Response( 

103 create_property_value_tag_json(self.get_queryset()), 

104 content_type="application/json", 

105 headers={ 

106 "Content-Disposition": 'attachment; filename="property-value-tag-mappings.json"' 

107 }, 

108 ) 

109 

110 

111def create_property_value_tag_json( 

112 queryset: QuerySet[PropertyValue, PropertyValue], 

113) -> dict[str, dict[str, str | int | None]]: 

114 """ 

115 Create a JSON representation of tagged PropertyValues on a flowsheet. 

116 The keys are the PropertyValue tags and the values are the related 

117 PropertyInfo ID and units. 

118 """ 

119 data = { 

120 property_value.tag: { 

121 "property": property_value.id, 

122 "units": property_value.property.unit if property_value.property else None, 

123 } 

124 for property_value in queryset.exclude(tag__isnull=True).exclude(tag="") 

125 } 

126 return data