File: test_utils.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 (138 lines) | stat: -rw-r--r-- 4,684 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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
import sys
from unittest.mock import patch

import pytest
from corehttp.utils import case_insensitive_dict
from corehttp.utils._utils import get_running_async_lock
from corehttp.runtime.policies._utils import parse_retry_after


@pytest.fixture()
def accept_cases():
    return ["accept", "Accept", "ACCEPT", "aCCePT"]


def test_case_insensitive_dict_basic(accept_cases):
    my_dict = case_insensitive_dict({"accept": "application/json"})
    for accept_case in accept_cases:
        assert my_dict[accept_case] == "application/json"


def test_case_insensitive_dict_override(accept_cases):
    for accept_case in accept_cases:
        my_dict = case_insensitive_dict({accept_case: "should-not/be-me"})
        my_dict["accept"] = "application/json"
        assert my_dict[accept_case] == my_dict["accept"] == "application/json"


def test_case_insensitive_dict_initialization():
    dict_response = {"platformUpdateDomainCount": 5, "platformFaultDomainCount": 3, "virtualMachines": []}
    a = case_insensitive_dict(platformUpdateDomainCount=5, platformFaultDomainCount=3, virtualMachines=[])
    b = case_insensitive_dict(
        zip(["platformUpdateDomainCount", "platformFaultDomainCount", "virtualMachines"], [5, 3, []])
    )
    c = case_insensitive_dict(
        [("platformFaultDomainCount", 3), ("platformUpdateDomainCount", 5), ("virtualMachines", [])]
    )
    d = case_insensitive_dict({"virtualMachines": [], "platformFaultDomainCount": 3, "platformUpdateDomainCount": 5})
    e = case_insensitive_dict({"platformFaultDomainCount": 3, "virtualMachines": []}, platformUpdateDomainCount=5)
    f = case_insensitive_dict(dict_response)
    g = case_insensitive_dict(**dict_response)
    assert a == b == c == d == e == f == g
    dicts = [a, b, c, d, e, f, g]
    for d in dicts:
        assert len(d) == 3
        assert d["platformUpdateDomainCount"] == d["platformupdatedomaincount"] == d["PLATFORMUPDATEDOMAINCOUNT"] == 5
        assert d["platformFaultDomainCount"] == d["platformfaultdomaincount"] == d["PLATFORMFAULTDOMAINCOUNT"] == 3
        assert d["virtualMachines"] == d["virtualmachines"] == d["VIRTUALMACHINES"] == []


def test_case_insensitive_dict_cant_compare():
    my_dict = case_insensitive_dict({"accept": "application/json"})
    assert my_dict != "accept"


def test_case_insensitive_dict_lowerkey_items():
    my_dict = case_insensitive_dict({"accept": "application/json"})
    assert list(my_dict.lowerkey_items()) == [("accept", "application/json")]


@pytest.mark.parametrize(
    "other, expected",
    (
        ({"PLATFORMUPDATEDOMAINCOUNT": 5}, True),
        ({}, False),
        (None, False),
    ),
)
def test_case_insensitive_dict_equality(other, expected):
    my_dict = case_insensitive_dict({"platformUpdateDomainCount": 5})
    result = my_dict == other
    assert result == expected


def test_case_insensitive_dict_keys():
    keys = ["One", "TWO", "tHrEe", "four"]
    my_dict = case_insensitive_dict({key: idx for idx, key in enumerate(keys, 1)})
    dict_keys = list(my_dict.keys())

    assert dict_keys == keys


def test_case_insensitive_copy():
    keys = ["One", "TWO", "tHrEe", "four"]
    my_dict = case_insensitive_dict({key: idx for idx, key in enumerate(keys, 1)})
    copy_my_dict = my_dict.copy()
    assert copy_my_dict is not my_dict
    assert copy_my_dict == my_dict


def test_case_insensitive_keys_present(accept_cases):
    my_dict = case_insensitive_dict({"accept": "application/json"})

    for key in accept_cases:
        assert key in my_dict


def test_case_insensitive_keys_delete(accept_cases):
    my_dict = case_insensitive_dict({"accept": "application/json"})

    for key in accept_cases:
        del my_dict[key]
        assert key not in my_dict
        my_dict[key] = "application/json"


def test_case_iter():
    keys = ["One", "TWO", "tHrEe", "four"]
    my_dict = case_insensitive_dict({key: idx for idx, key in enumerate(keys, 1)})

    for key in my_dict:
        assert key in keys


@pytest.mark.asyncio
async def test_get_running_async_module_asyncio():
    import asyncio

    assert isinstance(get_running_async_lock(), asyncio.Lock)


def test_get_running_async_module_sync():
    with pytest.raises(RuntimeError):
        get_running_async_lock()


def test_parse_retry_after():
    ret = parse_retry_after("100")
    assert ret == 100
    ret = parse_retry_after("Fri, 1 Oct 2100 00:00:00 GMT")
    assert ret > 0
    ret = parse_retry_after("0")
    assert ret == 0
    ret = parse_retry_after("0.9")
    assert ret == 0.9