phi/test/aux_async_model.py

# -*- coding: utf-8 -*-

import asyncio
from pprint import pprint as pp

from phi.async_ldap.new_model import (
    Hackers,
    User,
    Robots,
    Service,
    Group,
    Congregations,
)
from phi.async_ldap.mixins import build_heritage
from phi.async_ldap.client import AsyncClient
import phi.exceptions as e


async def dlv(h, cls):
    return [el async for el in build_heritage(h, cls)]


cl = AsyncClient(
    "ldap://localhost",
    port=389,
    encryption=True,
    #    validate=True,
    ca_cert="/home/leo/Documents/coding/phi/openldap/cert.pem",
    username="root",
    password="root",
    base_dn="dc=unit,dc=macaomilano,dc=org",
    attribute_id="cn",
)


async def get_all_children():
    h = Hackers(cl)
    r = Robots(cl)
    g = Congregations(cl)

    hackers = await dlv(h, User)
    robots = await dlv(r, Service)
    groups = await dlv(g, Group)

    return (hackers, robots, groups)


async def get_members(group):
    return [el async for el in group]


async def print_async(label, awaitable):
    print(label)
    result = await awaitable
    pp(result)


async def describe(obj):
    return await obj.describe()


async def _await(awaitable):
    return await awaitable


def sync_await(awaitable):
    return asyncio.run(_await(awaitable))


h = Hackers(cl)
r = Robots(cl)
c = Congregations(cl)


# asyncio.run(print_async("hackers:", describe(h)))
# asyncio.run(print_async("conte_mascetti:", describe(User(cl, "conte_mascetti"))))
# asyncio.run(print_async("robots:", describe(r)))
# asyncio.run(print_async("phi:", describe(Service(cl, "phi"))))
# asyncio.run(print_async("congregations:", describe(c)))
# asyncio.run(print_async("GitUsers:", describe(Group(cl, "GitUsers"))))

# asyncio.run(print_async("Hackers members:", get_members(h)))
# asyncio.run(print_async("Robots members:", get_members(r)))
# asyncio.run(print_async("Congregations members:", get_members(c)))

#
async def add_new(obj, name, **kw):
    try:
        _new = obj(cl, name, **kw)
        await _new.save()
    except e.PhiEntryExists as err:
        print(f"Failed add: {repr(err)}")


async def safe_search(group, name):
    try:
        res = await group.search(name)
        print("Search result:", res)
        return res
    except e.PhiEntryDoesNotExist as err:
        print(f"Failed search: {repr(err)}")


async def safe_delete(obj, cascade=None):
    try:
        if cascade:
            obj.delete_cascade = cascade
        await obj.delete()
    except Exception as err:
        print(f"Failed delete: {repr(err)}")


async def add_member(group, member):
    await group.add_member(member)


async def remove_member(group, member):
    await group.remove_member(member)


# asyncio.run(safe_search(h, "pippo"))
# asyncio.run(
#     add_new(User, "pippo", cn="Pippo (Goofy)", sn="Pippo", mail="pippo@unit.info")
# )
# asyncio.run(safe_search(h, "pippo"))
# asyncio.run(
#     add_new(User, "pippo", cn="Pippo (Goofy)", sn="Pippo", mail="pippo@unit.net")
# )
# asyncio.run(safe_delete(asyncio.run(safe_search(h, "pippo"))))
# asyncio.run(print_async("Hackers members:", get_members(h)))
# asyncio.run(safe_delete(h))
# asyncio.run(print_async("Hackers members:", get_members(h)))
# asyncio.run(safe_delete(h, True))
# asyncio.run(print_async("Hackers members:", get_members(h)))

# asyncio.run(safe_search(r, "phi"))
# asyncio.run(print_async("Robots members:", get_members(r)))
# asyncio.run(add_new(Service, "db", userPassword="lolpassword"))
# asyncio.run(print_async("Robots members:", get_members(r)))

asyncio.run(safe_search(c, "GitUsers"))
asyncio.run(print_async("Congregations members:", get_members(c)))
asyncio.run(
    add_new(Group, "naughty", member=[User(cl, "conte_mascetti"), User(cl, "necchi")])
)
asyncio.run(print_async("Congregations members:", get_members(c)))
asyncio.run(safe_delete(Group(cl, "naughty")))
asyncio.run(print_async("Congregations members:", get_members(c)))
asyncio.run(
    add_new(Group, "naughty", member=[User(cl, "conte_mascetti"), User(cl, "necchi")])
)
asyncio.run(print_async("Congregations members:", get_members(c)))
print("==> HERE <==")
naughty = sync_await(Group(cl, "naughty").sync())
print("NAUGHTY =>>", [m for m in naughty.get_members()])
asyncio.run(add_member(naughty, User(cl, "perozzi")))
print("NAUGHTY =>>", [m for m in naughty.get_members()])
asyncio.run(remove_member(naughty, User(cl, "conte_mascetti")))
print("NAUGHTY =>>", [m for m in naughty.get_members()])