add Mapping and Iterable interfaces to BaseObjectProperties

tests/test_base.py

@@ -0,0 +1,159 @@
+import random
+
+import attr
+import faker
+import pytest
+
+from udm_rest_client.udm import UDM
+
+fake = faker.Faker()
+
+
+@pytest.mark.asyncio
+async def test_base_obj_eq(user_created_via_http, udm_kwargs):
+    dn, url, user = user_created_via_http()
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj = await mod.get(dn)
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj2 = await mod.get(dn)
+
+    assert obj == obj2
+
+    for attri in ("dn", "policies", "position", "superordinate"):
+        ori_val = getattr(obj, attri)
+        setattr(obj, attri, fake.pystr())
+        assert obj != obj2
+        setattr(obj, attri, ori_val)
+        assert obj == obj2
+
+    opt = fake.pystr()
+    obj.options.append(opt)
+    assert obj != obj2
+    obj.options.remove(opt)
+    assert obj == obj2
+
+    attri = "firstname"
+    ori_val = getattr(obj.props, "firstname")
+    setattr(obj.props, attri, fake.pystr())
+    assert obj != obj2
+    setattr(obj.props, attri, ori_val)
+    assert obj == obj2
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("groups/group")
+        async for obj3 in mod.search():
+            break
+        assert obj3
+    assert obj != obj3
+
+
+@pytest.mark.asyncio
+async def test_base_obj_props_eq(user_created_via_http, udm_kwargs):
+    dn, url, user = user_created_via_http()
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj = await mod.get(dn)
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj2 = await mod.get(dn)
+
+    assert obj == obj2
+
+    for attri in ("firstname", "lastname", "username", "birthday"):
+        ori_val = getattr(obj.props, attri)
+        setattr(obj.props, attri, fake.pystr())
+        assert obj != obj2
+        setattr(obj.props, attri, ori_val)
+        assert obj == obj2
+
+    setattr(obj.props, "_underscore", fake.pystr())
+    assert obj.props == obj2.props
+    assert obj == obj2
+
+    missing_attr = random.choice(list(obj.props.keys()))
+    delattr(obj.props, missing_attr)
+    assert obj.props != obj2.props
+    assert obj != obj2
+
+
+@pytest.mark.asyncio
+async def test_base_obj_props_in(user_created_via_http, udm_kwargs):
+    dn, url, user = user_created_via_http()
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj = await mod.get(dn)
+    for attri in ("firstname", "lastname", "username", "birthday"):
+        assert attri in obj.props
+    assert fake.pystr() not in obj.props
+
+
+@pytest.mark.asyncio
+async def test_base_obj_props_getitem(user_created_via_http, udm_kwargs):
+    dn, url, user = user_created_via_http()
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj = await mod.get(dn)
+    assert user["properties"]["lastname"] == obj.props["lastname"]
+    assert user["properties"]["lastname"] == obj.props.get("lastname")
+    with pytest.raises(KeyError):
+        _ = obj.props[fake.pystr()]
+
+
+@pytest.mark.asyncio
+async def test_base_obj_props_setitem(user_created_via_http, udm_kwargs):
+    dn, url, user = user_created_via_http()
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj = await mod.get(dn)
+    new_val = fake.pystr()
+    obj.props["lastname"] = new_val
+    assert obj.props.lastname == new_val
+    with pytest.raises(TypeError):
+        obj.props[fake.pystr()] = fake.pystr()
+
+
+@pytest.mark.asyncio
+async def test_base_obj_props_iter_items_keys_values_len(
+    user_created_via_http, udm_kwargs
+):
+    dn, url, user = user_created_via_http()
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj = await mod.get(dn)
+    keys_exp = [k for k in obj.props.__dict__.keys() if not k.startswith("_")]
+    assert set(keys_exp) == set(obj.props.keys())  # keys
+    assert set(keys_exp) == set(k for k in obj.props)  # iter
+    assert len(keys_exp) == len(obj.props)  # len
+    for k in keys_exp:
+        assert getattr(obj.props, k) in obj.props.values()  # values
+        assert (k, getattr(obj.props, k)) in obj.props.items()  # items
+
+
+@pytest.mark.asyncio
+async def test_base_obj_props_update(user_created_via_http, fake_user, udm_kwargs):
+    dn, url, user = user_created_via_http()
+    new_user = fake_user()
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj = await mod.get(dn)
+    keys_exp = [k for k in obj.props.__dict__.keys() if not k.startswith("_")]
+    new_values = attr.asdict(new_user.props)
+    kv = list(new_values.items())[0]
+    kwargs = dict([kv])
+    del new_values[kv[0]]
+    obj.props.update(new_values, **kwargs)
+    for k in keys_exp:
+        if hasattr(new_user.props, k):
+            assert getattr(obj.props, k) == getattr(new_user.props, k)
+    assert getattr(obj.props, kv[0]) == kv[1]

