Improved user, global role and inventory role handling and tests.

c8y_api/model/__init__.py

@@ -56,8 +56,10 @@
     'Fragment',
     'NamedObject',
     'User',
+    'TfaSettings',
     'CurrentUser',
     'GlobalRole',
+    'InventoryRole',
     'Permission',
     'ReadPermission',
     'WritePermission',

c8y_api/model/administration.py

@@ -54,9 +54,9 @@ class Scope(object):
         OPERATION = 'OPERATION'
 
     _parser = SimpleObjectParser({
-            'level': 'permission',
-            'type': 'type',
-            'scope': 'scope'})
+        'level': 'permission',
+        'type': 'type',
+        'scope': 'scope'})
 
     def __init__(self, level: str = Level.ANY, scope: str = Scope.ANY, type: str = '*'):
         """Create a new Permission instance.
@@ -90,20 +90,23 @@ def to_json(self, only_updated=False) -> dict:
 
 class ReadPermission(Permission):
     """Represents a read permission within Cumulocity."""
+
     # pylint: disable=abstract-method
     def __init__(self, scope=Permission.Scope.ANY, type='*'):  # noqa
         super().__init__(level=Permission.Level.READ, scope=scope, type=type)
 
 
 class WritePermission(Permission):
     """Represents a write permission within Cumulocity."""
+
     # pylint: disable=abstract-method
     def __init__(self, scope=Permission.Scope.ANY, type='*'):  # noqa
         super().__init__(level=Permission.Level.WRITE, scope=scope, type=type)
 
 
 class AnyPermission(Permission):
     """Represents a read/write permission within Cumulocity."""
+
     # pylint: disable=abstract-method
     def __init__(self, scope=Permission.Scope.ANY, type='*'):  # noqa
         super().__init__(level=Permission.Level.ANY, scope=scope, type=type)
@@ -120,8 +123,8 @@ class InventoryRole(SimpleObject):
     """
 
     _parser = SimpleObjectParser({
-            '_u_name': 'name',
-            '_u_description': 'description'})
+        '_u_name': 'name',
+        '_u_description': 'description'})
     _resource = '/user/inventoryroles'
 
     def __init__(self, c8y: CumulocityRestApi = None, name: str = None, description: str = None,
@@ -191,7 +194,7 @@ class InventoryRoleAssignment(SimpleObject):
     See also: https://cumulocity.com/api/#tag/Inventory-Roles
     """
     _parser = SimpleObjectParser({
-            'managed_object': 'managedObject'})
+        'managed_object': 'managedObject'})
 
     def __init__(self, c8y: CumulocityRestApi = None, managed_object: str = None,
                  roles: List[InventoryRole] = None):
@@ -237,9 +240,9 @@ class GlobalRole(SimpleObject):
     """
 
     _parser = SimpleObjectParser({
-            'id': 'id',
-            '_u_name': 'name',
-            '_u_description': 'description'})
+        'id': 'id',
+        '_u_name': 'name',
+        '_u_description': 'description'})
     _resource = 'INVALID'  # needs to be dynamically generated. see _build_resource_path
     _accept = CumulocityRestApi.ACCEPT_GLOBAL_ROLE
     _custom_properties_parser = ComplexObjectParser({}, [])
@@ -378,10 +381,10 @@ class TfaSettings:
     """TFA settings representation within Cumulocity."""
 
     _parser = SimpleObjectParser(
-            enabled='tfaEnabled',
-            enforced='tfaEnforced',
-            strategy='strategy',
-            last_request_time='lastTfaRequestTime')
+        enabled='tfaEnabled',
+        enforced='tfaEnforced',
+        strategy='strategy',
+        last_request_time='lastTfaRequestTime')
 
     def __init__(self,
                  enabled: bool = None,
@@ -441,27 +444,26 @@ def to_json(self) -> dict:
 
 class _BaseUser(SimpleObject):
     _parser = SimpleObjectParser({
-            'username': 'userName',
-            'password_strength': 'passwordStrength',
-            'owner': 'owner',
-            'delegated_by': 'delegatedBy',
-            '_u_email': 'email',
-            '_u_enabled': 'enabled',
-            '_u_display_name': 'displayName',
-            '_u_password': 'password',
-            '_u_first_name': 'firstName',
-            '_u_last_name': 'lastName',
-            '_u_tfa_enabled': 'twoFactorAuthenticationEnabled',
-            '_u_require_password_reset': 'shouldResetPassword',
-            '_password_reset_mail': 'sendPasswordResetEmail',
-            '_last_password_change': 'lastPasswordChange'})
+        'username': 'userName',
+        'password_strength': 'passwordStrength',
+        'owner': 'owner',
+        'delegated_by': 'delegatedBy',
+        '_u_email': 'email',
+        '_u_enabled': 'enabled',
+        '_u_display_name': 'displayName',
+        '_u_password': 'password',
+        '_u_first_name': 'firstName',
+        '_u_last_name': 'lastName',
+        '_u_tfa_enabled': 'twoFactorAuthenticationEnabled',
+        '_u_require_password_reset': 'shouldResetPassword',
+        '_password_reset_mail': 'sendPasswordResetEmail',
+        '_last_password_change': 'lastPasswordChange'})
     _resource = 'INVALID'  # needs to be dynamically generated. see _build_resource_path
 
     def __init__(self, c8y: CumulocityRestApi = None, username: str = None, email: str = None,
                  enabled: bool = True, display_name: str = None, password: str = None,
                  first_name: str = None, last_name: str = None, phone: str = None,
                  tfa_enabled: bool = None, require_password_reset: bool = None):
