Coverage for benefits/enrollment_switchio/enrollment.py: 98%

1from dataclasses import dataclass 

2from datetime import datetime 

3 

4from django.conf import settings 

5from django.http import HttpRequest 

6from django.urls import reverse 

7from requests import HTTPError 

8 

9from benefits.core import session 

10from benefits.core.models.enrollment import EnrollmentFlow 

11from benefits.enrollment.enrollment import Status, resolve_enrollment_decision 

12from benefits.enrollment_switchio.api import ( 

13 EnrollmentClient, 

14 EshopResponseMode, 

15 Registration, 

16 RegistrationMode, 

17 RegistrationStatus, 

18 TokenizationClient, 

19) 

20from benefits.enrollment_switchio.models import SwitchioConfig, SwitchioGroup 

21 

22from .routes import routes 

23 

24 

25@dataclass 

26class RegistrationResponse: 

27 status: Status 

28 registration: Registration 

29 exception: Exception = None 

30 status_code: int = None 

31 

32 

33@dataclass 

34class RegistrationStatusResponse: 

35 status: Status 

36 registration_status: RegistrationStatus 

37 exception: Exception = None 

38 status_code: int = None 

39 

40 

41@dataclass 

42class Token: 

43 token: str 

44 tokenVersion: int 

45 tokenState: str 

46 validFrom: datetime 

47 validTo: datetime 

48 testOnly: bool 

49 par: str = None 

50 

51 

52def request_registration( 

53 request, switchio_config: SwitchioConfig, redirect_route: str = routes.ENROLLMENT_SWITCHIO_INDEX 

54) -> RegistrationResponse: 

55 try: 

56 client = TokenizationClient( 

57 api_url=switchio_config.tokenization_api_base_url, 

58 api_key=switchio_config.tokenization_api_key, 

59 api_secret=switchio_config.tokenization_api_secret, 

60 private_key=switchio_config.private_key_data, 

61 client_certificate=switchio_config.client_certificate_data, 

62 ca_certificate=switchio_config.ca_certificate_data, 

63 ) 

64 

65 route = reverse(redirect_route) 

66 redirect_url = _generate_redirect_uri(request, route) 

67 

68 registration = client.request_registration( 

69 eshopRedirectUrl=redirect_url, 

70 mode=RegistrationMode.REGISTER, 

71 eshopResponseMode=EshopResponseMode.QUERY, 

72 timeout=settings.REQUESTS_TIMEOUT, 

73 ) 

74 

75 return RegistrationResponse(status=Status.SUCCESS, registration=registration) 

76 except Exception as e: 

77 exception = e 

78 

79 if isinstance(e, HTTPError): 

80 status_code = e.response.status_code 

81 

82 if status_code >= 500: 

83 status = Status.SYSTEM_ERROR 

84 else: 

85 status = Status.EXCEPTION 

86 else: 

87 status_code = None 

88 status = Status.EXCEPTION 

89 

90 return RegistrationResponse(status=status, registration=None, exception=exception, status_code=status_code) 

91 

92 

93# copied from https://github.com/Office-of-Digital-Services/django-cdt-identity/blob/main/cdt_identity/views.py#L42-L50 

94def _generate_redirect_uri(request: HttpRequest, redirect_path: str): 

95 redirect_uri = str(request.build_absolute_uri(redirect_path)).lower() 

96 

97 # this is a temporary hack to ensure redirect URIs are HTTPS when the app is deployed 

98 # see https://github.com/cal-itp/benefits/issues/442 for more context 

99 if not redirect_uri.startswith("http://localhost"): 99 ↛ 102line 99 didn't jump to line 102 because the condition on line 99 was always true

100 redirect_uri = redirect_uri.replace("http://", "https://") 

101 

102 return redirect_uri 

103 

104 

105def get_registration_status(switchio_config: SwitchioConfig, registration_id: str) -> RegistrationStatusResponse: 

106 try: 

107 client = TokenizationClient( 

108 api_url=switchio_config.tokenization_api_base_url, 

109 api_key=switchio_config.tokenization_api_key, 

110 api_secret=switchio_config.tokenization_api_secret, 

111 private_key=switchio_config.private_key_data, 

112 client_certificate=switchio_config.client_certificate_data, 

113 ca_certificate=switchio_config.ca_certificate_data, 

114 ) 

