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

1import re 

2from dataclasses import dataclass 

3 

4from littlepay.api.client import Client 

5from littlepay.api.funding_sources import FundingSourceResponse 

6from requests.exceptions import HTTPError 

7 

8from benefits.core import session 

9from benefits.enrollment.enrollment import Status, resolve_enrollment_decision 

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 littlepay_config = agency.transit_processor 

29 client = Client( 

30 base_url=littlepay_config.api_base_url, 

31 client_id=littlepay_config.client_id, 

32 client_secret=littlepay_config.client_secret, 

33 audience=littlepay_config.audience, 

34 ) 

35 client.oauth.ensure_active_token(client.token) 

36 response = client.request_card_tokenization_access() 

37 

38 return CardTokenizationAccessResponse( 

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

40 ) 

41 except Exception as e: 

42 exception = e 

43 

44 if isinstance(e, HTTPError): 

45 status_code = e.response.status_code 

46 

47 if status_code >= 500: 

48 status = Status.SYSTEM_ERROR 

49 else: 

50 status = Status.EXCEPTION 

51 else: 

52 status_code = None 

53 status = Status.EXCEPTION 

54 

55 return CardTokenizationAccessResponse( 

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

57 ) 

58 

59 

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

61 """ 

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

63 

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

65 """ 

66 agency = session.agency(request) 

67 flow = session.flow(request) 

68 

69 littlepay_config = agency.transit_processor 

70 client = Client( 

71 base_url=littlepay_config.api_base_url, 

72 client_id=littlepay_config.client_id, 

73 client_secret=littlepay_config.client_secret, 

74 audience=littlepay_config.audience, 

75 ) 

76 client.oauth.ensure_active_token(client.token) 

77 

78 funding_source = client.get_funding_source_by_token(card_token) 

79 group_id = str(session.group(request).group_id) # needs to be a string for the API call 

80 

81 exception = None 

82 try: 

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

84 already_enrolled = group_funding_source is not None 

85 existing_expiry = group_funding_source.expiry_date if already_enrolled else None 

86 

87 decision = resolve_enrollment_decision(flow, already_enrolled, existing_expiry) 

88 status = decision.status 

89 

90 if decision.expiry_to_store is not None: 

91 session.update(request, enrollment_expiry=decision.expiry_to_store) 

92 

93 if status is Status.SUCCESS: 

94 if decision.should_remove_expiry: 

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

96 elif decision.should_enroll: 

97 if not already_enrolled: 

98 if decision.expiry_to_send is None: 

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

100 else: 

101 client.link_concession_group_funding_source( 

102 group_id=group_id, funding_source_id=funding_source.id, expiry=decision.expiry_to_send 

103 ) 

104 else: 

105 client.update_concession_group_funding_source_expiry( 

106 group_id=group_id, funding_source_id=funding_source.id, expiry=decision.expiry_to_send 

107 ) 

108 

109 except HTTPError as e: 

110 if e.response.status_code >= 500: 

111 status = Status.SYSTEM_ERROR 

112 exception = e 

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

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

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

116 status = Status.SUCCESS 

117 else: 

118 status = Status.EXCEPTION 

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

120 except Exception as e: 

121 status = Status.EXCEPTION 

122 exception = e 

123 

124 return status, exception, funding_source 

125 

126 

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

128 group_funding_sources = client.get_concession_group_linked_funding_sources(group_id) 

129 matching_group_funding_source = None 

130 for group_funding_source in group_funding_sources: 

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

132 matching_group_funding_source = group_funding_source 

133 break 

134 

135 return matching_group_funding_source