Coverage for benefits/enrollment/enrollment.py: 99%

1from dataclasses import dataclass 

2from datetime import datetime, timedelta 

3from enum import Enum 

4 

5import sentry_sdk 

6from django.shortcuts import redirect 

7from django.utils import timezone 

8 

9from benefits.core import models, session 

10from benefits.routes import routes 

11 

12from . import analytics 

13 

14 

15class Status(Enum): 

16 # SUCCESS means the enrollment went through successfully 

17 SUCCESS = 1 

18 

19 # SYSTEM_ERROR means the enrollment system encountered an internal error (returned a 500 HTTP status) 

20 SYSTEM_ERROR = 2 

21 

22 # EXCEPTION means the enrollment system is working, but something unexpected happened 

23 # because of a misconfiguration or invalid request from our side 

24 EXCEPTION = 3 

25 

26 # REENROLLMENT_ERROR means that the user tried to re-enroll but is not within the reenrollment window 

27 REENROLLMENT_ERROR = 4 

28 

29 

30@dataclass 

31class EnrollmentDecision: 

32 status: Status 

33 expiry_to_store: datetime | None = None 

34 expiry_to_send: datetime | None = None 

35 should_enroll: bool = False 

36 should_remove_expiry: bool = False 

37 

38 

39def _is_expired(expiry_date: datetime): 

40 """Returns whether the given datetime is in the past or not.""" 

41 return expiry_date <= timezone.now() 

42 

43 

44def _is_within_reenrollment_window(expiry_date: datetime, reenrollment_date: datetime): 

45 """Returns if the current datetime is between the reenrollment_date and the expiry_date (the reenrollment window).""" 

46 return reenrollment_date <= timezone.now() < expiry_date 

47 

48 

49def _calculate_expiry(expiration_days: int): 

50 """Returns the expiry datetime, which should be midnight in the configured timezone of the (N + 1)th day from now, 

51 where N is expiration_days.""" 

52 default_time_zone = timezone.get_default_timezone() 

53 expiry_date = timezone.localtime(timezone=default_time_zone) + timedelta(days=expiration_days + 1) 

54 expiry_datetime = expiry_date.replace(hour=0, minute=0, second=0, microsecond=0) 

55 

56 return expiry_datetime 

57 

58 

59def _calculate_reenrollment_start(expiry: datetime, reenrollment_days: int) -> datetime: 

60 """Returns the first date at which reenrollment can occur (renenrollment_days before the expiry).""" 

61 return expiry - timedelta(days=reenrollment_days) 

62 

63 

64def resolve_enrollment_decision( 

65 flow: models.EnrollmentFlow, already_enrolled: bool, existing_expiry: datetime | None 

66) -> EnrollmentDecision: 

67 """ 

68 Make a (provider-agnostic) enrollment decision given the inputs based on: 

69 

70 - If the flow supports expiration or not 

71 - If the user is already enrolled or not 

72 - If the enrollment has an existing expiry or not 

73 - If the existing expiry is valid, within the reenrollment window, or expired 

74 """ 

75 # flow does not support expiration 

76 if not flow.supports_expiration: 

77 if not already_enrolled: 

78 # not yet enrolled 

79 return EnrollmentDecision(status=Status.SUCCESS, should_enroll=True) 

80 

81 if existing_expiry is not None: 

82 # already enrolled with existing expiry: remove 

83 return EnrollmentDecision(status=Status.SUCCESS, should_remove_expiry=True) 

84 

85 # already enrolled without existing expiry: no-op 

86 return EnrollmentDecision(status=Status.SUCCESS) 

87 

88 # flow supports expiration 

89 new_expiry = _calculate_expiry(flow.expiration_days) 

90 

91 # not yet enrolled and/or no existing expiry 

92 if not already_enrolled or existing_expiry is None: 

93 return EnrollmentDecision( 

94 status=Status.SUCCESS, 

95 expiry_to_store=new_expiry, 

96 expiry_to_send=new_expiry, 

97 should_enroll=True, 

98 ) 

99 

