phi/async_tests/test_async_ldap_model.py
2019-07-01 22:48:04 +02:00

469 lines
11 KiB
Python

# -*- encoding: utf-8 -*-
from argparse import Namespace
import asyncio
from async_generator import asynccontextmanager
import pytest
from phi.async_ldap.model import (
get_class,
recall,
call_if_callable,
inheritance,
iter_children,
Entry,
build_heritage,
Hackers,
Robots,
Congregations,
Service,
User,
Group,
)
BASE_DN = "dc=test,dc=abbiamoundominio,dc=org"
USER_LIST = [{"uid": ["conte_mascetti"]}, {"uid": ["perozzi"]}, {"uid": ["necchi"]}]
SERVICE_LIST = [{"uid": ["phi"]}, {"uid": ["irc"]}]
GROUP_LIST = [{"cn": ["amici_miei"]}, {"cn": ["antani"]}]
EXISTING_USER = [
{
"dn": f"uid=existing_user,{BASE_DN}>",
"objectClass": ["inetOrgPerson", "organizationalPerson", "person", "top"],
"cn": ["exists"],
"sn": ["Existing User"],
"mail": ["existing@mail.org"],
"uid": ["existing_user"],
"userPassword": ["{SHA}oLY7P6V+DWAMJ9ix7vbMYGIfA+E="],
}
]
class MockClient:
def __init__(self, base_dn, *args):
self._base_dn = base_dn
self.args = args
@property
def base_dn(self):
return self._base_dn
@asynccontextmanager
async def connect(self, *args, **kwargs):
conn = Namespace()
async def _search(*args, **kwargs):
if "Services" in self.args:
return SERVICE_LIST
elif "Groups" in self.args:
return GROUP_LIST
elif "Users" in self.args or f"None=Entry,{BASE_DN}" in args:
return USER_LIST
if f"uid=existing_user,ou=Hackers,{BASE_DN}" in args:
return EXISTING_USER
elif f"uid=not_existing,ou=Hackers,{BASE_DN}" in args:
return []
conn.search = _search
async def _add(*args, **kwargs):
return
conn.add = _add
yield conn
@pytest.fixture
def client_fixture():
return MockClient(BASE_DN)
@pytest.fixture
def client_fixture_multi():
res = Namespace()
res.users = MockClient(BASE_DN, "Users")
res.services = MockClient(BASE_DN, "Services")
res.groups = MockClient(BASE_DN, "Groups")
return res
@pytest.fixture
def lineage_fixture():
class Grand:
@classmethod
def name(cls):
return cls.__name__
class Ma(Grand):
pass
class Child(Ma):
pass
grand = Grand()
ma = Ma()
child = Child()
return Grand, Ma, Child, grand, ma, child
def test_get_class():
c = MockClient(BASE_DN)
assert get_class(c) is MockClient
def test_recall(lineage_fixture):
Grand, Ma, Child, grand, ma, child = lineage_fixture
assert object is recall(grand)
assert recall(grand) is recall(Grand)
assert Grand is recall(ma)
assert recall(ma) is recall(Ma)
assert Ma is recall(child)
assert recall(child) is recall(Child)
def test_call_if_callable():
class Dummy:
classattr = "classattr"
@property
def prop(self):
return "prop"
def func(self):
return "func"
@classmethod
def clsmth(cls):
return "clsmth"
d = Dummy()
assert "classattr" == call_if_callable(d, "classattr")
assert "prop" == call_if_callable(d, "prop")
assert "func" == call_if_callable(d, "func")
assert "clsmth" == call_if_callable(d, "clsmth")
def test_inheritance(lineage_fixture):
Grand, Ma, Child, _, _, _ = lineage_fixture
assert inheritance(Grand, "name") == "Grand"
assert inheritance(Ma, "name") == "Ma,Grand"
assert inheritance(Child, "name") == "Child,Ma,Grand"
assert inheritance(Child, "name", Grand) == "Child,Ma"
assert inheritance(Child, "name", Ma) == "Child"
@pytest.mark.asyncio
async def test_iter_children():
LIST = [1, 2, 3, 4]
async def _async_gen():
for i in LIST:
yield i
ALIST = _async_gen()
assert LIST == await iter_children(ALIST)
def test_Entry(client_fixture):
e = Entry(client_fixture)
assert e.base_dn == BASE_DN
assert e.client is not None
def test_Entry_name(client_fixture):
e = Entry(client_fixture)
assert e.name() == "Entry"
def test_Entry_qualified_name(client_fixture):
e = Entry(client_fixture)
assert e.qualified_name() == "None=Entry"
def test_Entry_dn(client_fixture):
e = Entry(client_fixture)
assert e.dn == "None=Entry,{}".format(BASE_DN)
@pytest.mark.asyncio
async def test_Entry_describe(client_fixture):
e = Entry(client_fixture)
assert USER_LIST == await e.describe()
@pytest.mark.asyncio
async def test_build_heritage(client_fixture):
class MockAIterable(object):
client = None
def __init__(self, elements):
self.elements = elements
async def get_children(self):
for el in self.elements:
yield el
class Dummy(object):
def __init__(self, *args, **kwargs):
self.id = sum([id(el) for el in args]) + sum(
[id(v) for _, v in kwargs.items()]
)
def __repr__(self):
return f"<Dummy({self.id})>"
def __eq__(self, other):
return self.id == other.id
m = MockAIterable(USER_LIST)
res = [el async for el in build_heritage(m, Dummy, attribute_id="uid")]
exp_res = [Dummy(el["uid"][0], None) for el in USER_LIST]
assert res == exp_res
def test_Hackers(client_fixture_multi):
h = Hackers(client_fixture_multi.users)
assert h.kind == "ou"
assert h.dn == "ou={},{}".format(h.name(), BASE_DN)
assert repr(h) == f"<Hackers {h.kind}=Hackers,{BASE_DN}>"
def test_Hackers_singleton(client_fixture):
other_client = MockClient(BASE_DN)
h1 = Hackers(client_fixture)
h2 = Hackers(other_client)
h3 = Hackers(client_fixture)
assert client_fixture is not other_client
assert h1 is h3
assert h2 is not h1
@pytest.mark.asyncio
async def test_Entry_get_children(client_fixture_multi):
h = Hackers(client_fixture_multi.users)
assert USER_LIST == [el async for el in h.get_children()]
@pytest.mark.asyncio
async def test_Hackers_anext(client_fixture_multi):
h = Hackers(client_fixture_multi.users)
exp_res = [User(el["uid"][0], client_fixture_multi.users) for el in USER_LIST]
assert exp_res == [el async for el in h]
def test_Hackers_children(client_fixture_multi):
h = Hackers(client_fixture_multi.users)
assert USER_LIST == h.children
@pytest.mark.asyncio
async def test_Hackers_get_by_attr(client_fixture):
h = Hackers(client_fixture)
res = await h.get_by_attr("uid", "existing_user")
assert len(res) == 1
assert res[0] is User("existing_user", client_fixture)
@pytest.mark.asyncio
async def test_Hackers_get_by_attr_empty(client_fixture):
h = Hackers(client_fixture)
res = await h.get_by_attr("uid", "not_existing")
assert res is None
@pytest.mark.asyncio
async def test_Hackers_get_by_uid(client_fixture):
h = Hackers(client_fixture)
res = await h.get_by_uid("existing_user")
assert res is User("existing_user", client_fixture)
def test_Robots(client_fixture_multi):
r = Robots(client_fixture_multi.services)
assert r.kind == "ou"
assert r.dn == "ou={},{}".format(r.name(), BASE_DN)
assert repr(r) == f"<Services {r.kind}=Services,{BASE_DN}>"
def test_Robots_singleton(client_fixture):
other_client = MockClient(BASE_DN)
r1 = Robots(client_fixture)
r2 = Robots(other_client)
r3 = Robots(client_fixture)
assert client_fixture is not other_client
assert r1 is r3
assert r2 is not r1
@pytest.mark.asyncio
async def test_Robots_anext(client_fixture_multi):
r = Robots(client_fixture_multi.services)
exp_res = [
Service(el["uid"][0], client_fixture_multi.services) for el in SERVICE_LIST
]
assert exp_res == [el async for el in r]
def test_Robots_children(client_fixture_multi):
r = Robots(client_fixture_multi.services)
assert SERVICE_LIST == r.children
def test_Congregations(client_fixture_multi):
g = Congregations(client_fixture_multi.groups)
assert g.kind == "ou"
assert g.dn == "ou={},{}".format(g.name(), BASE_DN)
assert repr(g) == f"<Groups {g.kind}=Groups,{BASE_DN}>"
def test_Congregations_singleton(client_fixture):
other_client = MockClient(BASE_DN)
g1 = Congregations(client_fixture)
g2 = Congregations(other_client)
g3 = Congregations(client_fixture)
assert client_fixture is not other_client
assert g1 is g3
assert g2 is not g1
@pytest.mark.asyncio
async def test_Congregations_anext(client_fixture_multi):
c = Congregations(client_fixture_multi.groups)
exp_res = [Group(el["cn"][0], client_fixture_multi.groups) for el in GROUP_LIST]
assert exp_res == [el async for el in c]
def test_Congregations_children(client_fixture_multi):
c = Congregations(client_fixture_multi.groups)
assert GROUP_LIST == c.children
def test_User(client_fixture):
c = User("conte_mascetti", client_fixture)
assert c.kind == "uid"
assert c.name == "conte_mascetti"
assert c.dn == "uid=conte_mascetti,ou=Hackers,{}".format(BASE_DN)
assert repr(c) == f"<User({c.name}) {c.dn}>"
def test_User_unsettable_name(client_fixture):
c = User("conte_mascetti", client_fixture)
with pytest.raises(RuntimeError) as e:
c.name = "totò"
assert "Name property is not modifiable." in str(e)
def test_User_singleton(client_fixture):
other_client = MockClient(BASE_DN)
c1 = User("conte_mascetti", client_fixture)
c2 = User("perozzi", client_fixture)
c3 = User("conte_mascetti", client_fixture)
c4 = User("conte_mascetti", other_client)
assert client_fixture is not other_client
assert c1 is c3
assert c2 is not c1
assert c4 is not c1
def test_Service(client_fixture):
c = Service("phi", client_fixture)
assert c.kind == "uid"
assert c.name == "phi"
assert c.dn == "uid=phi,ou=Services,{}".format(BASE_DN)
assert repr(c) == f"<Service({c.name}) {c.dn}>"
def test_Service_unsettable_name(client_fixture):
c = Service("phi", client_fixture)
with pytest.raises(RuntimeError) as e:
c.name = "theta"
assert "Name property is not modifiable." in str(e)
def test_Service_singleton(client_fixture):
other_client = MockClient(BASE_DN)
c1 = Service("phi", client_fixture)
c2 = Service("irc", client_fixture)
c3 = Service("phi", client_fixture)
c4 = Service("phi", other_client)
assert client_fixture is not other_client
assert c1 is c3
assert c2 is not c1
assert c4 is not c1
def test_Group(client_fixture):
c = Group("amici_miei", client_fixture)
assert c.kind == "cn"
assert c.name == "amici_miei"
assert c.dn == "cn=amici_miei,ou=Groups,{}".format(BASE_DN)
assert repr(c) == f"<Group({c.name}) {c.dn}>"
def test_Group_unsettable_name(client_fixture):
c = Group("amici_miei", client_fixture)
with pytest.raises(RuntimeError) as e:
c.name = "nemici"
assert "Name property is not modifiable." in str(e)
def test_Group_singleton(client_fixture):
other_client = MockClient(BASE_DN)
c1 = Group("amici_miei", client_fixture)
c2 = Group("antani", client_fixture)
c3 = Group("amici_miei", client_fixture)
c4 = Group("amici_miei", other_client)
assert client_fixture is not other_client
assert c1 is c3
assert c2 is not c1
assert c4 is not c1