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

1from dataclasses import dataclass 

2from datetime import timedelta 

3from enum import Enum 

4 

5from django.utils import timezone 

6from littlepay.api.client import Client 

7from requests.exceptions import HTTPError 

8 

9from benefits.core import session 

10 

11 

12class Status(Enum): 

13 # SUCCESS means the enrollment went through successfully 

14 SUCCESS = 1 

15 

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

17 SYSTEM_ERROR = 2 

18 

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

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

21 EXCEPTION = 3 

22 

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

24 REENROLLMENT_ERROR = 4 

25 

26 

27@dataclass 

28class CardTokenizationAccessResponse: 

29 status: Status 

30 access_token: str 

31 expires_at: str 

32 exception: Exception = None 

33 status_code: int = None 

34 

35 

36def request_card_tokenization_access(request) -> CardTokenizationAccessResponse: 

37 """ 

38 Requests an access token to be used for card tokenization. 

39 """ 

40 agency = session.agency(request) 

41 

42 try: 

43 client = Client( 

44 base_url=agency.transit_processor.api_base_url, 

45 client_id=agency.transit_processor_client_id, 

46 client_secret=agency.transit_processor_client_secret, 

47 audience=agency.transit_processor_audience, 

48 ) 

49 client.oauth.ensure_active_token(client.token) 

50 response = client.request_card_tokenization_access() 

51 

52 return CardTokenizationAccessResponse( 

53 status=Status.SUCCESS, access_token=response.get("access_token"), expires_at=response.get("expires_at") 

54 ) 

55 except Exception as e: 

56 exception = e 

57 

58 if isinstance(e, HTTPError): 

59 status_code = e.response.status_code 

60 

61 if status_code >= 500: 

62 status = Status.SYSTEM_ERROR 

63 else: 

64 status = Status.EXCEPTION 

65 else: 

66 status_code = None 

67 status = Status.EXCEPTION 

68 

69 return CardTokenizationAccessResponse( 

70 status=status, access_token=None, expires_at=None, exception=exception, status_code=status_code 

71 ) 

72 

73 

74def enroll(request, card_token) -> tuple[Status, Exception]: 

75 """ 

76 Attempts to enroll this card into the transit processor group for the flow in the request's session. 

77 

78 Returns a tuple containing a Status indicating the result of the attempt and any exception that occurred. 

79 """ 

80 agency = session.agency(request) 

81 flow = session.flow(request) 

82 

83 client = Client( 

84 base_url=agency.transit_processor.api_base_url, 

85 client_id=agency.transit_processor_client_id, 

86 client_secret=agency.transit_processor_client_secret, 

87 audience=agency.transit_processor_audience, 

88 ) 

89 client.oauth.ensure_active_token(client.token) 

90 

91 funding_source = client.get_funding_source_by_token(card_token) 

92 group_id = flow.group_id 

93 

94 exception = None 

95 try: 

96 group_funding_source = _get_group_funding_source(client=client, group_id=group_id, funding_source_id=funding_source.id) 

97 

98 already_enrolled = group_funding_source is not None 

99 

100 if flow.supports_expiration: 

101 # set expiry on session 

102 if already_enrolled and group_funding_source.expiry_date is not None: 

103 session.update(request, enrollment_expiry=group_funding_source.expiry_date) 

104 else: 

105 session.update(request, enrollment_expiry=_calculate_expiry(flow.expiration_days)) 

106 

107 if not already_enrolled: 

108 # enroll user with an expiration date, return success 

109 client.link_concession_group_funding_source( 

110 group_id=group_id, funding_source_id=funding_source.id, expiry=session.enrollment_expiry(request) 

111 ) 

112 status = Status.SUCCESS 

113 else: # already_enrolled 

114 if group_funding_source.expiry_date is None: 

115 # update expiration of existing enrollment, return success 

116 client.update_concession_group_funding_source_expiry( 

117 group_id=group_id, 

118 funding_source_id=funding_source.id, 

119 expiry=session.enrollment_expiry(request), 

120 ) 

121 status = Status.SUCCESS 

122 else: 

123 is_expired = _is_expired(group_funding_source.expiry_date) 

124 is_within_reenrollment_window = _is_within_reenrollment_window( 

125 group_funding_source.expiry_date, session.enrollment_reenrollment(request) 

126 ) 

127 

128 if is_expired or is_within_reenrollment_window: 

129 # update expiration of existing enrollment, return success 

130 client.update_concession_group_funding_source_expiry( 

131 group_id=group_id, 

132 funding_source_id=funding_source.id, 

133 expiry=session.enrollment_expiry(request), 

134 ) 

135 status = Status.SUCCESS 

136 else: 

137 # re-enrollment error, return enrollment error with expiration and reenrollment_date 

138 status = Status.REENROLLMENT_ERROR 

139 else: # eligibility does not support expiration 

140 if not already_enrolled: 

141 # enroll user with no expiration date, return success 

142 client.link_concession_group_funding_source(group_id=group_id, funding_source_id=funding_source.id) 

143 status = Status.SUCCESS 

144 else: # already_enrolled 

145 if group_funding_source.expiry_date is None: 

146 # no action, return success 

147 status = Status.SUCCESS 

148 else: 

149 # remove expiration date, return success 

150 raise NotImplementedError("Removing expiration date is currently not supported") 

151 

152 except HTTPError as e: 

153 if e.response.status_code >= 500: 

154 status = Status.SYSTEM_ERROR 

155 exception = e 

156 else: 

157 status = Status.EXCEPTION 

158 exception = Exception(f"{e}: {e.response.json()}") 

159 except Exception as e: 

160 status = Status.EXCEPTION 

161 exception = e 

162 

163 return status, exception 

164 

165 

166def _get_group_funding_source(client: Client, group_id, funding_source_id): 

167 group_funding_sources = client.get_concession_group_linked_funding_sources(group_id) 

168 matching_group_funding_source = None 

169 for group_funding_source in group_funding_sources: 

170 if group_funding_source.id == funding_source_id: 170 ↛ 169line 170 didn't jump to line 169 because the condition on line 170 was always true

171 matching_group_funding_source = group_funding_source 

172 break 

173 

174 return matching_group_funding_source 

175 

176 

177def _is_expired(expiry_date): 

178 """Returns whether the passed in datetime is expired or not.""" 

179 return expiry_date <= timezone.now() 

180 

181 

182def _is_within_reenrollment_window(expiry_date, enrollment_reenrollment_date): 

183 """Returns if we are currently within the reenrollment window.""" 

184 return enrollment_reenrollment_date <= timezone.now() < expiry_date 

185 

186 

187def _calculate_expiry(expiration_days): 

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

189 where N is expiration_days.""" 

190 default_time_zone = timezone.get_default_timezone() 

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

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

193 

194 return expiry_datetime