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

1from rest_framework import serializers 

2from drf_spectacular.utils import extend_schema_field 

3from core.auxiliary.models.Flowsheet import Flowsheet 

4from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

5from authentication.user.AccessTable import AccessTable 

6from authentication.user.models import User 

7 

8 

9class OwnerSerializer(serializers.ModelSerializer): 

10 class Meta: 

11 model = User 

12 fields = [ 

13 'id', 

14 'email', 

15 'first_name', 

16 'last_name' 

17 ] 

18 read_only_fields = [ 

19 'id', 

20 'email', 

21 'first_name', 

22 'last_name' 

23 ] 

24 

25 

26class FlowsheetAccessSerializer(serializers.Serializer): 

27 is_owner = serializers.BooleanField() 

28 read_only = serializers.BooleanField() 

29 can_edit = serializers.BooleanField() 

30 can_share = serializers.BooleanField() 

31 can_copy = serializers.BooleanField() 

32 can_export = serializers.BooleanField() 

33 can_manage_template_settings = serializers.BooleanField() 

34 

35 

36class FlowsheetSerializer(serializers.ModelSerializer): 

37 owner = OwnerSerializer(read_only=True) 

38 access = serializers.SerializerMethodField() 

39 

40 class Meta: 

41 model = Flowsheet 

42 fields = "__all__" 

43 read_only_fields = ['owner'] 

44 

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

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

47 return None 

48 

49 prefetched_access_entries = getattr(flowsheet, "current_user_access_entries", None) 

50 if prefetched_access_entries is not None: 

51 return prefetched_access_entries[0] if prefetched_access_entries else None 

52 

53 return AccessTable.objects.filter(flowsheet=flowsheet, user=user).first() 

54 

55 @extend_schema_field(FlowsheetAccessSerializer) 

56 def get_access(self, obj: Flowsheet): 

57 """ 

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

59 share rules that the backend enforces for mutations. 

60 """ 

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

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

63 access_row = self._get_user_access_row(obj, user) 

64 

65 is_owner = bool(user and user.is_authenticated and obj.owner_id == user.id) 

66 can_edit_public_template = bool( 

67 user 

68 and user.is_authenticated 

69 and user.is_staff 

70 and obj.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate 

71 ) 

72 has_direct_access = is_owner or access_row is not None 

73 has_read_access = has_direct_access or can_edit_public_template 

74 

75 if is_owner or can_edit_public_template: 

76 read_only = False 

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

78 read_only = bool(access_row.read_only) 

79 else: 

80 read_only = True 

81 

82 can_edit = has_read_access and not read_only 

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

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

85 can_export = has_read_access 

86 

87 if obj.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate: 

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

89 else: 

90 can_manage_template_settings = is_owner 

91 

92 return { 

93 "is_owner": is_owner, 

94 "read_only": read_only, 

95 "can_edit": can_edit, 

96 "can_share": can_share, 

97 "can_copy": can_copy, 

98 "can_export": can_export, 

99 "can_manage_template_settings": can_manage_template_settings, 

100 } 

101 

102 def create(self, validated_data): 

103 context = self.context 

104 request = context["request"] 

105 owner = request.user 

106 

107 validated_data["owner"] = owner 

108 

109 # Return primary key of Flowsheet 

110 return Flowsheet.create(**validated_data) 

111 

112 def validate(self, attrs): 

113 context = self.context 

114 request = context["request"] 

115 owner = request.user 

116 

117 if owner is None: 

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

119 

120 return super().validate(attrs)