forked from microsoftgraph/msgraph-sdk-python-core
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_graph_client_factory.py
More file actions
145 lines (122 loc) · 6.19 KB
/
test_graph_client_factory.py
File metadata and controls
145 lines (122 loc) · 6.19 KB
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
139
140
141
142
143
144
145
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
import httpx
import pytest
from kiota_http.middleware import MiddlewarePipeline, RedirectHandler, RetryHandler
from kiota_http.middleware.options import RedirectHandlerOption, RetryHandlerOption
from msgraph_core import APIVersion, GraphClientFactory, NationalClouds
from msgraph_core.middleware import AsyncGraphTransport, GraphTelemetryHandler
def test_create_with_default_middleware():
"""Test creation of GraphClient using default middleware"""
client = GraphClientFactory.create_with_default_middleware()
assert isinstance(client, httpx.AsyncClient)
assert isinstance(client._transport, AsyncGraphTransport)
pipeline = client._transport.pipeline
assert isinstance(pipeline, MiddlewarePipeline)
assert isinstance(pipeline._first_middleware, RedirectHandler)
assert isinstance(pipeline._current_middleware, GraphTelemetryHandler)
def test_create_with_default_middleware_custom_client():
"""Test creation of GraphClient using default middleware"""
timeout = httpx.Timeout(20, connect=10)
custom_client = httpx.AsyncClient(timeout=timeout, http2=True)
client = GraphClientFactory.create_with_default_middleware(client=custom_client)
assert isinstance(client, httpx.AsyncClient)
assert client.timeout == httpx.Timeout(connect=10, read=20, write=20, pool=20)
assert isinstance(client._transport, AsyncGraphTransport)
pipeline = client._transport.pipeline
assert isinstance(pipeline, MiddlewarePipeline)
assert isinstance(pipeline._first_middleware, RedirectHandler)
assert isinstance(pipeline._current_middleware, GraphTelemetryHandler)
def test_create_with_default_middleware_custom_client_with_proxy():
"""Test creation of GraphClient using default middleware"""
proxies = {
"http://": httpx.HTTPTransport(proxy="http://localhost:8030"),
"https://": httpx.HTTPTransport(proxy="http://localhost:8031"),
}
timeout = httpx.Timeout(20, connect=10)
custom_client = httpx.AsyncClient(timeout=timeout, http2=True, mounts=proxies)
client = GraphClientFactory.create_with_default_middleware(client=custom_client)
assert isinstance(client, httpx.AsyncClient)
assert client.timeout == httpx.Timeout(connect=10, read=20, write=20, pool=20)
assert isinstance(client._transport, AsyncGraphTransport)
pipeline = client._transport.pipeline
assert isinstance(pipeline, MiddlewarePipeline)
assert isinstance(pipeline._first_middleware, RedirectHandler)
assert isinstance(pipeline._current_middleware, GraphTelemetryHandler)
assert client._mounts
for pattern, transport in client._mounts.items():
assert isinstance(transport, AsyncGraphTransport)
def test_create_default_with_custom_middleware():
"""Test creation of HTTP Client using default middleware and custom options"""
retry_options = RetryHandlerOption(max_retries=5)
options = {f'{retry_options.get_key()}': retry_options}
client = GraphClientFactory.create_with_default_middleware(options=options)
assert isinstance(client, httpx.AsyncClient)
assert isinstance(client._transport, AsyncGraphTransport)
pipeline = client._transport.pipeline
assert isinstance(pipeline, MiddlewarePipeline)
assert isinstance(pipeline._first_middleware, RedirectHandler)
retry_handler = pipeline._first_middleware.next
assert isinstance(retry_handler, RetryHandler)
assert retry_handler.options.max_retry == retry_options.max_retry
assert isinstance(pipeline._current_middleware, GraphTelemetryHandler)
def test_create_with_custom_middleware_custom_client():
"""Test creation of HTTP Clients with custom middleware"""
timeout = httpx.Timeout(20, connect=10)
custom_client = httpx.AsyncClient(timeout=timeout, http2=True)
middleware = [
GraphTelemetryHandler(),
]
client = GraphClientFactory.create_with_custom_middleware(
middleware=middleware, client=custom_client
)
assert isinstance(client, httpx.AsyncClient)
assert client.timeout == httpx.Timeout(connect=10, read=20, write=20, pool=20)
assert isinstance(client._transport, AsyncGraphTransport)
pipeline = client._transport.pipeline
assert isinstance(pipeline._first_middleware, GraphTelemetryHandler)
def test_create_with_custom_middleware_custom_client_with_proxy():
"""Test creation of HTTP Clients with custom middleware"""
proxies = {
"http://": httpx.HTTPTransport(proxy="http://localhost:8030"),
"https://": httpx.HTTPTransport(proxy="http://localhost:8031"),
}
timeout = httpx.Timeout(20, connect=10)
custom_client = httpx.AsyncClient(timeout=timeout, http2=True, mounts=proxies)
middleware = [
GraphTelemetryHandler(),
]
client = GraphClientFactory.create_with_custom_middleware(
middleware=middleware, client=custom_client
)
assert isinstance(client, httpx.AsyncClient)
assert client.timeout == httpx.Timeout(connect=10, read=20, write=20, pool=20)
assert isinstance(client._transport, AsyncGraphTransport)
pipeline = client._transport.pipeline
assert isinstance(pipeline._first_middleware, GraphTelemetryHandler)
assert client._mounts
for pattern, transport in client._mounts.items():
assert isinstance(transport, AsyncGraphTransport)
pipeline = transport.pipeline
assert isinstance(pipeline._first_middleware, GraphTelemetryHandler)
def test_graph_client_factory_with_custom_configuration():
"""
Test creating a graph client with custom url overrides the default
"""
graph_client = GraphClientFactory.create_with_default_middleware(
api_version=APIVersion.beta, host=NationalClouds.China
)
assert isinstance(graph_client, httpx.AsyncClient)
assert str(graph_client.base_url) == f'{NationalClouds.China}/{APIVersion.beta}/'
def test_get_base_url():
"""
Test base url is formed by combining the national cloud endpoint with
Api version
"""
url = GraphClientFactory._get_base_url(
host=NationalClouds.Germany,
api_version=APIVersion.beta,
)
assert url == f'{NationalClouds.Germany}/{APIVersion.beta}'