Coverage for backend/django/core/auxiliary/serializers/FlowsheetRevisionSerializer.py: 94%

1"""Schema-backed serializers for the stable-flowsheet revision API.""" 

2 

3from rest_framework import serializers 

4 

5from core.auxiliary.models.FlowsheetState import ( 

6 FlowsheetRevisionKind, 

7 FlowsheetState, 

8) 

9from core.auxiliary.serializers.FlowsheetSerializer import ( 

10 FlowsheetSerializer, 

11 OwnerSerializer, 

12) 

13from core.auxiliary.services.flowsheet_states import FlowsheetRevisionCategory 

14 

15 

16class FlowsheetRevisionSerializer(serializers.ModelSerializer): 

17 """Product-facing metadata for one immutable state.""" 

18 

19 state_id = serializers.IntegerField(source="pk", read_only=True) 

20 kind = serializers.CharField(source="revision_kind", read_only=True) 

21 display_label = serializers.SerializerMethodField() 

22 creator = OwnerSerializer(source="created_by", read_only=True, allow_null=True) 

23 root_grouping_id = serializers.IntegerField(read_only=True, allow_null=True) 

24 

25 class Meta: 

26 model = FlowsheetState 

27 fields = [ 

28 "state_id", 

29 "revision_number", 

30 "kind", 

31 "label", 

32 "display_label", 

33 "creator", 

34 "root_grouping_id", 

35 "created_at", 

36 "source_saved_at", 

37 "restored_from_revision_number", 

38 ] 

39 read_only_fields = fields 

40 

41 def get_display_label(self, obj: FlowsheetState) -> str: 

42 """Return product copy without exposing internal lifecycle roles.""" 

43 

44 if obj.label: 

45 return obj.label 

46 return { 

47 FlowsheetRevisionKind.AUTO_SOLVE: "After solve", 

48 FlowsheetRevisionKind.BEFORE_RESTORE: "Before restore", 

49 }.get(obj.revision_kind, "Saved revision") 

50 

51 

52class CurrentFlowsheetStateSerializer(serializers.ModelSerializer): 

53 """Minimal summary for the mutable state shown as Current in the UI.""" 

54 

55 state_id = serializers.IntegerField(source="pk", read_only=True) 

56 display_label = serializers.SerializerMethodField() 

57 

58 class Meta: 

59 model = FlowsheetState 

60 fields = [ 

61 "state_id", 

62 "display_label", 

63 "restored_from_revision_number", 

64 ] 

65 read_only_fields = fields 

66 

67 def get_display_label(self, obj: FlowsheetState) -> str: 

68 """Use the product name for the mutable state.""" 

69 

70 return "Current" 

71 

72 

73class FlowsheetRevisionListSerializer(serializers.Serializer): 

74 current = CurrentFlowsheetStateSerializer() 

75 has_active_tasks = serializers.BooleanField() 

76 count = serializers.IntegerField() 

77 page = serializers.IntegerField() 

78 page_size = serializers.IntegerField() 

79 pages = serializers.IntegerField() 

80 revisions = FlowsheetRevisionSerializer(many=True) 

81 

82 

83class ListFlowsheetRevisionsQuerySerializer(serializers.Serializer): 

84 """Validate the product category used to filter revision history.""" 

85 

86 category = serializers.ChoiceField( 

87 choices=FlowsheetRevisionCategory, 

88 required=False, 

89 default=FlowsheetRevisionCategory.ALL, 

90 ) 

91 

92 

93class CreateFlowsheetRevisionSerializer(serializers.Serializer): 

94 repair_legacy_formula_units = serializers.BooleanField( 

95 required=False, 

96 default=False, 

97 write_only=True, 

98 ) 

99 label = serializers.CharField( 

100 min_length=1, 

101 max_length=64, 

102 trim_whitespace=True, 

103 required=False, 

104 ) 

105 

106 def validate_label(self, value: str) -> str: 

107 """Validate the final trimmed product name.""" 

108 

109 normalized = value.strip() 

110 if not normalized: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 raise serializers.ValidationError("Enter a revision name.") 

112 return normalized 

113 

114 

115class UpdateFlowsheetRevisionSerializer(serializers.Serializer): 

116 """Validate the editable product name for an immutable revision.""" 

117 

118 label = serializers.CharField( 

119 min_length=1, 

120 max_length=64, 

121 trim_whitespace=True, 

122 ) 

123 

124 def validate_label(self, value: str) -> str: 

125 """Reject names that contain only whitespace.""" 

126 

127 normalized = value.strip() 

128 if not normalized: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true

129 raise serializers.ValidationError("Enter a revision name.") 

130 return normalized 

131 

132 

133class RestoreFlowsheetRevisionWarningSerializer(serializers.Serializer): 

134 """Structured warning emitted when a logical reference cannot be restored.""" 

135 

136 code = serializers.ChoiceField(choices=["economics_reference_unavailable"]) 

137 owning_study_lineage = serializers.UUIDField() 

138 reference_kind = serializers.CharField() 

139 

140 

141class RestoreFlowsheetRevisionResponseSerializer(serializers.Serializer): 

142 flowsheet = FlowsheetSerializer() 

143 current = CurrentFlowsheetStateSerializer() 

144 cleared_calculated_data_categories = serializers.ListField( 

145 child=serializers.CharField(), 

146 ) 

147 restored_from_revision_number = serializers.IntegerField() 

148 warnings = RestoreFlowsheetRevisionWarningSerializer(many=True)