Coverage for backend/django/Economics/results/services/comparison/targets.py: 94%

1"""Resolve and validate studies selected for Economics comparisons.""" 

2 

3from __future__ import annotations 

4 

5from authentication.user.models import User 

6from core.managers import has_flowsheet_read_access 

7from core.validation import flowsheet_context 

8from Economics.shared.choices import EconomicsBaselineMode 

9from Economics.studies.models import EconomicsStudy, EconomicsStudyComparisonSelection 

10from Economics.studies.services.project_scope import study_is_in_current_project 

11from Economics.studies.services.baseline_access import ( 

12 resolve_study_reference, 

13 stable_flowsheet_id, 

14) 

15 

16from .contracts import ComparisonAvailabilityStatus, ComparisonTarget, ComparisonValidationError 

17 

18 

19def comparison_targets( 

20 *, 

21 active_study: EconomicsStudy, 

22 selections: list[EconomicsStudyComparisonSelection], 

23 user: User, 

24) -> list[ComparisonTarget]: 

25 """Return the active study followed by saved comparison selections.""" 

26 targets = [ 

27 ComparisonTarget( 

28 study_id=active_study.pk, 

29 flowsheet_id=stable_flowsheet_id(active_study), 

30 name=active_study.name, 

31 flowsheet_name=active_study.flowsheet_state.flowsheet.name, 

32 availability_status=ComparisonAvailabilityStatus.AVAILABLE, 

33 study=active_study, 

34 ) 

35 ] 

36 for selection in selections: 

37 targets.append( 

38 _target_from_selection( 

39 active_study=active_study, 

40 selection=selection, 

41 user=user, 

42 ) 

43 ) 

44 return targets 

45 

46 

47def validate_comparison_targets( 

48 *, 

49 active_study: EconomicsStudy, 

50 user: User, 

51 comparison_study_ids: list[int], 

52 baseline_mode: EconomicsBaselineMode, 

53 baseline_study_id: int, 

54) -> list[EconomicsStudy]: 

55 """Validate selected studies and preserve the user's selected ordering.""" 

56 if not 1 <= len(comparison_study_ids) <= 10: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 raise ComparisonValidationError("Select between one and ten comparison studies.") 

58 if len(set(comparison_study_ids)) != len(comparison_study_ids): 

59 raise ComparisonValidationError("Comparison studies must be unique.") 

60 if active_study.pk in comparison_study_ids: 

61 raise ComparisonValidationError("The active study is already included in the comparison.") 

62 if ( 

63 baseline_mode == EconomicsBaselineMode.STUDY 

64 and baseline_study_id != active_study.pk 

65 and baseline_study_id not in comparison_study_ids 

66 ): 

67 raise ComparisonValidationError("Baseline must be the active study or one selected comparison study.") 

68 

69 studies_by_id: dict[int, EconomicsStudy] = {} 

70 for study_id in comparison_study_ids: 

71 study = _accessible_study( 

72 active_study=active_study, 

73 study_id=study_id, 

74 user=user, 

75 ) 

76 if study is None: 

77 raise ComparisonValidationError("Selected comparison study was not found.") 

78 studies_by_id[study_id] = study 

79 return [studies_by_id[study_id] for study_id in comparison_study_ids] 

80 

81 

82def _target_from_selection( 

83 *, 

84 active_study: EconomicsStudy, 

85 selection: EconomicsStudyComparisonSelection, 

86 user: User, 

87) -> ComparisonTarget: 

88 """Resolve a saved selection to a study or fall back to its denormalized snapshot.""" 

89 study = resolve_study_reference( 

90 owner_state_id=active_study.flowsheet_state_id, 

91 target_flowsheet_id=selection.target_flowsheet_id, 

92 target_lineage_id=selection.target_study_lineage_id, 

93 ) 

94 if study is None or not has_flowsheet_read_access(user, selection.target_flowsheet_id): 

95 return _unavailable_target(selection, expose_snapshot=study is None) 

96 return ComparisonTarget( 

97 study_id=study.pk, 

98 flowsheet_id=stable_flowsheet_id(study), 

99 name=study.name, 

100 flowsheet_name=study.flowsheet_state.flowsheet.name, 

101 availability_status=ComparisonAvailabilityStatus.AVAILABLE, 

102 study=study, 

103 ) 

104 

105 

106def _unavailable_target( 

107 selection: EconomicsStudyComparisonSelection, 

108 *, 

109 expose_snapshot: bool, 

110) -> ComparisonTarget: 

111 """Expose deleted selections while hiding metadata for inaccessible live studies.""" 

112 return ComparisonTarget( 

113 study_id=None, 

114 flowsheet_id=selection.target_flowsheet_id, 

115 name=(selection.study_name_snapshot or "Unavailable study") if expose_snapshot else "Unavailable study", 

116 flowsheet_name=selection.flowsheet_name_snapshot if expose_snapshot else "", 

117 availability_status=ComparisonAvailabilityStatus.UNAVAILABLE, 

118 ) 

119 

120 

121def _accessible_study( 

122 *, 

123 active_study: EconomicsStudy, 

124 study_id: int, 

125 user: User, 

126) -> EconomicsStudy | None: 

127 """Return a project-local study only after checking access to its owning flowsheet.""" 

128 flowsheet_id = _study_flowsheet_id(study_id) 

129 if flowsheet_id is None or not has_flowsheet_read_access(user, flowsheet_id): 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true

130 return None 

131 with flowsheet_context(flowsheet_id, user): 

132 study = EconomicsStudy.objects.select_related( 

133 "flowsheet_state__flowsheet" 

134 ).filter(pk=study_id).first() 

135 if study is None or not study_is_in_current_project( 

136 study, 

137 stable_flowsheet_id(active_study), 

138 ): 

139 return None 

140 return study 

141 

142 

143def _study_flowsheet_id(study_id: int) -> int | None: 

144 """Read only the ownership key needed before the access-controlled study fetch.""" 

145 row = ( 

146 # Saved or submitted comparison studies can belong to another 

147 # flowsheet. Read only the ownership key, then re-fetch the study 

148 # through objects once access to that flowsheet is confirmed. 

149 EconomicsStudy._base_manager.filter(pk=study_id) 

150 .values("flowsheet_state__flowsheet_id") 

151 .first() 

152 ) 

153 return None if row is None else row["flowsheet_state__flowsheet_id"]