1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77
|
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from azure.core.exceptions import ServiceRequestError
from azure.identity._internal.msal_client import MsalClient
import pytest
from helpers import mock, mock_response, validating_transport, Request
def test_retries_requests():
"""The client should retry token requests"""
message = "can't connect"
transport = mock.Mock(send=mock.Mock(side_effect=ServiceRequestError(message)))
client = MsalClient(transport=transport)
with pytest.raises(ServiceRequestError, match=message):
client.post("https://localhost")
assert transport.send.call_count > 1
transport.send.reset_mock()
with pytest.raises(ServiceRequestError, match=message):
client.get("https://localhost")
assert transport.send.call_count > 1
def test_get_error_response():
first_result = {"error": "first"}
first_response = mock_response(401, json_payload=first_result)
second_result = {"error": "second"}
second_response = mock_response(401, json_payload=second_result)
transport = validating_transport(
requests=[Request(url="https://localhost")] * 2, responses=[first_response, second_response]
)
client = MsalClient(transport=transport)
for result in (first_result, second_result):
assert not client.get_error_response(result)
client.get("https://localhost")
response = client.get_error_response(first_result)
assert response is first_response
client.post("https://localhost")
response = client.get_error_response(second_result)
assert response is second_response
assert not client.get_error_response(first_result)
def test_string_response_content():
"""Test client handling string response content with error message"""
# String response content with error message
string_error = "Some string error message containing the word error"
response = mock.Mock(status_code=401, headers={})
response.text = lambda encoding=None: string_error
response.headers["content-type"] = "text/plain"
response.content_type = "text/plain"
# Create a transport that returns the string response
transport = validating_transport(
requests=[Request(url="https://localhost")],
responses=[response],
)
client = MsalClient(transport=transport)
# Make a request to store the auth error.
client.get("https://localhost")
error_code, saved_response = getattr(client._local, "error", (None, None))
assert error_code == "oauth_error"
assert saved_response is response
|