phi/async_tests/test_async_ldap_model.py

# -*- 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}>"


@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}>"


@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}>"


@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_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_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_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