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

1import re 

2 

3from dataclasses import dataclass 

4 

5from littlepay.api.client import Client 

6from requests.exceptions import HTTPError 

7 

8from benefits.core import session 

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

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 elif e.response.status_code == 409 and re.search(r"Funding source .+ already in group", e.response.text): 

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

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

144 status = Status.SUCCESS 

145 else: 

146 status = Status.EXCEPTION 

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

148 except Exception as e: 

149 status = Status.EXCEPTION 

150 exception = e 

151 

152 return status, exception, funding_source 

153 

154 

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

156 group_funding_sources = client.get_concession_group_linked_funding_sources(group_id) 

157 matching_group_funding_source = None 

158 for group_funding_source in group_funding_sources: 

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

160 matching_group_funding_source = group_funding_source 

161 break 

162 

163 return matching_group_funding_source