Coverage for benefits/enrollment_init/api.py: 100%

1import base64 

2from dataclasses import dataclass 

3 

4import requests 

5from django.conf import settings 

6 

7from benefits.enrollment.api import BaseDataClass 

8 

9 

10@dataclass 

11class TransitAccountResponse(BaseDataClass): 

12 Id: int 

13 CardholderId: int 

14 

15 

16@dataclass 

17class CardholderResponse(BaseDataClass): 

18 FareCategory: int 

19 FareCategoryExpiry: str 

20 

21 

22class Client: 

23 def __init__(self, api_url, username, password): 

24 self.api_url = api_url.strip("/") 

25 

26 encoded_bytes = base64.b64encode(bytes(f"{username}:{password}", "utf-8")) 

27 self.authorization_header_value = f"Basic {encoded_bytes.decode()}" 

28 

29 def _endpoint_url(self, endpoint): 

30 return f"{self.api_url}/api/{endpoint}" 

31 

32 def _authorization_header(self): 

33 return {"Authorization": self.authorization_header_value} 

34 

35 def get_transit_account(self, banking_service_token, timeout=settings.REQUESTS_TIMEOUT) -> TransitAccountResponse: 

36 url = self._endpoint_url("TransitAccounts") 

37 response = requests.get( 

38 url, 

39 params={ 

40 "BankingServiceToken": banking_service_token, 

41 }, 

42 headers=self._authorization_header(), 

43 timeout=timeout, 

44 ) 

45 

46 response.raise_for_status() 

47 

48 response_json = response.json() 

49 total_count = response_json["TotalCount"] 

50 results = response_json["Result"] 

51 

52 if total_count == 1: 

53 return TransitAccountResponse.from_kwargs(**results[0]) 

54 elif total_count == 0: 

55 return None 

56 else: 

57 raise ValueError(f"Unexpectedly received more than 1 TransitAccount for token {banking_service_token}") 

58 

59 def get_cardholder(self, cardholder_id: int, timeout=settings.REQUESTS_TIMEOUT) -> CardholderResponse: 

60 url = self._endpoint_url("Cardholders") + f"/{cardholder_id}" 

61 response = requests.get( 

62 url, 

63 headers=self._authorization_header(), 

64 timeout=timeout, 

65 ) 

66 

67 response.raise_for_status() 

68 

69 response_json = response.json() 

70 

71 if "Id" in response_json and response_json["Id"] == cardholder_id: 

72 return CardholderResponse.from_kwargs(**response_json) 

73 else: 

74 raise ValueError(f"Unexpected response when querying for cardholder {cardholder_id}: {response_json}") 

75 

76 def post_cardholder( 

77 self, fare_category: int, transit_account_id: int, timeout=settings.REQUESTS_TIMEOUT 

78 ) -> CardholderResponse: 

79 url = self._endpoint_url("Cardholders") 

80 request_body = { 

81 "FareCategory": fare_category, 

82 "TransitAccountId": transit_account_id, 

83 } 

84 response = requests.post( 

85 url, 

86 json=request_body, 

87 headers=self._authorization_header(), 

88 timeout=timeout, 

89 ) 

90 

91 response.raise_for_status() 

92 

93 return CardholderResponse.from_kwargs(**response.json()) 

94 

95 def put_cardholder(self, cardholder_id: int, fare_category: int, timeout=settings.REQUESTS_TIMEOUT) -> CardholderResponse: 

96 url = self._endpoint_url("Cardholders") + f"/{cardholder_id}" 

97 request_body = { 

98 "FareCategory": fare_category, 

99 } 

100 response = requests.put( 

101 url, 

102 json=request_body, 

103 headers=self._authorization_header(), 

104 timeout=timeout, 

105 ) 

106 

107 response.raise_for_status() 

108 

109 return CardholderResponse.from_kwargs(**response.json())