-
         super().__init__(c8y)
         self.username = username
         self.password_strength = None
@@ -548,7 +550,6 @@ def __init__(self, c8y=None, username=None, email=None, enabled=True, display_na
         # self.effective_permission_ids = set()
         # self.custom_properties = WithUpdatableFragments()
 
-
     @classmethod
     def from_json(cls, json: dict) -> User:
         user = cls._from_json(json, User())
@@ -733,6 +734,7 @@ class CurrentUser(_BaseUser):
 
     class TotpActivity:
         """User's TOTP activity information."""
+
         def __init__(self, is_active: bool = None):
             self.is_active = is_active
 
@@ -769,13 +771,13 @@ def to_json(self) -> dict:
     _resource = '/user/currentUser'
     _accept = CumulocityRestApi.ACCEPT_CURRENT_USER
 
-    def __init__(self, c8y:CumulocityRestApi = None):
+    def __init__(self, c8y: CumulocityRestApi = None):
         super().__init__(c8y)
         self.effective_permission_ids = {}
 
     @classmethod
     def from_json(cls, json: dict) -> CurrentUser:
-        user:CurrentUser = cls._from_json(json, CurrentUser())
+        user: CurrentUser = cls._from_json(json, CurrentUser())
         if 'effectiveRoles' in json:
             user.effective_permission_ids = {ref['id'] for ref in json['effectiveRoles']}
         return user
@@ -1057,23 +1059,44 @@ def get_current(self) -> CurrentUser:
         user.c8y = self.c8y
         return user
 
-    def select(self,
-               username: str = None,
-               groups: str | int | GlobalRole | List[str] | List[int] | List[GlobalRole] = None,
-               page_size: int = 5):
+    def select(
+            self,
+            expression: str = None,
+            username: str = None,
+            groups: str | int | GlobalRole | List[str] | List[int] | List[GlobalRole] = None,
+            owner: str = None,
+            only_devices: bool = None,
+            with_subusers_count: bool = None,
+            limit: int = None,
+            page_size: int = 5,
+            page_number: int = None,
+            **kwargs
+    ) -> Generator[User]:
         """Lazily select and yield User instances.
 
         The result can be limited by username (prefix) and/or group membership.
 
         Args:
+            expression (str):  Arbitrary filter expression which will be
+                passed to Cumulocity without change; all other filters
+                are ignored if this is provided
             username (str): A user's username or a prefix thereof
             groups (int, [int], str, [str], GlobalRole, [GlobalRole]): a scalar
                 or list of int (actual group ID), string (group names), or actual
                 Group instances
-            page_size (int):  Number of results fetched per request
+            owner (str):  Username of the owner of the user
+            only_devices (bool):  Only return device users (starting with `device_`)
+                If absent or False, the users will be excluded.
+            with_subusers_count (bool):  Whether to include an additional field
+             `subusersCount` which holds the number of direct sub users.
+            limit (int): Limit the number of results to this number.
+            page_size (int): Define the number of events which are read (and
+                parsed in one chunk). This is a performance related setting.
+            page_number (int): Pull a specific page; this effectively disables
+                automatic follow-up page retrieval.
 
         Returns:
-            Generator of Group instances
+            Generator of User instances
         """
         # group_list can be ints, strings (names) or Group objects
         # it needs to become a comma-separated string
@@ -1092,36 +1115,46 @@ def select(self,
                 raise ValueError("Unable to identify type of given group identifiers.")
             groups_string = ','.join(groups_string)
         # lazily yield parsed objects page by page
-        base_query = super()._build_base_query(username=username, groups=groups_string, page_size=page_size)
-        page_number = 1
-        while True:
-            page_results = [User.from_json(x) for x in self._get_page(base_query, page_number)]
-            if not page_results:
-                break
-            for user in page_results:
-                user.c8y = self.c8y  # inject c8y connection into instance
-                yield user
-            page_number = page_number + 1
-
-    def get_all(self,
-                username: str = None,
-                groups: str | int | GlobalRole | List[str] | List[int] | List[GlobalRole] = None,
-                page_size: int = 1000):
+        base_query = super()._prepare_query(
+            expression=expression,
+            username=username,
+            groups=groups_string,
+            owner=owner,
+            only_devices=only_devices,
+            with_subusers_count=with_subusers_count,
+            page_size=page_size,
+
+            **kwargs
+        )
+        return super()._iterate(base_query, page_number, limit, User.from_json)
+
+    def get_all(
+            self,
+            username: str = None,
+            groups: str | int | GlobalRole | List[str] | List[int] | List[GlobalRole] = None,
+            owner: str = None,
+            only_devices: bool = None,
+            with_subusers_count: bool = None,
+            page_size: int = 1000,
+            **kwargs
+    ) -> list[User]:
         """Select and retrieve User instances as list.
 
         The result can be limited by username (prefix) and/or group membership.
 
-        Args:
-            username (str): A user's username or a prefix thereof
-            groups: a scalar or list of int (actual group ID), string (group names),
-                or actual Group instances
-            page_size (int):  Maximum number of entries fetched per requests;
-            this is a performance setting
+        See `select` for a documentation of arguments.
 
         Returns:
-            List of User
+            List of User instances
         """
-        return list(self.select(username, groups, page_size))
+        return list(self.select(
+            username=username,
+            groups=groups,
+            owner=owner,
+            only_devices=only_devices,
+            with_subusers_count=with_subusers_count,
+            page_size=page_size,
+            **kwargs))
 
     def create(self, *users):
         """Create users within the database.
@@ -1257,6 +1290,8 @@ def select(self, username: str = None, page_size: int = 5) -> Generator[GlobalRo
         Return:
             Generator of GlobalRole instances
         """
+        # unfortunately, as selecting by username can't be implemented using the
+        # generic _iterate method, we have to do everything manually.
         if username:
             # select by username
             query = f'/user/{self.c8y.tenant_id}/users/{username}/groups?pageSize={page_size}&currentPage='
@@ -1273,17 +1308,8 @@ def select(self, username: str = None, page_size: int = 5) -> Generator[GlobalRo
                 page_number = page_number + 1
         else:
             # select all
-            query = self._prepare_query(page_size=page_size)
-            page_number = 1
-            while True:
-                role_jsons = self._get_page(query, page_number)
-                if not role_jsons:
-                    break
-                for role_json in role_jsons:
-                    result = GlobalRole.from_json(role_json)
-                    result.c8y = self.c8y
-                    yield result
-                page_number = page_number + 1
+            base_query = self._prepare_query(page_size=page_size)
+            yield from super()._iterate(base_query, page_number=None, limit=None, parse_fun=GlobalRole.from_json)
 
     def get_all(self, username: str = None, page_size: int = 1000) -> List[GlobalRole]:
         """Retrieve global roles.
@@ -1297,7 +1323,7 @@ def get_all(self, username: str = None, page_size: int = 1000) -> List[GlobalRol
         Return:
             List of GlobalRole instances
         """
-        return list(self.select(username, page_size))
+        return list(self.select(username=username, page_size=page_size))
 
     def assign_users(self, role_id: int | str, *usernames: str):
         """Add users to a global role.

c8y_tk/notification2/listener.py

@@ -130,7 +130,7 @@ async def _callback(msg):
                 c = await self._get_connection()
                 payload = await c.recv()
                 self._log.debug("Received message: {}.", payload)
-                asyncio.create_task(_callback(AsyncListener.Message(listener=self, payload=payload)))
+                await asyncio.create_task(_callback(AsyncListener.Message(listener=self, payload=payload)))
             except ws.ConnectionClosed as e:
                 self._log.info("Websocket connection closed: {}", e)
 

integration_tests/test_global_roles.py

@@ -6,10 +6,12 @@
 
 from __future__ import annotations
 
+import random
+
 import pytest
 
 from c8y_api import CumulocityApi
-from c8y_api.model import GlobalRole
+from c8y_api.model import GlobalRole, User
 
 from util.testing_util import RandomNameGenerator
 
@@ -45,6 +47,24 @@ def test_CRUD(live_c8y: CumulocityApi):  # noqa (case)
         assert rolename in str(e)
 
 
+def test_select(live_c8y: CumulocityApi):
+    """Verify that selection works as expected."""
+    # (1) get all defined global roles
+    all_roles = live_c8y.global_roles.get_all()
+
+    # (2) create a user and assign roles
+    username = RandomNameGenerator.random_name(2)
+    email = f'{username}@c8y.com'
+    user = User(live_c8y, username=username, email=email, enabled=True).create()
+    selected_roles = random.sample(all_roles, k=5)
+    for role in selected_roles:
+        user.assign_global_role(role.id)
+
+    # (3) select by user
+    for role in live_c8y.global_roles.get_all(username=username):
+        assert role.id in [x.id for x in selected_roles]
+
+
 def test_updating_users(live_c8y: CumulocityApi, factory):
     """Verify that users can be added/removed to/from a global role."""
 

integration_tests/test_inventoryroles.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from c8y_api import CumulocityApi
 from c8y_api.model import User, InventoryRole, Permission, ReadPermission, WritePermission, AnyPermission
 
 from util.testing_util import RandomNameGenerator
@@ -79,6 +80,12 @@ def test_CRUD2(live_c8y):
         live_c8y.inventory_roles.get(created_role.id)
 
 
+def test_select_inventory_roles(live_c8y: CumulocityApi):
+    """Verify that selection works as expected."""
+    # (1) get all defined inventory roles
+    assert live_c8y.inventory_roles.get_all()
+
+
 def test_assignments(live_c8y, sample_device, factory):
     """Verify that inventory roles can be assigned, retrieved and unassigned."""
     email = 'user_' + RandomNameGenerator.random_name(2) + '@test.com'