tests/test_base_http.py

@@ -869,3 +869,25 @@ async def test_modify_users_self_redirects_to_users_user(
         mod = udm.get("users/user")
         obj = await mod.get(administrator_dn)
         assert obj.props.firstname == new_fn
+
+
+@pytest.mark.asyncio
+async def test_obj_eq(user_created_via_http, udm_kwargs):
+    dn, url, user = user_created_via_http()
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj = await mod.get(dn)
+
+    async with UDM(**udm_kwargs) as udm:
+        mod = udm.get("users/user")
+        obj2 = await mod.get(dn)
+
+    assert obj == obj2
+
+    for attri in ("uri", "uuid"):
+        ori_val = getattr(obj, attri)
+        setattr(obj, attri, fake.pystr())
+        assert obj != obj2
+        setattr(obj, attri, ori_val)
+        assert obj == obj2

tests/test_udm.py

@@ -61,6 +61,7 @@ async def load_obj_by_dn(udm, result):
         assert obj.dn == dn
         assert obj._udm_module.name == object_type
         assert obj.uuid == uuid
+        return obj
 
     with ldap_connection(connection_kwargs={"read_only": True}) as conn:
         logger.info("Successful LDAP login.")
@@ -72,7 +73,7 @@ async def load_obj_by_dn(udm, result):
             ")",
             attributes=["univentionObjectType", "univentionObjectFlag", "entryUUID"],
         )
-        all_objs = {}
+    all_objs = {}
     async with UDM(**udm_kwargs) as udm:
         # test one object per udm module
         for result in conn.entries:
@@ -86,16 +87,10 @@ async def load_obj_by_dn(udm, result):
         logger.info(
             "Reading %d objects of different UDM module types...", len(module_names)
         )
-        objs = await asyncio.gather(
-            *(
-                load_obj_by_dn(udm, random.choice(all_objs[module_name]))
-                for module_name in module_names
-            )
-        )
-        with open("/tmp/objs", "a") as fp:
-            for obj in objs:
-                fp.write(f"{obj!r}\n")
-        # TODO: what's this? all None!
+        entries = [random.choice(all_objs[module_name]) for module_name in module_names]
+        objs = await asyncio.gather(*(load_obj_by_dn(udm, entry) for entry in entries))
+        for entry, obj in zip(entries, objs):
+            assert entry.entry_dn == obj.dn
 
 
 def test_version():

udm_rest_client/base.py

