File: test_msal_client.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 851,724 kB
  • sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (77 lines) | stat: -rw-r--r-- 2,677 bytes parent folder | download
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