Coverage for backend/django/Economics/studies/services/baseline_access.py: 97%

1"""Access helpers for study-baseline references.""" 

2 

3from collections.abc import Iterable 

4 

5from rest_framework.exceptions import PermissionDenied 

6 

7from authentication.user.models import User 

8from core.managers import has_flowsheet_read_access 

9from Economics.studies.models import EconomicsStudy 

10from Economics.studies.services.project_scope import study_is_in_current_project 

11 

12 

13def stable_flowsheet_id(study: EconomicsStudy) -> int: 

14 """Return the stable flowsheet identity that owns a physical study row.""" 

15 

16 return study.flowsheet_state.flowsheet_id 

17 

18 

19def resolve_study_reference( 

20 *, 

21 owner_state_id: int, 

22 target_flowsheet_id: int | None, 

23 target_lineage_id, 

24) -> EconomicsStudy | None: 

25 """Resolve a logical study reference in the appropriate live state. 

26 

27 Same-flowsheet references resolve in the owning state so revision clones stay 

28 self-contained. Cross-flowsheet references follow the target stable 

29 flowsheet's current state. 

30 """ 

31 

32 if target_flowsheet_id is None or target_lineage_id is None: 

33 return None 

34 

35 from core.auxiliary.models.FlowsheetState import FlowsheetState 

36 from core.auxiliary.models.Flowsheet import Flowsheet 

37 

38 owner_flowsheet_id = ( 

39 FlowsheetState._base_manager.filter(pk=owner_state_id) 

40 .values_list("flowsheet_id", flat=True) 

41 .first() 

42 ) 

43 if owner_flowsheet_id == target_flowsheet_id: 

44 target_state_id = owner_state_id 

45 else: 

46 target_state_id = ( 

47 Flowsheet.objects.filter(pk=target_flowsheet_id) 

48 .values_list("current_state_id", flat=True) 

49 .first() 

50 ) 

51 if target_state_id is None: 

52 return None 

53 return EconomicsStudy._base_manager.filter( 

54 flowsheet_state_id=target_state_id, 

55 lineage_id=target_lineage_id, 

56 ).first() 

57 

58 

59def resolve_baseline_study(study: EconomicsStudy) -> EconomicsStudy | None: 

60 """Resolve a study's logical baseline reference to its current physical row.""" 

61 

62 if study.flowsheet_state_id is None: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 return None 

64 return resolve_study_reference( 

65 owner_state_id=study.flowsheet_state_id, 

66 target_flowsheet_id=study.baseline_flowsheet_id, 

67 target_lineage_id=study.baseline_study_lineage_id, 

68 ) 

69 

70 

71def set_baseline_study_reference( 

72 study: EconomicsStudy, 

73 baseline_study: EconomicsStudy | None, 

74) -> None: 

75 """Store or clear the logical identity of a selected baseline study.""" 

76 

77 if baseline_study is None: 

78 study.baseline_flowsheet_id = None 

79 study.baseline_study_lineage_id = None 

80 return 

81 study.baseline_flowsheet_id = stable_flowsheet_id(baseline_study) 

82 study.baseline_study_lineage_id = baseline_study.lineage_id 

83 

84 

85def baseline_study_is_readable_by_user(user: User, baseline_study: EconomicsStudy) -> bool: 

86 """Return whether the user can read a selected baseline study.""" 

87 flowsheet_id = stable_flowsheet_id(baseline_study) 

88 return flowsheet_id in _readable_flowsheet_ids( 

89 user=user, 

90 flowsheet_ids={flowsheet_id}, 

91 ) 

92 

93 

94def require_readable_baseline_studies_for_user( 

95 instance_or_instances: EconomicsStudy | Iterable[EconomicsStudy], 

96 *, 

97 user: User, 

98) -> None: 

99 """Raise 403 when a selected study baseline points at an unreadable flowsheet.""" 

100 if isinstance(instance_or_instances, EconomicsStudy): 

101 _require_readable_baseline_studies([instance_or_instances], user=user) 

102 return 

103 instances = list(instance_or_instances) 

104 _require_readable_baseline_studies(instances, user=user) 

105 

106 

107def _require_readable_baseline_studies(studies: list[EconomicsStudy], *, user: User) -> None: 

108 baseline_flowsheet_ids = { 

109 study.baseline_flowsheet_id 

110 for study in studies 

111 if study.baseline_flowsheet_id is not None 

112 } 

113 readable_flowsheet_ids = _readable_flowsheet_ids( 

114 user=user, 

115 flowsheet_ids=baseline_flowsheet_ids, 

116 ) 

117 for study in studies: 

118 if study.baseline_flowsheet_id is None: 

119 continue 

120 baseline_study = resolve_baseline_study(study) 

121 baseline_is_invalid = not baseline_study_matches_target_study(study, baseline_study) 

122 baseline_is_unreadable = study.baseline_flowsheet_id not in readable_flowsheet_ids 

123 if baseline_is_invalid or baseline_is_unreadable: 

124 raise PermissionDenied("Selected baseline study is not accessible.") 

125 

126 

127def _readable_flowsheet_ids(*, user: User, flowsheet_ids: set[int]) -> set[int]: 

128 if not flowsheet_ids: 

129 return set() 

130 return { 

131 flowsheet_id 

132 for flowsheet_id in flowsheet_ids 

133 if has_flowsheet_read_access(user, flowsheet_id) 

134 } 

135 

136 

137def baseline_study_matches_target_study( 

138 target_study: EconomicsStudy, 

139 baseline_study: EconomicsStudy | None, 

140) -> bool: 

141 """Return whether a baseline points at a non-template sibling flowsheet.""" 

142 

143 return ( 

144 baseline_study is not None 

145 and target_study.flowsheet_state_id is not None 

146 and study_is_in_current_project( 

147 baseline_study, 

148 stable_flowsheet_id(target_study), 

149 ) 

150 )