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

1from rest_framework import serializers 

2from core.auxiliary.models.Flowsheet import Flowsheet 

3from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

4from authentication.user.AccessTable import AccessTable 

5from authentication.user.models import User 

6 

7 

8class OwnerSerializer(serializers.ModelSerializer): 

9 class Meta: 

10 model = User 

11 fields = [ 

12 'id', 

13 'email', 

14 'first_name', 

15 'last_name' 

16 ] 

17 read_only_fields = [ 

18 'id', 

19 'email', 

20 'first_name', 

21 'last_name' 

22 ] 

23 

24 

25class FlowsheetSerializer(serializers.ModelSerializer): 

26 owner = OwnerSerializer(read_only=True) 

27 access = serializers.SerializerMethodField() 

28 

29 class Meta: 

30 model = Flowsheet 

31 fields = "__all__" 

32 read_only_fields = ['owner'] 

33 

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

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

36 return None 

37 

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

39 if prefetched_access_entries is not None: 

40 return prefetched_access_entries[0] if prefetched_access_entries else None 

41 

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

43 

44 def get_access(self, obj: Flowsheet): 

45 """ 

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

47 share rules that the backend enforces for mutations. 

48 """ 

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

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

51 access_row = self._get_user_access_row(obj, user) 

52 

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

54 can_edit_public_template = bool( 

55 user 

56 and user.is_authenticated 

57 and user.is_staff 

58 and obj.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate 

59 ) 

60 has_direct_access = is_owner or access_row is not None 

61 has_read_access = has_direct_access or can_edit_public_template 

62 

63 if is_owner or can_edit_public_template: 

64 read_only = False 

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

66 read_only = bool(access_row.read_only) 

67 else: 

68 read_only = True 

69 

70 can_edit = has_read_access and not read_only 

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

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

73 can_export = has_read_access 

74 

75 if obj.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate: 

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

77 else: 

78 can_manage_template_settings = is_owner 

79 

80 return { 

81 "is_owner": is_owner, 

82 "read_only": read_only, 

83 "can_edit": can_edit, 

84 "can_share": can_share, 

85 "can_copy": can_copy, 

86 "can_export": can_export, 

87 "can_manage_template_settings": can_manage_template_settings, 

88 } 

89 

90 def create(self, validated_data): 

91 context = self.context 

92 request = context["request"] 

93 owner = request.user 

94 

95 validated_data["owner"] = owner 

96 

97 # Return primary key of Flowsheet 

98 return Flowsheet.create(**validated_data) 

99 

100 def validate(self, attrs): 

101 context = self.context 

102 request = context["request"] 

103 owner = request.user 

104 

105 if owner is None: 

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

107 

108 return super().validate(attrs)