from argparse import Namespace import asyncio from async_generator import asynccontextmanager import pytest from phi.ldap.async_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"]}] 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 return USER_LIST conn.search = _search yield conn @pytest.fixture def client_fixture(): return MockClient(BASE_DN) @pytest.fixture def client_fixture_multi(): res = Namespace() res.users = MockClient(BASE_DN) 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"" 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"" @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 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"" @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"" @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"" 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"" 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"" 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