Coverage for postrfp/shared/issue_transition.py: 89%

1from datetime import datetime, timezone 

2from typing import Any, List, Tuple 

3 

4from .exceptions import AuthorizationFailure 

5from postrfp.authorisation import perms 

6from postrfp.model import Issue, RespondentOrganisation 

7from postrfp.model.questionnaire.b36 import from_b36 

8from postrfp.model.audit import evt_types 

9from postrfp.model.exc import ( 

10 IllegalStatusAction, 

11 DeadlineHasPassed, 

12) 

13 

14 

15def change_issue_status(issue: Issue, user: Any, new_status: str) -> None: 

16 """ 

17 Performs status transitions based on current status and new_status. 

18 

19 Validates permissions, performs necessary actions (like setting dates), 

20 logs AuditEvents, and updates the status attribute. 

21 

22 TODO - Remove this module when Issue is FSM integrated 

23 """ 

24 if issue.project.status_name != "Live": 

25 raise IllegalStatusAction( 

26 "Project status must be Live to change status of an Issue" 

27 ) 

28 if new_status not in Issue.issue_statuses_int: 

29 raise ValueError(f'"{new_status}" is not a valid Issue Status') 

30 current_status = issue.status 

31 changes: List[Tuple[str, str, str]] = [("status", current_status, new_status)] 

32 match current_status: 

33 case "Not Sent": 

34 event_name = _handle_not_sent(issue, user, new_status, changes) 

35 case "Opportunity": 

36 event_name = _handle_opportunity(issue, user, new_status, changes) 

37 case "Accepted": 

38 event_name = _handle_accepted(issue, user, new_status, changes) 

39 case "Updateable": 

40 event_name = _handle_updateable(issue, user, new_status, changes) 

41 case "Submitted": 

42 event_name = _handle_submitted(issue, user, new_status, changes) 

43 case "Retracted": 

44 event_name = _handle_retracted(issue, user, new_status, changes) 

45 case "Declined": 

46 raise IllegalStatusAction( 

47 f"Cannot change status from {current_status} to {new_status}" 

48 ) 

49 case _: 

50 raise IllegalStatusAction(f"Unknown current status: {current_status}") 

51 

52 # --- Perform the change --- 

53 if event_name: 

54 issue.log_event(event_name, user, changes) 

55 # Update the status attribute directly 

56 issue.status = new_status 

57 else: 

58 # This case should ideally be caught by the specific transition checks above 

59 tmpl = "Cannot change status from %s to %s" 

60 msg = tmpl % (current_status, new_status) 

61 raise IllegalStatusAction(msg) 

62 

63 

64def _handle_not_sent( 

65 issue: Issue, user: Any, new_status: str, changes: List[Tuple[str, str, str]] 

66) -> str: 

67 if not user.organisation.is_buyside: 

68 raise AuthorizationFailure( 

69 "User must be a participant (Buyer or Consultant) to perform this action." 

70 ) 

71 user.check_permission(perms.ISSUE_PUBLISH) 

72 if new_status in ("Opportunity", "Updateable", "Accepted"): 

73 issue.issue_date = datetime.now(timezone.utc) 

74 changes.append(("issue_date", "", str(issue.issue_date))) 

75 return "ISSUE_RELEASED" 

76 raise IllegalStatusAction(f"Cannot change status from Not Sent to {new_status}") 

77 

78 

79def _handle_opportunity( 

80 issue: Issue, user: Any, new_status: str, changes: List[Tuple[str, str, str]] 

81) -> str: 

82 if new_status == "Accepted": 

83 if not isinstance(user.organisation, RespondentOrganisation): 

84 raise AuthorizationFailure( 

85 "User must be a respondent to perform this action." 

86 ) 

87 user.check_permission(perms.ISSUE_ACCEPT) 

88 return "ISSUE_ACCEPTED" 

89 elif new_status == "Declined": 

90 if not isinstance(user.organisation, RespondentOrganisation): 

91 raise AuthorizationFailure( 

92 "User must be a respondent to perform this action." 

93 ) 

94 user.check_permission(perms.ISSUE_DECLINE) 

95 return "ISSUE_DECLINED" 

96 elif new_status == "Retracted": 

97 if not user.organisation.is_buyside: 

98 raise AuthorizationFailure( 

99 "User must be a participant (Buyer or Consultant) to perform this action." 

100 ) 