100 # already enrolled, with expired enrollment or within reenrollment window 

101 reenrollment_start = _calculate_reenrollment_start(existing_expiry, flow.expiration_reenrollment_days) 

102 if _is_expired(existing_expiry) or _is_within_reenrollment_window(existing_expiry, reenrollment_start): 

103 return EnrollmentDecision( 

104 status=Status.SUCCESS, 

105 expiry_to_store=new_expiry, 

106 expiry_to_send=new_expiry, 

107 should_enroll=True, 

108 ) 

109 

110 # already enrolled, not expired and not within reenrollment window 

111 return EnrollmentDecision(status=Status.REENROLLMENT_ERROR, expiry_to_store=existing_expiry) 

112 

113 

114def handle_enrollment_results( 

115 request, 

116 status: Status, 

117 verified_by: str, 

118 exception: Exception = None, 

119 enrollment_method: str = models.EnrollmentMethods.SELF_SERVICE, 

120 route_reenrollment_error=routes.ENROLLMENT_REENROLLMENT_ERROR, 

121 route_success=routes.ENROLLMENT_SUCCESS, 

122 route_system_error=routes.ENROLLMENT_SYSTEM_ERROR, 

123 card_category: str = None, 

124 card_scheme: str = None, 

125): 

126 """ 

127 Handle the results of a provider enrollment call by sending the appropriate analytics events, 

128 and redirecting the user to the correct route. 

129 """ 

130 flow = session.flow(request) 

131 agency = session.agency(request) 

132 group_id = str(session.group(request).group_id) # needs to be a string for the API call 

133 match (status): 

134 case Status.SUCCESS: 

135 expiry = session.enrollment_expiry(request) 

136 oauth_extra_claims = session.oauth_extra_claims(request) 

137 # EnrollmentEvent expects a string value for extra_claims 

138 if oauth_extra_claims: 

139 str_extra_claims = ", ".join(oauth_extra_claims) 

140 else: 

141 str_extra_claims = "" 

142 

143 agencies_to_report = [agency] 

144 agencies_to_report.extend(agency.group_agencies()) 

145 

146 for agency in agencies_to_report: 

147 event = models.EnrollmentEvent.objects.create( 

148 transit_agency=agency, 

149 enrollment_flow=flow, 

150 enrollment_method=enrollment_method, 

151 verified_by=verified_by, 

152 expiration_datetime=expiry, 

153 extra_claims=str_extra_claims, 

154 ) 

155 event.save() 

156 

157 analytics.returned_success( 

158 request, 

159 agency=agency, 

160 enrollment_group=group_id, 

161 transit_processor=agency.transit_processor_system_name, 

162 enrollment_method=enrollment_method, 

163 extra_claims=oauth_extra_claims, 

164 card_scheme=card_scheme, 

165 card_category=card_category, 

166 ) 

167 

168 return redirect(route_success) 

169 

170 case Status.SYSTEM_ERROR: 

171 analytics.returned_error( 

172 request, 

173 str(exception), 

174 agency=agency, 

175 enrollment_group=group_id, 

176 transit_processor=agency.transit_processor_system_name, 

177 enrollment_method=enrollment_method, 

178 ) 

179 sentry_sdk.capture_exception(exception) 

180 return redirect(route_system_error) 

181 

182 case Status.EXCEPTION: 

183 analytics.returned_error( 

184 request, 

185 str(exception), 

186 agency=agency, 

187 enrollment_group=group_id, 

188 transit_processor=agency.transit_processor_system_name, 

189 enrollment_method=enrollment_method, 

190 ) 

191 raise exception 

192 

193 case Status.REENROLLMENT_ERROR: 193 ↛ exitline 193 didn't return from function 'handle_enrollment_results' because the pattern on line 193 always matched

194 analytics.returned_error( 

195 request, 

196 "Re-enrollment error.", 

197 agency=agency, 

198 enrollment_group=group_id, 

199 transit_processor=agency.transit_processor_system_name, 

200 enrollment_method=enrollment_method, 

201 ) 

202 return redirect(route_reenrollment_error)