@@ -30,32 +30,22 @@
 Base classes for (simplified) UDM modules and objects.
 """
 
+import collections
 import copy
 import pprint
-from collections import namedtuple
-from typing import Any, Dict, Iterable, Iterator, List, Union
+from typing import Any, Dict, Iterable, Iterator, List, Tuple, Union
 
-LdapMapping = namedtuple("LdapMapping", ("ldap2udm", "udm2ldap"))
+LdapMapping = collections.namedtuple("LdapMapping", ("ldap2udm", "udm2ldap"))
 
 
-class BaseObjectProperties:
+class BaseObjectProperties(collections.abc.Mapping, collections.abc.Iterable):
     """Container for UDM properties."""
 
     def __init__(self, udm_obj: "BaseObject") -> None:
         self._udm_obj = udm_obj
 
-    def __repr__(self) -> str:
-        return "{}({})".format(
-            self.__class__.__name__,
-            pprint.pformat(
-                dict(
-                    (k, v)
-                    for k, v in self.__dict__.items()
-                    if not str(k).startswith("_")
-                ),
-                indent=2,
-            ),
-        )
+    def __contains__(self, item) -> bool:
+        return hasattr(self, item)
 
     def __deepcopy__(
         self, memo: Dict[int, "BaseObjectProperties"]
@@ -70,6 +60,56 @@ def __deepcopy__(
                     setattr(memo[id_self], k, copy.deepcopy(v))
         return memo[id_self]
 
+    def __eq__(self, other: "BaseObjectProperties") -> bool:
+        # compare keys first (fast and low memory)
+        if set(self.keys()) != set(other.keys()):
+            return False
+        # compare values one at a time to reduce memory usage
+        for k, v in self.items():
+            if v != getattr(other, k):
+                return False
+        return True
+
+    def __getitem__(self, key):
+        try:
+            return getattr(self, key)
+        except AttributeError as exc:
+            raise KeyError(
+                f"{self.__class__.__name__} does not have key {key!r}."
+            ) from exc
+
+    def __iter__(self) -> Iterator[str]:
+        return (k for k in self.__dict__.keys() if not str(k).startswith("_"))
+
+    def __len__(self):
+        return len(list(self.keys()))
+
+    def __repr__(self) -> str:
+        return "{}({})".format(
+            self.__class__.__name__,
+            pprint.pformat(dict((k, v) for k, v in self.items()), indent=2,),
+        )
+
+    def __setitem__(self, key, value):
+        if key not in self:
+            raise TypeError(f"Assignment to non existent attribute {key!r} forbidden.")
+        setattr(self, key, value)
+
+    def items(self) -> Iterable[Tuple[str, Any]]:
+        return ((k, v) for k, v in self.__dict__.items() if not str(k).startswith("_"))
+
+    def keys(self) -> Iterable[str]:
+        return (k for k in iter(self))
+
+    def update(self, other: "BaseObjectProperties" = None, **kwargs) -> None:
+        for k in other or []:
+            self[k] = other[k]
+        for k in kwargs:
+            self[k] = kwargs[k]
+
+    def values(self) -> Iterable[Any]:
+        return (v for k, v in self.__dict__.items() if not str(k).startswith("_"))
+
 
 class BaseObject:
     """
@@ -121,6 +161,16 @@ def __init__(self):
         self.superordinate: str = None
         self._udm_module: BaseModule = None
 
+    def __eq__(self, other: "BaseObject") -> bool:
+        if self._udm_module.name != other._udm_module.name:
+            return False
+        for attr in ("dn", "props", "policies", "position", "superordinate"):
+            if getattr(self, attr) != getattr(other, attr):
+                return False
+        if set(self.options) != set(other.options):
+            return False
+        return True
+
     def __repr__(self) -> str:
         return "{}({!r}, {!r})".format(
             self.__class__.__name__,

udm_rest_client/base_http.py

@@ -498,11 +498,7 @@ class UdmObjectProperties(BaseObjectProperties):
     """Container for UDM properties."""
 
     def _to_dict(self) -> Dict[str, Any]:
-        return dict(
-            (k, _serialize_obj(v))
-            for k, v in self.__dict__.items()
-            if not str(k).startswith("_")
-        )
+        return dict((k, _serialize_obj(v)) for k, v in self.items())
 
 
 class UdmObject(BaseObject):
@@ -572,6 +568,14 @@ def __deepcopy__(self, memo: Dict[int, "UdmObject"]) -> "UdmObject":
             # _udm_module must be set in the current session
         return memo[id_self]
 
+    def __eq__(self, other: "UdmObject") -> bool:
+        if not super().__eq__(other):
+            return False
+        for attr in ("uri", "uuid"):
+            if getattr(self, attr) != getattr(other, attr):
+                return False
+        return True
+
     async def reload(self) -> "UdmObject":
         """
         Refresh object from LDAP.