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

1import re 

2from dataclasses import dataclass 

3 

4from littlepay.api.client import Client 

5from requests.exceptions import HTTPError 

6 

7from benefits.core import session 

8from benefits.enrollment.enrollment import Status, _calculate_expiry, _is_expired, _is_within_reenrollment_window 

9 

10 

11@dataclass 

12class CardTokenizationAccessResponse: 

13 status: Status 

14 access_token: str 

15 expires_at: int 

16 exception: Exception = None 

17 status_code: int = None 

18 

19 

20def request_card_tokenization_access(request) -> CardTokenizationAccessResponse: 

21 """ 

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

23 """ 

24 agency = session.agency(request) 

25 

26 try: 

27 client = Client( 

28 base_url=agency.littlepay_config.api_base_url, 

29 client_id=agency.littlepay_config.client_id, 

30 client_secret=agency.littlepay_config.client_secret, 

31 audience=agency.littlepay_config.audience, 

32 ) 

33 client.oauth.ensure_active_token(client.token) 

34 response = client.request_card_tokenization_access() 

35 

36 return CardTokenizationAccessResponse( 

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

38 ) 

39 except Exception as e: 

40 exception = e 

41 

42 if isinstance(e, HTTPError): 

43 status_code = e.response.status_code 

44 

45 if status_code >= 500: 

46 status = Status.SYSTEM_ERROR 

47 else: 

48 status = Status.EXCEPTION 

49 else: 

50 status_code = None 

51 status = Status.EXCEPTION 

52 

53 return CardTokenizationAccessResponse( 

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

55 ) 

56 

57 

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

59 """ 

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

61 

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

63 """ 

64 agency = session.agency(request) 

65 flow = session.flow(request) 

66 

67 client = Client( 

68 base_url=agency.littlepay_config.api_base_url, 

69 client_id=agency.littlepay_config.client_id, 

70 client_secret=agency.littlepay_config.client_secret, 

71 audience=agency.littlepay_config.audience, 

72 ) 

73 client.oauth.ensure_active_token(client.token) 

74 

75 funding_source = client.get_funding_source_by_token(card_token) 

76 group_id = flow.group_id 

77 

78 exception = None 

79 try: 

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

81 

82 already_enrolled = group_funding_source is not None 

83 

84 if flow.supports_expiration: 

85 # set expiry on session 

86 if already_enrolled and group_funding_source.expiry_date is not None: 

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

88 else: 

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

90 

91 if not already_enrolled: 

92 # enroll user with an expiration date, return success 

93 client.link_concession_group_funding_source( 

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

95 ) 

96 status = Status.SUCCESS 

97 else: # already_enrolled 

98 if group_funding_source.expiry_date is None: 

99 # update expiration of existing enrollment, return success 

100 client.update_concession_group_funding_source_expiry( 

101 group_id=group_id, 

102 funding_source_id=funding_source.id, 

103 expiry=session.enrollment_expiry(request), 

104 ) 

105 status = Status.SUCCESS 

106 else: 

107 is_expired = _is_expired(group_funding_source.expiry_date) 

108 is_within_reenrollment_window = _is_within_reenrollment_window( 

109 group_funding_source.expiry_date, session.enrollment_reenrollment(request) 

110 ) 

111 

112 if is_expired or is_within_reenrollment_window: 

113 # update expiration of existing enrollment, return success 

114 client.update_concession_group_funding_source_expiry( 

115 group_id=group_id, 

116 funding_source_id=funding_source.id, 

117 expiry=session.enrollment_expiry(request), 

118 ) 

119 status = Status.SUCCESS 

120 else: 

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

122 status = Status.REENROLLMENT_ERROR 

123 else: # eligibility does not support expiration 

124 if not already_enrolled: 

125 # enroll user with no expiration date, return success 

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

127 status = Status.SUCCESS 

128 else: # already_enrolled 

129 if group_funding_source.expiry_date is None: 

130 # no action, return success 

131 status = Status.SUCCESS 

132 else: 

133 # remove expiration date, return success 

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

135 

136 except HTTPError as e: 

137 if e.response.status_code >= 500: 

138 status = Status.SYSTEM_ERROR 

139 exception = e 

140 elif e.response.status_code == 409 and re.search(r"Funding source .+ already in group", e.response.text): 

141 # Handle situations where we errantly tried to link an already-enrolled funding source. 

142 # See: https://github.com/cal-itp/benefits/issues/3292 

143 status = Status.SUCCESS 

144 else: 

145 status = Status.EXCEPTION 

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

147 except Exception as e: 

148 status = Status.EXCEPTION 

149 exception = e 

150 

151 return status, exception, funding_source 

152 

153 

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

155 group_funding_sources = client.get_concession_group_linked_funding_sources(group_id) 

156 matching_group_funding_source = None 

157 for group_funding_source in group_funding_sources: 

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

159 matching_group_funding_source = group_funding_source 

160 break 

161 

162 return matching_group_funding_source