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

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 

12 

13from idaes_factory.endpoints import BuildStateSolveError 

14 

15class PropertyValueViewSet(ModelViewSet): 

16 serializer_class = PropertyValueSerializer 

17 

18 def get_queryset(self): 

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

20 

21 def perform_update(self, serializer): 

22 instance = serializer.save() 

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

24 property_value_saved(instance) 

25 

26 def perform_destroy(self, instance): 

27 property_info = instance.property 

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

29 response = super().perform_destroy(instance) 

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

31 property_value_deleted(instance) 

32 return response 

33 

34 

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

36 try: 

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

38 except BuildStateSolveError as e: 

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

40 except Exception as e: 

41 import traceback 

42 traceback.print_exc() 

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

44 

45 

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

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

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

49 instance = self.get_object() 

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

51 instance.auto_replace() 

52 if instance.property_id is not None: 52 ↛ 54line 52 didn't jump to line 54 because the condition on line 52 was always true

53 property_value_saved(instance) 

54 return Response(PropertyValueSerializer(instance).data) 

55 

56 @extend_schema( 

57 responses=DictField(), 

58 parameters=[ 

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

60 ], 

61 ) 

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

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

64 return Response( 

65 create_property_value_tag_json(self.get_queryset()), 

66 content_type="application/json", 

67 headers={ 

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

69 }, 

70 ) 

71 

72 

73def create_property_value_tag_json( 

74 queryset: QuerySet[PropertyValue, PropertyValue], 

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

76 """ 

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

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

79 PropertyInfo ID and units. 

80 """ 

81 property_value: PropertyValue 

82 data = { 

83 property_value.tag: { 

84 "property": property_value.id, 

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

86 } 

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

88 } 

89 return data