Using lowercase boolean values; Adding references to get function of inventory API.

chsou · chsou · commit 7a888013f848 · 2025-08-01T16:00:27.000+02:00
diff --git a/c8y_api/model/_base.py b/c8y_api/model/_base.py
@@ -685,6 +685,11 @@ def _map_params(
         def multi(*xs):
             return sum(bool(x) for x in xs) > 1
 
+        def stringify(value):
+            if isinstance(value, bool):
+                return str(value).lower()
+            return value
+
         if multi(min_age, before, date_to):
             raise ValueError("Only one of 'min_age', 'before' and 'date_to' query parameters must be used.")
         if multi(max_age, after, date_from):
@@ -739,11 +744,11 @@ def multi(*xs):
             'createdTo': created_to,
             'lastUpdatedFrom': updated_from,
             'lastUpdatedTo': updated_to,
-            'withSourceAssets': with_source_assets,
-            'withSourceDevices': with_source_devices,
-            'revert': str(reverse).lower() if reverse is not None else None,
+            'withSourceAssets': stringify(with_source_assets),
+            'withSourceDevices': stringify(with_source_devices),
+            'revert': stringify(reverse),
             'pageSize': page_size}.items() if v is not None}
-        params.update({_StringUtil.to_pascal_case(k): v for k, v in kwargs.items() if v is not None})
+        params.update({_StringUtil.to_pascal_case(k): stringify(v) for k, v in kwargs.items() if v is not None})
         tuples = list(params.items())
         if series:
             if isinstance(series, list):
@@ -758,8 +763,9 @@ def _prepare_query(self, resource: str = None, expression: str = None, **kwargs)
             return resource or self.resource
         return (resource or self.resource) + '?' + encoded
 
-    def _get_object(self, object_id):
-        return self.c8y.get(self.build_object_path(object_id))
+    def _get_object(self, object_id, **kwargs):
+        query = self._prepare_query(self.build_object_path(object_id), **kwargs)
+        return self.c8y.get(query)
 
     def _get_page(self, base_query: str, page_number: int):
         sep = '&' if '?' in base_query else '?'
diff --git a/c8y_api/model/inventory.py b/c8y_api/model/inventory.py
@@ -23,19 +23,46 @@ class Inventory(CumulocityResource):
     def __init__(self, c8y):
         super().__init__(c8y, 'inventory/managedObjects')
 
-    def get(self, id) -> ManagedObject:  # noqa (id)
+    def get(
+            self,
+            id: str,  # noqa
+            with_children: bool = None,
+            with_children_count: bool = None,
+            skip_children_names: bool = None,
+            with_parents: bool = None,
+            with_latest_values: bool = None,
+            **kwargs) -> ManagedObject:
         """ Retrieve a specific managed object from the database.
 
         Args:
-            ID of the managed object
+            id (str): Cumulocity ID of the managed object
+            with_children (bool):  Whether children with ID and name should be
+                included with each returned object
+            with_children_count (bool): When set to true, the returned result
+                will contain the total number of children in the respective
+                child additions, assets and devices sub fragments.
+            skip_children_names (bool):  If true, returned references of child
+                devices won't contain their names.
+            with_parents (bool): Whether to include a device's parents.
+            with_latest_values (bool):  If true the platform includes the
+                fragment `c8y_LatestMeasurements, which contains the latest
+                measurement values reported by the device to the platform.
 
         Returns:
              A ManagedObject instance
 
         Raises:
             KeyError:  if the ID is not defined within the database
         """
-        managed_object = ManagedObject.from_json(self._get_object(id))
+        managed_object = ManagedObject.from_json(self._get_object(
+            id,
+            with_children=with_children,
+            with_children_count=with_children_count,
+            skip_children_names=skip_children_names,
+            with_parents=with_parents,
+            with_latest_values=with_latest_values,
+            **kwargs)
+        )
         managed_object.c8y = self.c8y  # inject c8y connection into instance
         return managed_object
 
@@ -155,7 +182,8 @@ def get_by(
             **kwargs))
         if len(result) == 1:
             return result[0]
