Coverage for backend/django/core/managers.py: 91%

1from django.db import models 

2from authentication.user.models import User 

3from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

4from core.auxiliary.models.Flowsheet import Flowsheet 

5from core.validation import get_current_flowsheet, cache_result 

6from authentication.user.AccessTable import AccessTable 

7from typing import TypeVar 

8t = TypeVar('t', bound=models.Model, covariant=True) 

9 

10 

11def verify_flowsheet_access(user: User, flowsheet_id: int) -> bool: 

12 """ 

13 Verify if the user has access to the flowsheet either directly or through public template edit access. 

14 

15 :param user: User object 

16 :param flowsheet_id: Flowsheet ID to check access for 

17 

18 :return: True if the user has access, False otherwise 

19 """ 

20 user_has_direct_access = AccessTable.objects.filter( 

21 user_id=user.id, 

22 flowsheet_id=flowsheet_id 

23 ).exists() 

24 

25 user_can_edit_template = False 

26 

27 # Check if this is a public flowsheet template that the user can edit (only admins can edit public templates) 

28 if user.is_staff and not user_has_direct_access: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 flowsheet: Flowsheet = Flowsheet.objects.get(id=flowsheet_id) 

30 user_can_edit_template = flowsheet.flowsheet_template_type == FlowsheetTemplateType.PublicTemplate 

31 

32 access_granted = user_has_direct_access or user_can_edit_template 

33 

34 cache_result(has_access=access_granted) 

35 

36 return access_granted 

37 

38 

39class AccessControlManager(models.Manager): 

40 """ 

41 Custom object manager to enforce access control based on the current flowsheet and user context. 

42 This manager overrides the default queryset to filter objects based on the flowsheet. 

43 

44 By default, object modification methods (update, delete, etc.) will be scoped only to the 

45 active flowsheet that the user has access to. Creation methods check explicitly if the user has 

46 access to the flowsheet before allowing creation. 

47 """ 

48 

49 def __init__(self): 

50 super().__init__() 

51 self.flowsheet_related_name = "flowsheet" 

52 

53 def create(self, **kwargs): 

54 ctx = get_current_flowsheet() or {} 

55 

56 flowsheet = ctx.get("flowsheet") 

57 user = ctx.get("user") 

58 has_access = ctx.get("has_access") 

59 

60 # This is only the case when tests uses the objects directly without going through a view 

61 if flowsheet == None or user == None: 

62 return super().create(**kwargs) 

63 

64 if has_access is None: 64 ↛ 66line 64 didn't jump to line 66 because the condition on line 64 was never true

65 # Check access and cache the result 

66 has_access = verify_flowsheet_access(user=user, flowsheet_id=flowsheet) 

67 

68 kwargs[f"{self.flowsheet_related_name}_id"] = flowsheet 

69 

70 if has_access: 70 ↛ 73line 70 didn't jump to line 73 because the condition on line 70 was always true

71 return super().create(**kwargs) 

72 else: 

73 raise PermissionError("User does not have access to this flowsheet.") 

74 

75 def bulk_create(self, objs, **kwargs): 

76 ctx = get_current_flowsheet() or {} 

77 

78 flowsheet = ctx.get("flowsheet") 

79 user = ctx.get("user") 

80 has_access = ctx.get("has_access") 

81 

82 # This is only the case when tests uses the objects directly without going through a view 

83 if flowsheet == None or user == None: 

84 return super().bulk_create(objs, **kwargs) 

85 

86 if has_access is None: 

87 # Check access and cache the result 

88 has_access = verify_flowsheet_access(user=user, flowsheet_id=flowsheet) 

89 

90 # Set flowsheet ID on each object before bulk create 

91 for obj in objs: 

92 setattr(obj, f"{self.flowsheet_related_name}_id", flowsheet) 

93 

94 if has_access: 94 ↛ 97line 94 didn't jump to line 97 because the condition on line 94 was always true

95 return super().bulk_create(objs, **kwargs) 

96 else: 

97 raise PermissionError("User does not have access to this flowsheet.") 

98 def get_queryset(self): 

99 """ 

100 Override the get_queryset method to filter objects based on the current flowsheet and user access. 

101 """ 

102 qs = super().get_queryset() 

103 ctx = get_current_flowsheet() or {} 

104 

105 flowsheet = ctx.get("flowsheet") 

106 user = ctx.get("user") 

107 has_access = ctx.get("has_access") 

108 

109 # This is only the case when tests uses the objects directly without going through a view 

110 if flowsheet == None or user == None: 

111 return qs 

112 

113 # Use cached access result if present 

114 if has_access is None: 

115 # Check access and cache the result 

116 has_access = verify_flowsheet_access(user=user, flowsheet_id=flowsheet) 

117 

118 filter_kwargs = {f"{self.flowsheet_related_name}_id": flowsheet} 

119 

120 if has_access: 

121 return qs.filter(**filter_kwargs) 

122 else: 

123 return qs.none() 

124 

125class SoftDeleteManager(AccessControlManager): 

126 def __init__(self): 

127 super().__init__() 

128 def get_queryset(self): 

129 # filter out deleted objects 

130 return super().get_queryset().filter(is_deleted=False) 

131 

132 def include_deleted(self): 

133 # include deleted objects 

134 return super().get_queryset() 

135 

136 

137 

138def include_soft_deleted[t](objects: models.Manager[t]) -> models.QuerySet[t]: 

139 """ 

140 In some operations, including copying a flowsheet, 

141 we need to include the soft deleted objects 

142 Example usage: 

143 include_soft_deleted(SimulationObject.objects).filter(flowsheet_id=1) 

144 """ 

145 if objects is not None and isinstance(objects, SoftDeleteManager): 

146 return objects.include_deleted().all() 

147 else: 

148 return objects.all() 

149 

150 

151