I have been struggling on this topic for few days. Read through many articles but cant find a solution. Hence hope someone here can help me out, or give me some directions.
I have the decorator check_response_status below which is used by other methods (for example start_export_results() in below).
**other import
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), reraise=True)
def check_response_status(custom_message=None):
"""
A wrapper method for status code check
Args:
custom_message(str) = Error message
Returns: response
"""
def decorator(func):
def wrapper(*args, **kwargs):
try:
response = func(*args, **kwargs)
if response.status_code != 200:
logger.error(f"{response.status_code}: {custom_message} - {response.json()}")
raise APIError
return response
except(requests.exceptions.ConnectionError,
requests.exceptions.Timeout) as e:
logger.error(f"{custom_message}: {e}")
raise
return wrapper
return decorator
@check_response_status("Bad HTTP status. Failed to start export")
def start_export_results(survey_base_url: str, oauth_token: str, survey_id: str, start_date: str, end_date: str,
body: dict) -> Response:
"""
Initiates the export of survey responses within a specified date range.
Args:
survey_base_url (str): The base URL of the survey service.
oauth_token (str): OAuth token for authentication.
survey_id (str): The ID of the survey for which responses are to be exported.
start_date (str): The start date of the response data to be exported. YYYY-MM-DD.
end_date (str): The cut-off date excluding any response data came in after. YYYY-MM-DD.
body (dict): Additional parameters for filtering the export results.
Returns:
Response: Response object containing the result of the export request.
"""
export_url = f'{survey_base_url}{survey_id}/export-responses'
headers = {
'Authorization': f'Bearer {oauth_token}',
'Accept': 'application/json',
'Content-Type': 'application/json'
}
request_body = body
request_body.update(
{
"startDate": start_date,
"endDate": end_date
}
)
response = requests.post(export_url, headers=headers, json=request_body)
return response
I have tried different ways to mock the retries but no matter how, the call_count is always 1 instead of 1+3, as a result the test always fails and I just cant check if the retries were executed.. Below is the last version i had for unit test.
Can anyone please give me a pointer, or where I can get more information on testing the retries? Or am I doing it completely wrong 🙁
**other import
from unittest.mock import patch, call
@pytest.mark.parametrize(
"exception", [
requests.exceptions.ConnectionError,
requests.exceptions.Timeout
]
)
@patch("flde.lib.etg_csat.csat_api_utils.requests.post")
def test_get_oauth2_token_bad_requests(mock_post, api_oauth_headers_input, exception):
mock_post.side_effect = [exception, exception, exception, exception]
# mock_post.side_effect = exception
encoded_clients_creds, base_url = api_oauth_headers_input
with pytest.raises(exception):
get_oauth2_token(base_url=base_url, encoded_token=encoded_clients_creds)
expected_calls = [call(), call(), call(), call()]
mock_post.assert_has_calls(expected_calls, any_order=True)
Thanks so much!
I expect the call_count to be 4 or at least from some attributes of mock, I can see that the retries have been executed.
Wu Pui Yan is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.