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
|
"""Connection test module."""
from inspect import iscoroutinefunction, signature
import pytest
from keycloak.connection import ConnectionManager
from keycloak.exceptions import KeycloakConnectionError
def test_connection_proxy() -> None:
"""Test proxies of connection manager."""
cm = ConnectionManager(
base_url="http://test.test",
proxies={"http://test.test": "http://localhost:8080"},
)
assert cm._s.proxies == {"http://test.test": "http://localhost:8080"}
def test_headers() -> None:
"""Test headers manipulation."""
cm = ConnectionManager(base_url="http://test.test", headers={"H": "A"})
assert cm.param_headers(key="H") == "A"
assert cm.param_headers(key="A") is None
cm.clean_headers()
assert cm.headers == {}
cm.add_param_headers(key="H", value="B")
assert cm.exist_param_headers(key="H")
assert not cm.exist_param_headers(key="B")
cm.del_param_headers(key="H")
assert not cm.exist_param_headers(key="H")
def test_bad_connection() -> None:
"""Test bad connection."""
cm = ConnectionManager(base_url="http://not.real.domain")
with pytest.raises(KeycloakConnectionError):
cm.raw_get(path="bad")
with pytest.raises(KeycloakConnectionError):
cm.raw_delete(path="bad")
with pytest.raises(KeycloakConnectionError):
cm.raw_post(path="bad", data={})
with pytest.raises(KeycloakConnectionError):
cm.raw_put(path="bad", data={})
@pytest.mark.asyncio
async def a_test_bad_connection() -> None:
"""Test bad connection."""
cm = ConnectionManager(base_url="http://not.real.domain")
with pytest.raises(KeycloakConnectionError):
await cm.a_raw_get(path="bad")
with pytest.raises(KeycloakConnectionError):
await cm.a_raw_delete(path="bad")
with pytest.raises(KeycloakConnectionError):
await cm.a_raw_post(path="bad", data={})
with pytest.raises(KeycloakConnectionError):
await cm.a_raw_put(path="bad", data={})
def test_counter_part() -> None:
"""Test that each function has its async counter part."""
con_methods = [
func for func in dir(ConnectionManager) if callable(getattr(ConnectionManager, func))
]
sync_methods = [
method
for method in con_methods
if not method.startswith("a_") and not method.startswith("_")
]
async_methods = [
method for method in con_methods if iscoroutinefunction(getattr(ConnectionManager, method))
]
for method in sync_methods:
if method in [
"aclose",
"add_param_headers",
"del_param_headers",
"clean_headers",
"exist_param_headers",
"param_headers",
]:
continue
async_method = f"a_{method}"
assert (async_method in con_methods) is True
sync_sign = signature(getattr(ConnectionManager, method))
async_sign = signature(getattr(ConnectionManager, async_method))
assert sync_sign.parameters == async_sign.parameters
for async_method in async_methods:
if async_method in ["aclose"]:
continue
if async_method[2:].startswith("_"):
continue
assert async_method[2:] in sync_methods
|
