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

1from dataclasses import dataclass 

2from datetime import timedelta 

3import json 

4 

5from django.conf import settings 

6from django.utils import timezone 

7from littlepay.api.client import Client 

8from requests.exceptions import HTTPError 

9 

10from benefits.core import session 

11from benefits.enrollment.enrollment import Status 

12 

13 

14@dataclass 

15class CardTokenizationAccessResponse: 

16 status: Status 

17 access_token: str 

18 expires_at: int 

19 exception: Exception = None 

20 status_code: int = None 

21 

22 

23def get_card_types_for_js() -> str: 

24 """Get a list of card types to use for enrollment, converted to a JSON string for use in JavaScript.""" 

25 card_types = ["visa", "mastercard"] 

26 if settings.LITTLEPAY_ADDITIONAL_CARDTYPES: 

27 card_types.extend(["discover", "amex"]) 

28 

29 return json.dumps(card_types) 

30 

31 

32def request_card_tokenization_access(request) -> CardTokenizationAccessResponse: 

33 """ 

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

35 """ 

36 agency = session.agency(request) 

37 

38 try: 

39 client = Client( 

40 base_url=agency.littlepay_config.api_base_url, 

41 client_id=agency.littlepay_config.client_id, 

42 client_secret=agency.littlepay_config.client_secret, 

43 audience=agency.littlepay_config.audience, 

44 ) 

45 client.oauth.ensure_active_token(client.token) 

46 response = client.request_card_tokenization_access() 

47 

48 return CardTokenizationAccessResponse( 

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

50 ) 

51 except Exception as e: 

52 exception = e 

53 

54 if isinstance(e, HTTPError): 

55 status_code = e.response.status_code 

56 

57 if status_code >= 500: 

58 status = Status.SYSTEM_ERROR 

59 else: 

60 status = Status.EXCEPTION 

61 else: 

62 status_code = None 

63 status = Status.EXCEPTION 

64 

65 return CardTokenizationAccessResponse( 

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

67 ) 

68 

69 

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

71 """ 

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

73 

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

75 """ 

76 agency = session.agency(request) 

77 flow = session.flow(request) 

78 

79 client = Client( 

80 base_url=agency.littlepay_config.api_base_url, 

81 client_id=agency.littlepay_config.client_id, 

82 client_secret=agency.littlepay_config.client_secret, 

83 audience=agency.littlepay_config.audience, 

84 ) 

85 client.oauth.ensure_active_token(client.token) 

86 

87 funding_source = client.get_funding_source_by_token(card_token) 

88 group_id = flow.group_id 

89 

90 exception = None 

91 try: 

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

93 

94 already_enrolled = group_funding_source is not None 

95 

96 if flow.supports_expiration: 

97 # set expiry on session 

98 if already_enrolled and group_funding_source.expiry_date is not None: 

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

100 else: 

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

102 

103 if not already_enrolled: 

104 # enroll user with an expiration date, return success 

105 client.link_concession_group_funding_source( 

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

107 ) 

108 status = Status.SUCCESS 

109 else: # already_enrolled 

110 if group_funding_source.expiry_date is None: 

111 # update expiration of existing enrollment, return success 

112 client.update_concession_group_funding_source_expiry( 

113 group_id=group_id, 

114 funding_source_id=funding_source.id, 

115 expiry=session.enrollment_expiry(request), 

116 ) 

117 status = Status.SUCCESS 

118 else: 

119 is_expired = _is_expired(group_funding_source.expiry_date) 

120 is_within_reenrollment_window = _is_within_reenrollment_window( 

121 group_funding_source.expiry_date, session.enrollment_reenrollment(request) 

122 ) 

123 

124 if is_expired or is_within_reenrollment_window: 

125 # update expiration of existing enrollment, return success 

126 client.update_concession_group_funding_source_expiry( 

127 group_id=group_id, 

128 funding_source_id=funding_source.id, 

129 expiry=session.enrollment_expiry(request), 

130 ) 

131 status = Status.SUCCESS 

132 else: 

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

134 status = Status.REENROLLMENT_ERROR 

135 else: # eligibility does not support expiration 

136 if not already_enrolled: 

137 # enroll user with no expiration date, return success 

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

139 status = Status.SUCCESS 

140 else: # already_enrolled 

141 if group_funding_source.expiry_date is None: 

142 # no action, return success 

143 status = Status.SUCCESS 

144 else: 

145 # remove expiration date, return success 

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

147 

148 except HTTPError as e: 

149 if e.response.status_code >= 500: 

150 status = Status.SYSTEM_ERROR 

151 exception = e 

152 else: 

153 status = Status.EXCEPTION 

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

155 except Exception as e: 

156 status = Status.EXCEPTION 

157 exception = e 

158 

159 return status, exception 

160 

161 

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

163 group_funding_sources = client.get_concession_group_linked_funding_sources(group_id) 

164 matching_group_funding_source = None 

165 for group_funding_source in group_funding_sources: 

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

167 matching_group_funding_source = group_funding_source 

168 break 

169 

170 return matching_group_funding_source 

171 

172 

173def _is_expired(expiry_date): 

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

175 return expiry_date <= timezone.now() 

176 

177 

178def _is_within_reenrollment_window(expiry_date, enrollment_reenrollment_date): 

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

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

181 

182 

183def _calculate_expiry(expiration_days): 

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

185 where N is expiration_days.""" 

186 default_time_zone = timezone.get_default_timezone() 

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

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

189 

190 return expiry_datetime