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

1from dataclasses import dataclass 

2 

3from littlepay.api.client import Client 

4from requests.exceptions import HTTPError 

5 

6from benefits.core import session 

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

8 

9 

10@dataclass 

11class CardTokenizationAccessResponse: 

12 status: Status 

13 access_token: str 

14 expires_at: int 

15 exception: Exception = None 

16 status_code: int = None 

17 

18 

19def request_card_tokenization_access(request) -> CardTokenizationAccessResponse: 

20 """ 

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

22 """ 

23 agency = session.agency(request) 

24 

25 try: 

26 client = Client( 

27 base_url=agency.littlepay_config.api_base_url, 

28 client_id=agency.littlepay_config.client_id, 

29 client_secret=agency.littlepay_config.client_secret, 

30 audience=agency.littlepay_config.audience, 

31 ) 

32 client.oauth.ensure_active_token(client.token) 

33 response = client.request_card_tokenization_access() 

34 

35 return CardTokenizationAccessResponse( 

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

37 ) 

38 except Exception as e: 

39 exception = e 

40 

41 if isinstance(e, HTTPError): 

42 status_code = e.response.status_code 

43 

44 if status_code >= 500: 

45 status = Status.SYSTEM_ERROR 

46 else: 

47 status = Status.EXCEPTION 

48 else: 

49 status_code = None 

50 status = Status.EXCEPTION 

51 

52 return CardTokenizationAccessResponse( 

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

54 ) 

55 

56 

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

58 """ 

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

60 

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

62 """ 

63 agency = session.agency(request) 

64 flow = session.flow(request) 

65 

66 client = Client( 

67 base_url=agency.littlepay_config.api_base_url, 

68 client_id=agency.littlepay_config.client_id, 

69 client_secret=agency.littlepay_config.client_secret, 

70 audience=agency.littlepay_config.audience, 

71 ) 

72 client.oauth.ensure_active_token(client.token) 

73 

74 funding_source = client.get_funding_source_by_token(card_token) 

75 group_id = flow.group_id 

76 

77 exception = None 

78 try: 

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

80 

81 already_enrolled = group_funding_source is not None 

82 

83 if flow.supports_expiration: 

84 # set expiry on session 

85 if already_enrolled and group_funding_source.expiry_date is not None: 

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

87 else: 

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

89 

90 if not already_enrolled: 

91 # enroll user with an expiration date, return success 

92 client.link_concession_group_funding_source( 

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

94 ) 

95 status = Status.SUCCESS 

96 else: # already_enrolled 

97 if group_funding_source.expiry_date is None: 

98 # update expiration of existing enrollment, return success 

99 client.update_concession_group_funding_source_expiry( 

100 group_id=group_id, 

101 funding_source_id=funding_source.id, 

102 expiry=session.enrollment_expiry(request), 

103 ) 

104 status = Status.SUCCESS 

105 else: 

106 is_expired = _is_expired(group_funding_source.expiry_date) 

107 is_within_reenrollment_window = _is_within_reenrollment_window( 

108 group_funding_source.expiry_date, session.enrollment_reenrollment(request) 

109 ) 

110 

111 if is_expired or is_within_reenrollment_window: 

112 # update expiration of existing enrollment, return success 

113 client.update_concession_group_funding_source_expiry( 

114 group_id=group_id, 

115 funding_source_id=funding_source.id, 

116 expiry=session.enrollment_expiry(request), 

117 ) 

118 status = Status.SUCCESS 

119 else: 

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

121 status = Status.REENROLLMENT_ERROR 

122 else: # eligibility does not support expiration 

123 if not already_enrolled: 

124 # enroll user with no expiration date, return success 

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

126 status = Status.SUCCESS 

127 else: # already_enrolled 

128 if group_funding_source.expiry_date is None: 

129 # no action, return success 

130 status = Status.SUCCESS 

131 else: 

132 # remove expiration date, return success 

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

134 

135 except HTTPError as e: 

136 if e.response.status_code >= 500: 

137 status = Status.SYSTEM_ERROR 

138 exception = e 

139 else: 

140 status = Status.EXCEPTION 

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

142 except Exception as e: 

143 status = Status.EXCEPTION 

144 exception = e 

145 

146 return status, exception 

147 

148 

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

150 group_funding_sources = client.get_concession_group_linked_funding_sources(group_id) 

151 matching_group_funding_source = None 

152 for group_funding_source in group_funding_sources: 

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

154 matching_group_funding_source = group_funding_source 

155 break 

156 

157 return matching_group_funding_source