-        raise ValueError("No matching object found." if not result else "Ambiguous query; multiple matching objects found.")
+        raise ValueError("No matching object found." if not result
+                         else "Ambiguous query; multiple matching objects found.")
 
     def get_count(
             self,
@@ -514,19 +542,46 @@ def accept(self, id: str):  # noqa (id)
         """
         self.c8y.put('/devicecontrol/newDeviceRequests/' + str(id), {'status': 'ACCEPTED'})
 
-    def get(self, id: str) -> Device:  # noqa (id)
+    def get(
+            self,
+            id: str,  # noqa
+            with_children: bool = None,
+            with_children_count: bool = None,
+            skip_children_names: bool = None,
+            with_parents: bool = None,
+            with_latest_values: bool = None,
+            **kwargs) -> Device:
         """ Retrieve a specific device object.
 
         Args:
-            id (str): ID of the device object
+            id (str): Cumulocity ID of the device object
+            with_children (bool):  Whether children with ID and name should be
+                included with each returned object
+            with_children_count (bool): When set to true, the returned result
+                will contain the total number of children in the respective
+                child additions, assets and devices sub fragments.
+            skip_children_names (bool):  If true, returned references of child
+                devices won't contain their names.
+            with_parents (bool): Whether to include a device's parents.
+            with_latest_values (bool):  If true the platform includes the
+                fragment `c8y_LatestMeasurements, which contains the latest
+                measurement values reported by the device to the platform.
 
         Returns:
             A Device instance
 
         Raises:
             KeyError:  if the ID is not defined within the database
         """
-        device = Device.from_json(self._get_object(id))
+        device = Device.from_json(self._get_object(
+            id,
+            with_children=with_children,
+            with_children_count=with_children_count,
+            skip_children_names=skip_children_names,
+            with_parents=with_parents,
+            with_latest_values=with_latest_values,
+            **kwargs)
+        )
         device.c8y = self.c8y
         return device
 
diff --git a/c8y_api/model/managedobjects.py b/c8y_api/model/managedobjects.py
@@ -286,7 +286,6 @@ def __init__(self, c8y: CumulocityRestApi = None,
         self.update_time = None
         self.child_devices = []
         self.child_assets = []
-        """List of NamedObject references to child assets."""
         self.child_additions = []
         self.parent_devices = []
         self.parent_assets = []
@@ -331,12 +330,16 @@ def _from_json(cls, json: dict, obj: Any) -> Any:
             mo.is_device = True
         if 'c8y_IsBinary' in json:
             mo.is_binary = True
-        if 'childDevices' in json:
-            mo.child_devices = cls._parse_references(json['childDevices'])
-        if 'childAssets' in json:
-            mo.child_assets = cls._parse_references(json['childAssets'])
-        if 'childAdditions' in json:
-            mo.child_additions = cls._parse_references(json['childAdditions'])
+        for fragment, field in [
+            ('childDevices', 'child_devices'),
+            ('childAssets', 'child_assets'),
+            ('childAdditions', 'child_additions'),
+            ('deviceParents', 'parent_devices'),
+            ('assetParents', 'parent_assets'),
+            ('additionParents', 'parent_additions'),
+        ]:
+            if fragment in json:
+                mo.__dict__[field] = cls._parse_references(json[fragment])
         return mo
 
     @classmethod
diff --git a/c8y_tk/app/subscription_listener.py b/c8y_tk/app/subscription_listener.py
@@ -119,6 +119,8 @@ def listen(self):
 
         This is blocking.
         """
+        # pylint: disable=too-many-branches
+
         # safely invoke a callback function blocking or non-blocking
         def invoke_callback(callback, is_blocking, _, arg):
             def safe_invoke(a):
@@ -128,7 +130,7 @@ def safe_invoke(a):
                         self._log.debug(f"Invoking callback: {callback.__module__}.{callback.__name__}")
                     callback(a)
                 except Exception as callback_error:
-                    self._log.error(f"Uncaught exception in callback: {callback_error}", exc_info=error)
+                    self._log.error(f"Uncaught exception in callback: {callback_error}", exc_info=callback_error)
             if is_blocking:
                 safe_invoke(arg)
             else:
@@ -180,8 +182,8 @@ def safe_invoke(a):
                 # schedule next run, skip if already exceeded
                 next_run = time.monotonic() + self.polling_interval
                 if self._log.isEnabledFor(logging.DEBUG):
-                    next_run_datetime = (datetime.now(timezone.utc) + timedelta(seconds=self.polling_interval) ).isoformat(sep=' ', timespec='seconds')
-                    self._log.debug(f"Next run at {next_run_datetime}.")
+                    next_run_datetime = datetime.now(timezone.utc) + timedelta(seconds=self.polling_interval)
+                    self._log.debug(f"Next run at {next_run_datetime.isoformat(sep=' ', timespec='seconds')}.")
                 # sleep until next poll
                 if not time.monotonic() > next_run:
                     self._is_closed.wait(next_run - time.monotonic())
@@ -192,6 +194,7 @@ def safe_invoke(a):
             if self._executor:
                 self._executor.shutdown(wait=False, cancel_futures=False)
 
+        # pylint: disable=broad-exception-caught
         except Exception as error:
             self._log.error(f"Uncaught exception during listen: {error}", exc_info=error)
 
diff --git a/integration_tests/test_alarms.py b/integration_tests/test_alarms.py
@@ -141,7 +141,6 @@ def sample_alarms(session_device, module_factory) -> List[Alarm]:
     ]
 
 
-
 def test_apply_by(live_c8y: CumulocityApi, session_device: Device):
     """Verify that the apply_by function works."""
 
diff --git a/integration_tests/test_inventory.py b/integration_tests/test_inventory.py
@@ -170,6 +170,99 @@ def test_reload(live_c8y):
     assert 'c8y_AdditionalFragment' not in obj2.fragments
 
 
+@pytest.fixture(name='asset_hierarchy_root_id', scope='module')
+def fix_asset_hierarchy_root_id(module_factory):
+    """Provide a (read-only) sample asset hierarchy for corresponding tests.
+
+    This fixture creates a root object with a child of each kind (asset,
+    device, addition). Each of the children references to another 'addition'
+    child to create a multi-level hierarchy.
+
+    It is automatically cleaned up after testing.
+    """
+    name = RandomNameGenerator.random_name()
+    obj = module_factory(ManagedObject(name=f'Root-{name}', type=f'Root-{name}'))
+
+    addition = module_factory(ManagedObject(name=f'Addition-{name}', type=f'Addition-{name}'))
+    asset = module_factory(ManagedObject(name=f'Asset-{name}', type=f'Asset-{name}'))
+    device = module_factory(Device(name=f'Device-{name}', type=f'Device-{name}'))
+    obj.add_child_addition(addition)
+    obj.add_child_asset(asset)
+    obj.add_child_device(device)
+
+    sub_addition =  module_factory(ManagedObject(name=f'SubAddition-{name}', type=f'Addition-{name}'))
+    addition.add_child_addition(sub_addition)
+    asset.add_child_addition(sub_addition)
+    device.add_child_addition(sub_addition)
+
+    return obj.id
+
+
+def test_references(live_c8y: CumulocityApi, asset_hierarchy_root_id):
+    """Verify that parent references are handles as expected.
+
+    This test uses the "asset_hierarchy" fixture which defines a root
+    with children of each kind.
+    """
+    root_id = asset_hierarchy_root_id
+
+    # (1) ignore children and parents
+    result = live_c8y.inventory.get(root_id, with_children=False)
+    assert not result.child_assets
+    assert not result.child_devices
+    assert not result.child_additions
+    assert not result.parent_assets
+    assert not result.parent_devices
+    assert not result.parent_additions
+
+    # (2) include children, with names
+    result = live_c8y.inventory.get(root_id, with_children=True, skip_children_names=False)
+    # -> the root object references one of each
+    assert len(result.child_assets) == 1
+    assert len(result.child_devices) == 1
+    assert len(result.child_additions) == 1
+    # -> including their names
+    assert result.child_assets[0].name
+    assert result.child_devices[0].name
+    assert result.child_additions[0].name
+    # -> but no parents
+    assert not result.parent_assets
+    assert not result.parent_devices
+    assert not result.parent_additions
+
+    # (3) include children, no names
+    result = live_c8y.inventory.get(root_id, with_children=True, skip_children_names=True)
+    # -> the root object references one of each
+    assert len(result.child_assets) == 1
+    assert len(result.child_devices) == 1
+    assert len(result.child_additions) == 1
+    # -> including their names
+    assert not result.child_assets[0].name
+    assert not result.child_devices[0].name
+    assert not result.child_additions[0].name
+
+
+@pytest.mark.parametrize('child_type', ['asset', 'device', 'addition'])
+def test_parent_references(live_c8y: CumulocityApi, asset_hierarchy_root_id, child_type):
+    """Verify that parent references are handles as expected.
+
+    This test uses the "asset_hierarchy" fixture which defines a root
+    with children of each kind. Each kind has another "addition" child.
+    """
+    root = live_c8y.inventory.get(asset_hierarchy_root_id, with_children=True)
+    child = root.__dict__[f'child_{child_type}s'][0]
+
+    # read child with references
+    result = live_c8y.inventory.get(child.id,  with_children=True, with_parents=True)
+    # parent (root) is linked by the child's type
+    parents = result.__dict__[f'parent_{child_type}s']
+    assert len(parents) == 1
+    assert parents[0].id == root.id
+    assert parents[0].name == root.name
+    # each child as an 'addition' child
+    assert len(result.child_additions) == 1
+
+
 def test_deletion(live_c8y: CumulocityApi, safe_create):
     """Verify that deletion works as expected.
 
diff --git a/tests/model/test_administration.py b/tests/model/test_administration.py
@@ -16,7 +16,7 @@
      ['username=U', 'groups=1,2,3', 'owner=O'],
      []),
     ({'only_devices': False, 'with_subusers_count': True},
-     ['onlyDevices=False', 'withSubusersCount=True'],
+     ['onlyDevices=false', 'withSubusersCount=true'],
      ['_']),
     ({'snake_case': 'SC', 'pascalCase': 'PC'},
      ['snakeCase=SC', 'pascalCase=PC'],
diff --git a/tests/model/test_alarms.py b/tests/model/test_alarms.py
@@ -37,10 +37,10 @@ def isolate_call_url(fun, **kwargs):
 @pytest.mark.parametrize('params, expected, not_expected', [
     ({'expression': 'EX', 'type': 'T'}, ['?EX'], ['type']),
     ({'type': 'T', 'source': 'S', 'fragment': 'F'}, ['type=T', 'source=S', 'fragmentType=F'], []),
-    ({'status': 'ST', 'severity': 'SE', 'resolved': False}, ['status=ST', 'severity=SE', 'resolved=False'], []),
+    ({'status': 'ST', 'severity': 'SE', 'resolved': False}, ['status=ST', 'severity=SE', 'resolved=false'], []),
     ({'reverse': False, 'page_size': 8}, ['revert=false', 'pageSize=8'], ['reverse']),
     ({'with_source_assets': False, 'with_source_devices': True, 'source': '123'},
-     ['withSourceAssets=False', 'withSourceDevices=True'],
+     ['withSourceAssets=false', 'withSourceDevices=true'],
      ['_']),
     # data priorities
     ({'date_from': '2020-12-31', 'date_to': '2021-12-31'},
diff --git a/tests/model/test_applications.py b/tests/model/test_applications.py
@@ -33,7 +33,7 @@ def isolate_call_url(fun, **kwargs):
     ({'tenant': 'T', 'subscriber': 'S', 'provided_for': 'P'},
      ['tenant=T', 'subscriber=S', 'providedFor=P'],
      ['_']),
-    ({'has_versions': False}, ['hasVersions=False'], ['_']),
+    ({'has_versions': False}, ['hasVersions=false'], ['_']),
     ({'snake_case': 'SC', 'pascalCase': 'PC'},
      ['snakeCase=SC', 'pascalCase=PC'],
      ['_']),
diff --git a/tests/model/test_events.py b/tests/model/test_events.py
@@ -32,10 +32,10 @@ def isolate_call_url(fun, **kwargs):
     ({'expression': 'EX', 'type': 'T'}, ['?EX'], ['type']),
     ({'type': 'T', 'source': 'S', 'fragment': 'F'}, ['type=T', 'source=S', 'fragmentType=F'], []),
     ({'fragment_type': 'T', 'fragment_value': 'V'}, ['fragmentType=T', 'fragmentValue=V'], ['_']),
-    ({'status': 'ST', 'severity': 'SE', 'resolved': False}, ['status=ST', 'severity=SE', 'resolved=False'], []),
+    ({'status': 'ST', 'severity': 'SE', 'resolved': False}, ['status=ST', 'severity=SE', 'resolved=false'], []),
     ({'reverse': False}, ['revert=false'], ['reverse']),
     ({'with_source_assets': False, 'with_source_devices': True, 'source': '123'},
-     ['withSourceAssets=False', 'withSourceDevices=True'],
+     ['withSourceAssets=false', 'withSourceDevices=true'],
      ['_']),
     # data priorities
     ({'date_from': '2020-12-31', 'date_to': '2021-12-31'},
diff --git a/tests/model/test_inventory.py b/tests/model/test_inventory.py
diff --git a/tests/model/test_resource.py b/tests/model/test_resource.py
diff --git a/tests/utils.py b/tests/utils.py

Original file line number	Diff line number	Diff line change
`@@ -141,7 +141,6 @@ def sample_alarms(session_device, module_factory) -> List[Alarm]:`
`141`	`141`	`]`
`142`	`142`
`143`	`143`
`144`		`-`
`145`	`144`	`def test_apply_by(live_c8y: CumulocityApi, session_device: Device):`
`146`	`145`	`"""Verify that the apply_by function works."""`
`147`	`146`