From c80b837adb2ab87ea110007be1bcbfbf139409af Mon Sep 17 00:00:00 2001 From: Chris Ballinger Date: Mon, 10 Nov 2025 13:27:10 -0800 Subject: [PATCH] Add response_headers to APIException --- appstoreserverlibrary/api_client.py | 24 +++++++++-- tests/test_api_client.py | 63 ++++++++++++++++++++++++++- tests/test_api_client_async.py | 67 ++++++++++++++++++++++++++++- 3 files changed, 147 insertions(+), 7 deletions(-) diff --git a/appstoreserverlibrary/api_client.py b/appstoreserverlibrary/api_client.py index bd583464..20a9f1bf 100644 --- a/appstoreserverlibrary/api_client.py +++ b/appstoreserverlibrary/api_client.py @@ -574,16 +574,32 @@ class APIError(IntEnum): @define class APIException(Exception): + """ + Exception raised when the App Store Server API returns an error response. + + Attributes: + http_status_code: The HTTP status code from the response + api_error: The parsed APIError enum value, if recognized + raw_api_error: The raw error code from the API response + error_message: The error message from the API response + response_headers: HTTP response headers from the error response. This is useful for accessing + rate limiting information such as the 'Retry-After' header when receiving a 429 response. + Note: Header key casing depends on the HTTP client library used. The async client (httpx) + normalizes all header keys to lowercase, while the sync client (requests) uses case-insensitive + access. For portability, use lowercase header keys (e.g., 'retry-after' instead of 'Retry-After'). + """ http_status_code: int api_error: Optional[APIError] raw_api_error: Optional[int] error_message: Optional[str] + response_headers: Optional[Dict[str, str]] = None - def __init__(self, http_status_code: int, raw_api_error: Optional[int] = None, error_message: Optional[str] = None): + def __init__(self, http_status_code: int, raw_api_error: Optional[int] = None, error_message: Optional[str] = None, response_headers: Optional[Dict[str, str]] = None): self.http_status_code = http_status_code self.raw_api_error = raw_api_error self.api_error = None self.error_message = error_message + self.response_headers = response_headers or {} try: if raw_api_error is not None: self.api_error = APIError(raw_api_error) @@ -653,14 +669,14 @@ def _parse_response(self, status_code: int, headers: MutableMapping, json_suppli else: # Best effort parsing of the response body if not 'content-type' in headers or headers['content-type'] != 'application/json': - raise APIException(status_code) + raise APIException(status_code, response_headers=dict(headers)) try: response_body = json_supplier() - raise APIException(status_code, response_body['errorCode'], response_body['errorMessage']) + raise APIException(status_code, response_body['errorCode'], response_body['errorMessage'], response_headers=dict(headers)) except APIException as e: raise e except Exception as e: - raise APIException(status_code) from e + raise APIException(status_code, response_headers=dict(headers)) from e class AppStoreServerAPIClient(BaseAppStoreServerAPIClient): diff --git a/tests/test_api_client.py b/tests/test_api_client.py index 3d683c57..31d5c490 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -434,6 +434,64 @@ def test_unknown_error(self): self.assertFalse(True) + def test_api_exception_includes_response_headers(self): + # Test that response headers are captured in APIException + custom_headers = { + 'X-Custom-Header': 'test-value', + 'X-Another-Header': 'another-value' + } + client = self.get_client_with_body(b'{"errorCode": 4040010, "errorMessage": "Transaction id not found."}', + 'POST', + 'https://local-testing-base-url/inApps/v1/notifications/test', + {}, + None, + 404, + response_headers=custom_headers) + try: + client.request_test_notification() + except APIException as e: + self.assertEqual(404, e.http_status_code) + self.assertEqual(4040010, e.raw_api_error) + self.assertEqual(APIError.TRANSACTION_ID_NOT_FOUND, e.api_error) + self.assertEqual("Transaction id not found.", e.error_message) + # Verify response_headers are captured + self.assertIsNotNone(e.response_headers) + self.assertEqual('test-value', e.response_headers['X-Custom-Header']) + self.assertEqual('another-value', e.response_headers['X-Another-Header']) + self.assertEqual('application/json', e.response_headers['Content-Type']) + return + + self.assertFalse(True) + + def test_rate_limit_with_retry_after_header(self): + # Test that Retry-After header is accessible for rate limiting + retry_after_time = '1699564800000' # Unix time in milliseconds + rate_limit_headers = { + 'Retry-After': retry_after_time, + 'X-Rate-Limit-Remaining': '0' + } + client = self.get_client_with_body(b'{"errorCode": 4290000, "errorMessage": "Rate limit exceeded."}', + 'POST', + 'https://local-testing-base-url/inApps/v1/notifications/test', + {}, + None, + 429, + response_headers=rate_limit_headers) + try: + client.request_test_notification() + except APIException as e: + self.assertEqual(429, e.http_status_code) + self.assertEqual(4290000, e.raw_api_error) + self.assertEqual(APIError.RATE_LIMIT_EXCEEDED, e.api_error) + self.assertEqual("Rate limit exceeded.", e.error_message) + # Verify Retry-After header is accessible + self.assertIsNotNone(e.response_headers) + self.assertEqual(retry_after_time, e.response_headers['Retry-After']) + self.assertEqual('0', e.response_headers['X-Rate-Limit-Remaining']) + return + + self.assertFalse(True) + def test_get_transaction_history_with_unknown_environment(self): client = self.get_client_with_body_from_file('tests/resources/models/transactionHistoryResponseWithMalformedEnvironment.json', 'GET', @@ -723,7 +781,7 @@ def test_get_app_transaction_info_transaction_id_not_found(self): def get_signing_key(self): return read_data_from_binary_file('tests/resources/certs/testSigningKey.p8') - def get_client_with_body(self, body: str, expected_method: str, expected_url: str, expected_params: Dict[str, Union[str, List[str]]], expected_json: Dict[str, Any], status_code: int = 200, expected_data: bytes = None, expected_content_type: str = None): + def get_client_with_body(self, body: str, expected_method: str, expected_url: str, expected_params: Dict[str, Union[str, List[str]]], expected_json: Dict[str, Any], status_code: int = 200, expected_data: bytes = None, expected_content_type: str = None, response_headers: Dict[str, str] = None): signing_key = self.get_signing_key() client = AppStoreServerAPIClient(signing_key, 'keyId', 'issuerId', 'com.example', Environment.LOCAL_TESTING) def fake_execute_and_validate_inputs(method: bytes, url: str, params: Dict[str, Union[str, List[str]]], headers: Dict[str, str], json: Dict[str, Any], data: bytes): @@ -753,6 +811,9 @@ def fake_execute_and_validate_inputs(method: bytes, url: str, params: Dict[str, response.status_code = status_code response.raw = BytesIO(body) response.headers['Content-Type'] = 'application/json' + if response_headers: + for key, value in response_headers.items(): + response.headers[key] = value return response client._execute_request = fake_execute_and_validate_inputs diff --git a/tests/test_api_client_async.py b/tests/test_api_client_async.py index f6a9ad1b..aad0be0c 100644 --- a/tests/test_api_client_async.py +++ b/tests/test_api_client_async.py @@ -440,6 +440,66 @@ async def test_unknown_error(self): self.assertFalse(True) + async def test_api_exception_includes_response_headers(self): + # Test that response headers are captured in APIException + custom_headers = { + 'X-Custom-Header': 'test-value', + 'X-Another-Header': 'another-value' + } + client = self.get_client_with_body(b'{"errorCode": 4040010, "errorMessage": "Transaction id not found."}', + 'POST', + 'https://local-testing-base-url/inApps/v1/notifications/test', + {}, + None, + 404, + response_headers=custom_headers) + try: + await client.request_test_notification() + except APIException as e: + self.assertEqual(404, e.http_status_code) + self.assertEqual(4040010, e.raw_api_error) + self.assertEqual(APIError.TRANSACTION_ID_NOT_FOUND, e.api_error) + self.assertEqual("Transaction id not found.", e.error_message) + # Verify response_headers are captured + # Note: httpx normalizes all header keys to lowercase + self.assertIsNotNone(e.response_headers) + self.assertEqual('test-value', e.response_headers['x-custom-header']) + self.assertEqual('another-value', e.response_headers['x-another-header']) + self.assertEqual('application/json', e.response_headers['content-type']) + return + + self.assertFalse(True) + + async def test_rate_limit_with_retry_after_header(self): + # Test that Retry-After header is accessible for rate limiting + retry_after_time = '1699564800000' # Unix time in milliseconds + rate_limit_headers = { + 'Retry-After': retry_after_time, + 'X-Rate-Limit-Remaining': '0' + } + client = self.get_client_with_body(b'{"errorCode": 4290000, "errorMessage": "Rate limit exceeded."}', + 'POST', + 'https://local-testing-base-url/inApps/v1/notifications/test', + {}, + None, + 429, + response_headers=rate_limit_headers) + try: + await client.request_test_notification() + except APIException as e: + self.assertEqual(429, e.http_status_code) + self.assertEqual(4290000, e.raw_api_error) + self.assertEqual(APIError.RATE_LIMIT_EXCEEDED, e.api_error) + self.assertEqual("Rate limit exceeded.", e.error_message) + # Verify Retry-After header is accessible + # Note: httpx normalizes all header keys to lowercase + self.assertIsNotNone(e.response_headers) + self.assertEqual(retry_after_time, e.response_headers['retry-after']) + self.assertEqual('0', e.response_headers['x-rate-limit-remaining']) + return + + self.assertFalse(True) + async def test_get_transaction_history_with_unknown_environment(self): client = self.get_client_with_body_from_file('tests/resources/models/transactionHistoryResponseWithMalformedEnvironment.json', 'GET', @@ -728,7 +788,7 @@ async def test_get_app_transaction_info_transaction_id_not_found(self): def get_signing_key(self): return read_data_from_binary_file('tests/resources/certs/testSigningKey.p8') - def get_client_with_body(self, body: str, expected_method: str, expected_url: str, expected_params: Dict[str, Union[str, List[str]]], expected_json: Dict[str, Any], status_code: int = 200, expected_data: bytes = None, expected_content_type: str = None): + def get_client_with_body(self, body: str, expected_method: str, expected_url: str, expected_params: Dict[str, Union[str, List[str]]], expected_json: Dict[str, Any], status_code: int = 200, expected_data: bytes = None, expected_content_type: str = None, response_headers: Dict[str, str] = None): signing_key = self.get_signing_key() client = AsyncAppStoreServerAPIClient(signing_key, 'keyId', 'issuerId', 'com.example', Environment.LOCAL_TESTING) async def fake_execute_and_validate_inputs(method: bytes, url: str, params: Dict[str, Union[str, List[str]]], headers: Dict[str, str], json: Dict[str, Any], data: bytes): @@ -754,7 +814,10 @@ async def fake_execute_and_validate_inputs(method: bytes, url: str, params: Dict self.assertEqual(['User-Agent', 'Authorization', 'Accept'], list(headers.keys())) self.assertEqual(expected_json, json) - response = Response(status_code, headers={'Content-Type': 'application/json'}, content=body) + response_headers_dict = {'Content-Type': 'application/json'} + if response_headers: + response_headers_dict.update(response_headers) + response = Response(status_code, headers=response_headers_dict, content=body) return response client._execute_request = fake_execute_and_validate_inputs