phi/test/aux_async_model.py

160 lines
4.3 KiB
Python

# -*- coding: utf-8 -*-
import asyncio
from pprint import pprint as pp
from phi.async_ldap.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()])