Handle missing D-Bus property in pool listing

Signed-off-by: mulhern <amulhern@redhat.com>

Author: mulkieran

src/stratis_cli/_actions/_list_pool.py

@@ -30,6 +30,9 @@
 # isort: THIRDPARTY
 from justbytes import Range
 
+# isort: FIRSTPARTY
+from dbus_client_gen import DbusClientMissingPropertyError
+
 from .._alerts import (
     PoolAllocSpaceAlert,
     PoolDeviceSizeChangeAlert,
@@ -251,6 +254,8 @@ def metadata_version(mopool: Any) -> MetadataVersion | None:
             return MetadataVersion(int(mopool.MetadataVersion()))
         except ValueError:  # pragma: no cover
             return None
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            return None
 
     @staticmethod
     def _volume_key_loaded(mopool: Any) -> tuple[bool, bool] | tuple[bool, str]:
@@ -268,62 +273,86 @@ def alert_codes(
         mopool: Any,
     ) -> list[PoolEncryptionAlert | PoolAllocSpaceAlert | PoolMaintenanceAlert]:
         """
-        Return alert code objects for a pool.
+        Return alert code objects for a pool. If some D-Bus properties are
+        missing, the information related to those D-Bus properties is simply
+        omitted.
 
         :param mopool: object to access pool properties
 
         :returns: list of alerts obtainable from GetManagedObjects properties
         """
-        action_availability = PoolActionAvailability[str(mopool.AvailableActions())]
-        availability_alerts = action_availability.pool_maintenance_alerts()
+        try:
+            action_availability = PoolActionAvailability[str(mopool.AvailableActions())]
+            availability_alerts = action_availability.pool_maintenance_alerts()
 
-        no_alloc_space_alerts = (
-            [PoolAllocSpaceAlert.NO_ALLOC_SPACE] if mopool.NoAllocSpace() else []
-        )
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            availability_alerts = []
 
-        metadata_version = Default.metadata_version(mopool)
+        try:
+            no_alloc_space_alerts = (
+                [PoolAllocSpaceAlert.NO_ALLOC_SPACE] if mopool.NoAllocSpace() else []
+            )
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            no_alloc_space_alerts = []
 
-        (vkl_is_bool, volume_key_loaded) = Default._volume_key_loaded(mopool)
-
-        pool_encryption_alerts: list[PoolEncryptionAlert] = (
-            [PoolEncryptionAlert.VOLUME_KEY_NOT_LOADED]
-            if metadata_version is MetadataVersion.V2
-            and mopool.Encrypted()
-            and vkl_is_bool
-            and not volume_key_loaded
-            else []
-        ) + (
-            [PoolEncryptionAlert.VOLUME_KEY_STATUS_UNKNOWN]
-            if metadata_version is MetadataVersion.V2
-            and mopool.Encrypted()
-            and not vkl_is_bool
-            else []
-        )
+        metadata_version = Default.metadata_version(mopool)
 
+        try:
+            (vkl_is_bool, volume_key_loaded) = Default._volume_key_loaded(mopool)
+            encrypted = bool(mopool.Encrypted())
+            pool_encryption_alerts: list[PoolEncryptionAlert] = (
+                [PoolEncryptionAlert.VOLUME_KEY_NOT_LOADED]
+                if metadata_version is MetadataVersion.V2
+                and encrypted
+                and vkl_is_bool
+                and not volume_key_loaded
+                else []
+            ) + (
+                [PoolEncryptionAlert.VOLUME_KEY_STATUS_UNKNOWN]
+                if metadata_version is MetadataVersion.V2
+                and encrypted
+                and not vkl_is_bool
+                else []
+            )
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            pool_encryption_alerts = [PoolEncryptionAlert.VOLUME_KEY_STATUS_UNKNOWN]
         return availability_alerts + no_alloc_space_alerts + pool_encryption_alerts
 
     @staticmethod
     def size_triple(mopool: Any) -> SizeTriple:
         """
         Calculate SizeTriple from size information.
         """
-        return SizeTriple(
-            Range(mopool.TotalPhysicalSize()),
-            get_property(mopool.TotalPhysicalUsed(), Range, None),
-        )
+        try:
+            size = Range(mopool.TotalPhysicalSize())
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            size = None
+
+        try:
+            used = get_property(mopool.TotalPhysicalUsed(), Range, None)
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            used = None
+
+        return SizeTriple(size, used)
 
     def uuid_str(self, mopool: Any) -> str:
         """
         Return a string representation of UUID, correctly formatted.
         """
-        return self.uuid_formatter(mopool.Uuid())
+        try:
+            return self.uuid_formatter(mopool.Uuid())
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            return TABLE_UNKNOWN_STRING
 
     @staticmethod
     def name_str(mopool: Any) -> str:
         """
         Return a string representation of the pool name.
         """
-        return mopool.Name()
+        try:
+            return mopool.Name()
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            return TABLE_UNKNOWN_STRING
 
 
 class DefaultDetail(Default):  # pylint: disable=too-few-public-methods
@@ -343,7 +372,7 @@ def __init__(self, uuid_formatter: Callable[[str | UUID], str], selection: PoolI
 
     def _print_detail_view(
         self, pool_object_path: str, mopool: Any, alerts: DeviceSizeChangedAlerts
-    ):  # pylint: disable=too-many-locals
+    ):  # pylint: disable=too-many-locals,too-many-statements,too-many-branches
         """
         Print the detailed view for a single pool.
 
@@ -352,8 +381,6 @@ def _print_detail_view(
         :param MOPool mopool: properties of the pool
         :param DeviceSizeChangedAlerts alerts: pool alerts
         """
-        encrypted = mopool.Encrypted()
-
         print(f"UUID: {self.uuid_str(mopool)}")
         print(f"Name: {Default.name_str(mopool)}")
 
@@ -367,71 +394,128 @@ def _print_detail_view(
             print(f"     {line}")
 
         metadata_version = Default.metadata_version(mopool)
+        metadata_version_str = (
+            metadata_version if metadata_version is not None else TABLE_UNKNOWN_STRING
+        )
+        print(f"Metadata Version: {metadata_version_str}")
+
+        try:
+            actions_allowed_str = PoolActionAvailability[
+                str(mopool.AvailableActions())
+            ].name
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            actions_allowed_str = TABLE_UNKNOWN_STRING
+        print(f"Actions Allowed: {actions_allowed_str}")
 
-        print(f"Metadata Version: {metadata_version}")
+        try:
+            cache_str = "Yes" if mopool.HasCache() else "No"
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            cache_str = TABLE_UNKNOWN_STRING
+        print(f"Cache: {cache_str}")
 
-        print(
-            f"Actions Allowed: "
-            f"{PoolActionAvailability[str(mopool.AvailableActions())].name}"
-        )
-        print(f"Cache: {'Yes' if mopool.HasCache() else 'No'}")
-        print(f"Filesystem Limit: {mopool.FsLimit()}")
-        print(
-            f"Allows Overprovisioning: "
-            f"{'Yes' if mopool.Overprovisioning() else 'No'}"
-        )
+        try:
+            fs_limit_str = mopool.FsLimit()
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            fs_limit_str = TABLE_UNKNOWN_STRING
+        print(f"Filesystem Limit: {fs_limit_str}")
+
+        try:
+            allow_overprovisioning_str = "Yes" if mopool.Overprovisioning() else "No"
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            allow_overprovisioning_str = TABLE_UNKNOWN_STRING
+        print(f"Allows Overprovisioning: {allow_overprovisioning_str}")
 
-        if encrypted:
+        try:
+            encrypted = bool(mopool.Encrypted())
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            encrypted = None
+
+        if encrypted is None:  # pragma: no cover
+            print(f"Encryption Enabled: {TABLE_UNKNOWN_STRING}")
+        elif encrypted is True:
             print("Encryption Enabled: Yes")
 
             if metadata_version is MetadataVersion.V1:  # pragma: no cover
-                key_description_str = _non_existent_or_inconsistent_to_str(
-                    EncryptionInfoKeyDescription(mopool.KeyDescriptions())
-                )
+                try:
+                    key_description_str = _non_existent_or_inconsistent_to_str(
+                        EncryptionInfoKeyDescription(mopool.KeyDescriptions())
+                    )
+                except DbusClientMissingPropertyError:
+                    key_description_str = TABLE_UNKNOWN_STRING
                 print(f"    Key Description: {key_description_str}")
 
-                clevis_info_str = _non_existent_or_inconsistent_to_str(
-                    EncryptionInfoClevis(mopool.ClevisInfos()),
-                    interp=_clevis_to_str,
-                )
+                try:
+                    clevis_info_str = _non_existent_or_inconsistent_to_str(
+                        EncryptionInfoClevis(mopool.ClevisInfos()),
+                        interp=_clevis_to_str,
+                    )
+                except DbusClientMissingPropertyError:
+                    clevis_info_str = TABLE_UNKNOWN_STRING
                 print(f"    Clevis Configuration: {clevis_info_str}")
             elif metadata_version is MetadataVersion.V2:
-                encryption_infos = sorted(
-                    [
+
+                try:
+                    free_str = get_property(
+                        mopool.FreeTokenSlots(),
+                        lambda x: str(int(x)),
+                        TABLE_UNKNOWN_STRING,
+                    )
+                except DbusClientMissingPropertyError:  # pragma: no cover
+                    free_str = TABLE_UNKNOWN_STRING
+                print(f"    Free Token Slots Remaining: {free_str}")
+
+                try:
+                    key_encryption_infos = [
                         TokenSlotInfo(token_slot, key=str(description))
                         for token_slot, description in mopool.KeyDescriptions()
                     ]
-                    + [
+                except DbusClientMissingPropertyError:  # pragma: no cover
+                    key_encryption_infos = []
+
+                try:
+                    clevis_encryption_infos = [
                         TokenSlotInfo(
                             token_slot, clevis=(str(pin), json.loads(str(config)))
                         )
                         for token_slot, (pin, config) in mopool.ClevisInfos()
-                    ],
-                    key=lambda x: x.token_slot,
-                )
+                    ]
+                except DbusClientMissingPropertyError:  # pragma: no cover
+                    clevis_encryption_infos = []
 
-                free_valid, free = mopool.FreeTokenSlots()
-                print(
-                    f'    Free Token Slots Remaining: {int(free) if free_valid else "<UNKNOWN>"}'
+                encryption_infos = sorted(
+                    key_encryption_infos + clevis_encryption_infos,
+                    key=lambda x: x.token_slot,
                 )
-
-                for info in encryption_infos:
-                    for line in str(info).split(os.linesep):
-                        print(f"    {line}")
+                if encryption_infos == []:
+                    print("    No token slot information available")
+                else:  # pragma: no cover
+                    for info in encryption_infos:
+                        for line in str(info).split(os.linesep):
+                            print(f"    {line}")
             else:  # pragma: no cover
-                pass
+                print("    No encryption information available")
         else:
             print("Encryption Enabled: No")
 
         size_triple = Default.size_triple(mopool)
 
-        print(f"Fully Allocated: {'Yes' if mopool.NoAllocSpace() else 'No'}")
-        print(f"    Size: {size_triple.total()}")
-        print(f"    Allocated: {Range(mopool.AllocatedSize())}")
-        print(
-            "    Used: "
-            f"{TABLE_UNKNOWN_STRING if size_triple.used() is None else size_triple.used()}"
-        )
+        def size_str(value: Range | None) -> str:
+            return TABLE_UNKNOWN_STRING if value is None else str(value)
+
+        try:
+            fully_allocated_str = "Yes" if mopool.NoAllocSpace() else "No"
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            fully_allocated_str = TABLE_UNKNOWN_STRING
+        print(f"Fully Allocated: {fully_allocated_str}")
+
+        print(f"    Size: {size_str(size_triple.total())}")
+
+        try:
+            allocated_size = Range(mopool.AllocatedSize())
+        except DbusClientMissingPropertyError:  # pragma: no cover
+            allocated_size = None
+        print(f"    Allocated: {size_str(allocated_size)}")
+        print(f"    Used: {size_str(size_triple.used())}")
 
     def display(self):
         """