115 

116 registration_status = client.get_registration_status( 

117 registration_id=registration_id, 

118 timeout=settings.REQUESTS_TIMEOUT, 

119 ) 

120 

121 return RegistrationStatusResponse(status=Status.SUCCESS, registration_status=registration_status, exception=None) 

122 except Exception as e: 

123 exception = e 

124 

125 if isinstance(e, HTTPError): 

126 status_code = e.response.status_code 

127 

128 if status_code >= 500: 

129 status = Status.SYSTEM_ERROR 

130 else: 

131 status = Status.EXCEPTION 

132 else: 

133 status_code = None 

134 status = Status.EXCEPTION 

135 

136 return RegistrationStatusResponse( 

137 status=status, registration_status=None, exception=exception, status_code=status_code 

138 ) 

139 

140 

141def get_latest_active_token_value(tokens): 

142 latest_active_token = None 

143 

144 for token_dict in tokens: 

145 token = Token(**token_dict) 

146 if token.tokenState == "active": 

147 if latest_active_token is None or token.validFrom > latest_active_token.validFrom: 147 ↛ 144line 147 didn't jump to line 144 because the condition on line 147 was always true

148 latest_active_token = token 

149 

150 return latest_active_token.token if latest_active_token else "" 

151 

152 

153def enroll( 

154 request, switchio_config: SwitchioConfig, flow: EnrollmentFlow, group: SwitchioGroup, token: str 

155) -> tuple[Status, Exception]: 

156 """ 

157 Attempts to enroll the token into the group. 

158 

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

160 """ 

161 client = EnrollmentClient( 

162 api_url=switchio_config.enrollment_api_base_url, 

163 authorization_header_value=switchio_config.enrollment_api_authorization_header, 

164 private_key=switchio_config.private_key_data, 

165 client_certificate=switchio_config.client_certificate_data, 

166 ca_certificate=switchio_config.ca_certificate_data, 

167 ) 

168 

169 pto_id = switchio_config.pto_id 

170 group_id = group.group_id 

171 

172 exception = None 

173 try: 

174 token_group = _get_group_for_token(client, pto_id, group_id, token) 

175 already_enrolled = token_group is not None 

176 existing_expiry = token_group.expiresAt if already_enrolled else None 

177 

178 decision = resolve_enrollment_decision(flow, already_enrolled, existing_expiry) 

179 status = decision.status 

180 

181 if decision.expiry_to_store is not None: 

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

183 

184 if status is Status.SUCCESS: 

185 if decision.should_remove_expiry: 

186 # when expiry is omitted, Switchio will set any existing expiration date to null 

187 client.add_group_to_token(pto_id=pto_id, group_id=group_id, token=token, timeout=settings.REQUESTS_TIMEOUT) 

188 elif decision.should_enroll: 

189 if decision.expiry_to_send is None: 

190 client.add_group_to_token(pto_id=pto_id, group_id=group_id, token=token, timeout=settings.REQUESTS_TIMEOUT) 

191 else: 

192 client.add_group_to_token( 

193 pto_id=pto_id, 

194 group_id=group_id, 

195 token=token, 

196 expiry=decision.expiry_to_send, 

197 timeout=settings.REQUESTS_TIMEOUT, 

198 ) 

199 

200 except HTTPError as e: 

201 if e.response.status_code >= 500: 

202 status = Status.SYSTEM_ERROR 

203 exception = e 

204 else: 

205 status = Status.EXCEPTION 

206 exception = Exception(f"{e}: {e.response.text}") 

207 except Exception as e: 

208 status = Status.EXCEPTION 

209 exception = e 

210 

211 return status, exception 

212 

213 

214def _get_group_for_token(client: EnrollmentClient, pto_id, group_id, token): 

215 already_enrolled_groups = client.get_groups_for_token(pto_id=pto_id, token=token, timeout=settings.REQUESTS_TIMEOUT) 

216 

217 for group in already_enrolled_groups: 

218 if group.group == group_id: 218 ↛ 217line 218 didn't jump to line 217 because the condition on line 218 was always true

219 return group 

220 

221 return None