You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

67 lines
2.3 KiB

"""Connection test module."""
import pytest
from keycloak.connection import ConnectionManager
from keycloak.exceptions import KeycloakConnectionError
def test_connection_proxy():
"""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():
"""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 == dict()
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():
"""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():
"""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():
"""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 method.startswith("a_")]
for method in sync_methods:
async_method = method[2:]
assert (async_method in con_methods) is True