101 user.check_permission(perms.ISSUE_RETRACT) 

102 return "ISSUE_RETRACTED" 

103 raise IllegalStatusAction(f"Cannot change status from Opportunity to {new_status}") 

104 

105 

106def _handle_accepted( 

107 issue: Issue, user: Any, new_status: str, changes: List[Tuple[str, str, str]] 

108) -> str: 

109 if new_status == "Declined": 

110 if not isinstance(user.organisation, RespondentOrganisation): 

111 raise AuthorizationFailure( 

112 "User must be a respondent to perform this action." 

113 ) 

114 user.check_permission(perms.ISSUE_DECLINE) 

115 return "ISSUE_DECLINED" 

116 elif new_status == "Submitted": 

117 if not isinstance(user.organisation, RespondentOrganisation): 

118 raise AuthorizationFailure( 

119 "User must be a respondent to perform this action." 

120 ) 

121 user.check_permission(perms.ISSUE_SUBMIT) 

122 from postrfp.shared import fetch # local import to avoid circular dependency 

123 

124 if issue.deadline and issue.deadline < datetime.now(): 

125 raise DeadlineHasPassed( 

126 "This Response cannot be submitted - Deadline has passed" 

127 ) 

128 unanswered = fetch.unanswered_mandatory(issue).all() 

129 if unanswered: 

130 uc = len(unanswered) 

131 nums = ", ".join(from_b36(num) for (_, num) in unanswered) 

132 v = ("is", "") if uc == 1 else ("are", "s") 

133 m = ( 

134 f"Cannot submit issue #{issue.id} ({issue.respondent_id}): there {v[0]} {uc} " 

135 f"unanswered mandatory question{v[1]} remaining: {nums}" 

136 ) 

137 raise IllegalStatusAction(m) 

138 previous_submitted_date = ( 

139 str(issue.submitted_date) if issue.submitted_date else "" 

140 ) 

141 issue.submitted_date = datetime.now() 

142 changes.append( 

143 ("submitted_date", previous_submitted_date, str(issue.submitted_date)) 

144 ) 

145 return "ISSUE_SUBMITTED" 

146 elif new_status == "Retracted": 

147 if not user.organisation.is_buyside: 

148 raise AuthorizationFailure( 

149 "User must be a participant (Buyer or Consultant) to perform this action." 

150 ) 

151 user.check_permission(perms.ISSUE_RETRACT) 

152 return "ISSUE_RETRACTED" 

153 raise IllegalStatusAction(f"Cannot change status from Accepted to {new_status}") 

154 

155 

156def _handle_updateable( 

157 issue: Issue, user: Any, new_status: str, changes: List[Tuple[str, str, str]] 

158) -> str: 

159 if new_status == "Retracted": 

160 if not user.organisation.is_buyside: 

161 raise AuthorizationFailure( 

162 "User must be a participant (Buyer or Consultant) to perform this action." 

163 ) 

164 user.check_permission(perms.ISSUE_RETRACT) 

165 return "ISSUE_RETRACTED" 

166 raise IllegalStatusAction(f"Cannot change status from Updateable to {new_status}") 

167 

168 

169def _handle_submitted( 

170 issue: Issue, user: Any, new_status: str, changes: List[Tuple[str, str, str]] 

171) -> str: 

172 if new_status == "Accepted": 

173 if not user.organisation.is_buyside: 

174 raise AuthorizationFailure( 

175 "User must be a participant (Buyer or Consultant) to perform this action." 

176 ) 

177 user.check_permission(perms.ISSUE_PUBLISH) 

178 return evt_types.ISSUE_REVERTED_TO_ACCEPTED 

179 raise IllegalStatusAction(f"Cannot change status from Submitted to {new_status}") 

180 

181 

182def _handle_retracted( 

183 issue: Issue, user: Any, new_status: str, changes: List[Tuple[str, str, str]] 

184) -> str: 

185 if not user.organisation.is_buyside: 

186 raise AuthorizationFailure( 

187 "User must be a participant (Buyer or Consultant) to perform this action." 

188 ) 

189 user.check_permission(perms.ISSUE_PUBLISH) 

190 if new_status in ("Accepted", "Opportunity"): 

191 return "ISSUE_RELEASED" 

192 raise IllegalStatusAction(f"Cannot change status from Retracted to {new_status}") 

193 

194 

195__all__ = ["change_issue_status"]