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

1from dataclasses import dataclass 

2from datetime import timedelta 

3 

4from django.utils import timezone 

5from littlepay.api.client import Client 

6from requests.exceptions import HTTPError 

7 

8from benefits.core import session 

9from benefits.enrollment.enrollment import Status 

10 

11 

12@dataclass 

13class CardTokenizationAccessResponse: 

14 status: Status 

15 access_token: str 

16 expires_at: int 

17 exception: Exception = None 

18 status_code: int = None 

19 

20 

21def request_card_tokenization_access(request) -> CardTokenizationAccessResponse: 

22 """ 

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

24 """ 

25 agency = session.agency(request) 

26 

27 try: 

28 client = Client( 

29 base_url=agency.littlepay_config.api_base_url, 

30 client_id=agency.littlepay_config.client_id, 

31 client_secret=agency.littlepay_config.client_secret, 

32 audience=agency.littlepay_config.audience, 

33 ) 

34 client.oauth.ensure_active_token(client.token) 

35 response = client.request_card_tokenization_access() 

36 

37 return CardTokenizationAccessResponse( 

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

39 ) 

40 except Exception as e: 

41 exception = e 

42 

43 if isinstance(e, HTTPError): 

44 status_code = e.response.status_code 

45 

46 if status_code >= 500: 

47 status = Status.SYSTEM_ERROR 

48 else: 

49 status = Status.EXCEPTION 

50 else: 

51 status_code = None 

52 status = Status.EXCEPTION 

53 

54 return CardTokenizationAccessResponse( 

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

56 ) 

57 

58 

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

60 """ 

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

62 

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

64 """ 

65 agency = session.agency(request) 

66 flow = session.flow(request) 

67 

68 client = Client( 

69 base_url=agency.littlepay_config.api_base_url, 

70 client_id=agency.littlepay_config.client_id, 

71 client_secret=agency.littlepay_config.client_secret, 

72 audience=agency.littlepay_config.audience, 

73 ) 

74 client.oauth.ensure_active_token(client.token) 

75 

76 funding_source = client.get_funding_source_by_token(card_token) 

77 group_id = flow.group_id 

78 

79 exception = None 

80 try: 

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

82 

83 already_enrolled = group_funding_source is not None 

84 

85 if flow.supports_expiration: 

86 # set expiry on session 

87 if already_enrolled and group_funding_source.expiry_date is not None: 

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

89 else: 

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

91 

92 if not already_enrolled: 

93 # enroll user with an expiration date, return success 

94 client.link_concession_group_funding_source( 

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

96 ) 

97 status = Status.SUCCESS 

98 else: # already_enrolled 

99 if group_funding_source.expiry_date is None: 

100 # update expiration of existing enrollment, return success 

101 client.update_concession_group_funding_source_expiry( 

102 group_id=group_id, 

103 funding_source_id=funding_source.id, 

104 expiry=session.enrollment_expiry(request), 

105 ) 

106 status = Status.SUCCESS 

107 else: 

108 is_expired = _is_expired(group_funding_source.expiry_date) 

109 is_within_reenrollment_window = _is_within_reenrollment_window( 

110 group_funding_source.expiry_date, session.enrollment_reenrollment(request) 

111 ) 

112 

113 if is_expired or is_within_reenrollment_window: 

114 # update expiration of existing enrollment, return success 

115 client.update_concession_group_funding_source_expiry( 

116 group_id=group_id, 

117 funding_source_id=funding_source.id, 

118 expiry=session.enrollment_expiry(request), 

119 ) 

120 status = Status.SUCCESS 

121 else: 

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

123 status = Status.REENROLLMENT_ERROR 

124 else: # eligibility does not support expiration 

125 if not already_enrolled: 

126 # enroll user with no expiration date, return success 

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

128 status = Status.SUCCESS 

129 else: # already_enrolled 

130 if group_funding_source.expiry_date is None: 

131 # no action, return success 

132 status = Status.SUCCESS 

133 else: 

134 # remove expiration date, return success 

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

136 

137 except HTTPError as e: 

138 if e.response.status_code >= 500: 

139 status = Status.SYSTEM_ERROR 

140 exception = e 

141 else: 

142 status = Status.EXCEPTION 

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

144 except Exception as e: 

145 status = Status.EXCEPTION 

146 exception = e 

147 

148 return status, exception 

149 

150 

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

152 group_funding_sources = client.get_concession_group_linked_funding_sources(group_id) 

153 matching_group_funding_source = None 

154 for group_funding_source in group_funding_sources: 

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

156 matching_group_funding_source = group_funding_source 

157 break 

158 

159 return matching_group_funding_source 

160 

161 

162def _is_expired(expiry_date): 

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

164 return expiry_date <= timezone.now() 

165 

166 

167def _is_within_reenrollment_window(expiry_date, enrollment_reenrollment_date): 

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

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

170 

171 

172def _calculate_expiry(expiration_days): 

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

174 where N is expiration_days.""" 

175 default_time_zone = timezone.get_default_timezone() 

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

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

178 

179 return expiry_datetime