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

1from django.db import transaction 

2from rest_framework import serializers 

3from drf_spectacular.types import OpenApiTypes 

4from drf_spectacular.utils import extend_schema_field 

5from core.auxiliary.models.Flowsheet import Flowsheet 

6from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

7from authentication.user.AccessTable import AccessTable 

8from authentication.user.models import User 

9from core.auxiliary.services.flowsheet_template_transitions import ( 

10 lock_flowsheet_content_mutation, 

11) 

12 

13 

14class OwnerSerializer(serializers.ModelSerializer): 

15 class Meta: 

16 model = User 

17 fields = [ 

18 'id', 

19 'email', 

20 'first_name', 

21 'last_name' 

22 ] 

23 read_only_fields = [ 

24 'id', 

25 'email', 

26 'first_name', 

27 'last_name' 

28 ] 

29 

30 

31class FlowsheetAccessSerializer(serializers.Serializer): 

32 is_owner = serializers.BooleanField() 

33 read_only = serializers.BooleanField() 

34 can_edit = serializers.BooleanField() 

35 can_share = serializers.BooleanField() 

36 can_copy = serializers.BooleanField() 

37 can_export = serializers.BooleanField() 

38 can_manage_template_settings = serializers.BooleanField() 

39 

40 

41class FlowsheetSerializer(serializers.ModelSerializer): 

42 owner = OwnerSerializer(read_only=True) 

43 access = serializers.SerializerMethodField() 

44 economics_study_count = serializers.SerializerMethodField() 

45 rootGrouping = serializers.SerializerMethodField() 

46 buildVersion = serializers.SerializerMethodField() 

47 buildDate = serializers.SerializerMethodField() 

48 

49 class Meta: 

50 model = Flowsheet 

51 fields = [ 

52 "id", 

53 "name", 

54 "project", 

55 "savedDate", 

56 "owner", 

57 "flowsheet_template_type", 

58 "created_at", 

59 "binned_at", 

60 "auto_snapshot_after_single_solve", 

61 "rootGrouping", 

62 "buildVersion", 

63 "buildDate", 

64 "access", 

65 "economics_study_count", 

66 ] 

67 read_only_fields = [ 

68 "owner", 

69 "project", 

70 "flowsheet_template_type", 

71 "rootGrouping", 

72 "buildVersion", 

73 "buildDate", 

74 ] 

75 

76 def _get_user_access_row(self, flowsheet: Flowsheet, user: User | None): 

77 if user is None or not user.is_authenticated: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 return None 

79 

80 project = flowsheet.project 

81 if project is None: 

82 return None 

83 

84 prefetched_access_entries = getattr(project, "current_user_access_entries", None) 

85 if prefetched_access_entries is not None: 

86 return prefetched_access_entries[0] if prefetched_access_entries else None 

87 

88 return AccessTable.objects.filter(project=project, user=user).first() 

89 

90 @extend_schema_field(FlowsheetAccessSerializer) 

91 def get_access(self, obj: Flowsheet): 

92 """ 

93 Expose frontend-facing capability flags derived from the same ownership / 

94 share rules that the backend enforces for mutations. 

95 """ 

96 request = self.context.get("request") 

97 user: User | None = getattr(request, "user", None) 

98 access_row = self._get_user_access_row(obj, user) 

99 project = obj.project 

100 

101 is_owner = bool(user and user.is_authenticated and project is not None and project.owner_id == user.id) 

102 can_edit_public_template = bool( 

103 user 

104 and user.is_authenticated 

105 and user.is_staff 

106 and obj.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate 

107 ) 

108 has_direct_access = is_owner or access_row is not None 

109 has_read_access = has_direct_access or can_edit_public_template 

110 

111 if is_owner or can_edit_public_template: 

112 read_only = False 

113 elif access_row is not None: 113 ↛ 116line 113 didn't jump to line 116 because the condition on line 113 was always true

114 read_only = bool(access_row.read_only) 

115 else: 

116 read_only = True 

117 

118 can_edit = has_read_access and not read_only 

119 can_share = is_owner and obj.flowsheet_template_type == FlowsheetTemplateType.NotTemplate 

120 can_copy = has_read_access or obj.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate 

121 can_export = has_read_access 

122 

123 if obj.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate: 

124 can_manage_template_settings = bool(user and user.is_authenticated and user.is_staff) 

125 else: 

126 can_manage_template_settings = is_owner 

127 

128 return { 

129 "is_owner": is_owner, 

130 "read_only": read_only, 

131 "can_edit": can_edit, 

132 "can_share": can_share, 

133 "can_copy": can_copy, 

134 "can_export": can_export, 

135 "can_manage_template_settings": can_manage_template_settings, 

136 } 

137 

138 @extend_schema_field(OpenApiTypes.INT) 

139 def get_economics_study_count(self, obj: Flowsheet): 

140 annotated_count = getattr(obj, "economics_study_count", None) 

141 if annotated_count is not None: 

142 return annotated_count 

143 if obj.current_state_id is None: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true

144 return 0 

145 return obj.current_state.economics_studies.count() 

146 

147 @extend_schema_field(OpenApiTypes.INT) 

148 def get_rootGrouping(self, obj: Flowsheet): 

149 """Keep the public root-group identifier stable across the state cutover.""" 

150 

151 return obj.current_state.root_grouping_id if obj.current_state_id else None 

152 

153 @extend_schema_field(OpenApiTypes.STR) 

154 def get_buildVersion(self, obj: Flowsheet): 

155 return obj.current_state.build_version if obj.current_state_id else None 

156 

157 @extend_schema_field(OpenApiTypes.STR) 

158 def get_buildDate(self, obj: Flowsheet): 

159 return obj.current_state.build_date if obj.current_state_id else None 

160 

161 def create(self, validated_data): 

162 context = self.context 

163 request = context["request"] 

164 project = validated_data.get("project") 

165 owner = project.owner if project is not None else request.user 

166 

167 validated_data["owner"] = owner 

168 

169 # Return primary key of Flowsheet 

170 return Flowsheet.create(**validated_data) 

171 

172 @transaction.atomic 

173 def update(self, instance, validated_data): 

174 """Apply writable fields to a fresh project/flowsheet-locked row.""" 

175 

176 try: 

177 locked_flowsheet = lock_flowsheet_content_mutation(candidate=instance) 

178 except Flowsheet.DoesNotExist as exc: 

179 raise serializers.ValidationError( 

180 {"detail": "Flowsheet changed; retry the update."} 

181 ) from exc 

182 return super().update(locked_flowsheet, validated_data) 

183 

184 def validate(self, attrs): 

185 context = self.context 

186 request = context["request"] 

187 owner = request.user 

188 

189 if owner is None: 

190 raise serializers.ValidationError("Flowsheet must have an owner") 

191 

192 # Template state changes coordinate project-menu state and row locks in 

193 # FlowsheetTemplateViewSet. Generic create/update must not bypass that 

194 # transition boundary, including requests from older API clients. 

195 if "flowsheet_template_type" in self.initial_data: 

196 raise serializers.ValidationError( 

197 { 

198 "flowsheet_template_type": ( 

199 "Template status is managed through template actions." 

200 ) 

201 } 

202 ) 

203 

204 return super().validate(attrs)