Add lsblk to "stratis pool report"

.github/workflows/main.yml

@@ -125,6 +125,7 @@ jobs:
           python3-psutil
           python3-wcwidth
           systemd-devel
+          util-linux
       - uses: dtolnay/rust-toolchain@master
         with:
           components: cargo

src/stratis_cli/_actions/_pool.py

@@ -46,6 +46,7 @@
 from ._constants import TOP_OBJECT
 from ._formatting import get_property, get_uuid_formatter
 from ._list_pool import list_pools
+from ._report_pool import report_pool
 from ._utils import ClevisInfo, PoolSelector
 
 
@@ -710,3 +711,18 @@ def explain_code(namespace):
         Print an explanation of pool error code.
         """
         print(PoolErrorCode.explain(namespace.code))
+
+    @staticmethod
+    def report_pool(namespace):
+        """
+        Report information about a pool.
+        """
+        uuid_formatter = get_uuid_formatter(namespace.unhyphenated_uuids)
+
+        selection = (
+            PoolSelector(PoolIdType.NAME, namespace.name)
+            if namespace.uuid is None
+            else PoolSelector(PoolIdType.UUID, namespace.uuid)
+        )
+
+        return report_pool(uuid_formatter, selection, stopped=namespace.stopped)

src/stratis_cli/_actions/_report_pool.py

@@ -0,0 +1,150 @@
+# Copyright 2023 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+Pool report.
+"""
+
+# isort: STDLIB
+import subprocess
+from abc import ABC, abstractmethod
+
+from .._errors import StratisCliResourceNotFoundError
+from ._connection import get_object
+from ._constants import TOP_OBJECT
+
+
+def _fetch_stopped_pools_property(proxy):
+    """
+    Fetch the StoppedPools property from stratisd.
+    :param proxy: proxy to the top object in stratisd
+    :return: a representation of stopped devices
+    :rtype: dict
+    :raises StratisCliEngineError:
+    """
+
+    # pylint: disable=import-outside-toplevel
+    from ._data import Manager
+
+    return Manager.Properties.StoppedPools.Get(proxy)
+
+
+def report_pool(uuid_formatter, selection, *, stopped=False):
+    """
+    List the specified information about pools.
+    """
+    klass = (
+        Stopped(uuid_formatter, selection=selection)
+        if stopped
+        else Default(uuid_formatter, selection=selection)
+    )
+    klass.report_pool()
+
+
+class Report(ABC):  # pylint: disable=too-few-public-methods
+    """
+    Handle reporting on a pool or pools.
+    """
+
+    def __init__(self, uuid_formatter, selection):
+        """
+        Initialize a List object.
+        :param uuid_formatter: function to format a UUID str or UUID
+        :param uuid_formatter: str or UUID -> str
+        :param selection: how to select the pool to display
+        :type selection: pair of str * object
+        """
+        self.uuid_formatter = uuid_formatter
+        self.selection = selection
+
+    @abstractmethod
+    def report_pool(self):
+        """
+        Report the pools.
+        """
+
+
+class Default(Report):  # pylint: disable=too-few-public-methods
+    """
+    Report on pools that stratis list by default.
+    """
+
+    def report_pool(self):
+        """
+        Do pool reports.
+        """
+        # pylint: disable=import-outside-toplevel
+        from ._data import MODev, ObjectManager, devs, pools
+
+        proxy = get_object(TOP_OBJECT)
+
+        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
+
+        (pool_object_path, _) = next(
+            pools(props=self.selection.managed_objects_key())
+            .require_unique_match(True)
+            .search(managed_objects)
+        )
+
+        # For sim engine tests, actually running report commands will go wrong.
+        # Therefore, all tests exit at the previous line.
+        modevs = (
+            MODev(info)
+            for objpath, info in devs(props={"Pool": pool_object_path}).search(
+                managed_objects
+            )
+        )  # pragma: no cover
+
+        blkdevs = [dev.Devnode() for dev in modevs]  # pragma: no cover
+        # Ignore bandit B603 errors.  Input comes from D-Bus and has
+        # been processed.
+        subprocess.run(
+            ["/usr/bin/lsblk", "-i"] + blkdevs, check=True
+        )  # nosec B603 # pragma: no cover
+
+
+class Stopped(Report):  # pylint: disable=too-few-public-methods
+    """
+    Support for reporting on stopped pools.
+    """
+
+    def report_pool(self):
+        """
+        Report on stopped pool.
+        """
+        proxy = get_object(TOP_OBJECT)
+
+        stopped_pools = _fetch_stopped_pools_property(proxy)
+
+        selection_func = self.selection.stopped_pools_func()
+
+        stopped_pool = next(
+            (
+                info
+                for (uuid, info) in stopped_pools.items()
+                if selection_func(uuid, info)
+            ),
+            None,
+        )
+
+        if stopped_pool is None:
+            raise StratisCliResourceNotFoundError("report", str(self.selection))
+
+        # For sim engine tests, actually running report commands will go wrong.
+        # Therefore, all tests exit at the previous line.
+        # Ignore bandit B603 errors.  Input comes from D-Bus and has
+        # been processed.
+        blkdevs = [dev.devnode for dev in stopped_pool.devs]  # pragma: no cover
+        subprocess.run(
+            ["/usr/bin/lsblk", "-i"] + blkdevs, check=True
+        )  # nosec B603 # pragma: no cover

src/stratis_cli/_parser/_pool.py

@@ -441,4 +441,43 @@ def _ensure_nat(arg):
             "subcmds": POOL_DEBUG_SUBCMDS,
         },
     ),
+    (
+        "report",
+        {
+            "help": "Report information about pools",
+            "description": "Generate report for pools",
+            "mut_ex_args": [
+                (
+                    True,
+                    [
+                        (
+                            "--uuid",
+                            {
+                                "action": "store",
+                                "type": UUID,
+                                "help": "UUID of pool",
+                            },
+                        ),
+                        (
+                            "--name",
+                            {
+                                "action": "store",
+                                "help": "name of pool",
+                            },
+                        ),
+                    ],
+                ),
+            ],
+            "args": [
+                (
+                    "--stopped",
+                    {
+                        "action": "store_true",
+                        "help": "Display information about stopped pools only.",
+                    },
+                ),
+            ],
+            "func": PoolActions.report_pool,
+        },
+    ),
 ]

tests/whitebox/integration/pool/test_report.py

@@ -0,0 +1,79 @@
+# Copyright 2022 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+Test 'report'.
+"""
+# isort: STDLIB
+from uuid import uuid4
+
+# isort: FIRSTPARTY
+from dbus_client_gen import DbusClientUniqueResultError
+
+# isort: LOCAL
+from stratis_cli import StratisCliErrorCodes
+from stratis_cli._errors import StratisCliResourceNotFoundError
+
+from .._misc import RUNNER, SimTestCase, device_name_list
+
+_ERROR = StratisCliErrorCodes.ERROR
+_DEVICE_STRATEGY = device_name_list(1, 1)
+
+
+class ReportTestCase(SimTestCase):
+    """
+    Test 'report' on a sim pool.
+    """
+
+    _MENU = ["--propagate", "pool", "report"]
+    _POOLNAME = "poolname"
+
+    def setUp(self):
+        super().setUp()
+        command_line = ["pool", "create", self._POOLNAME] + _DEVICE_STRATEGY()
+        RUNNER(command_line)
+
+    def test_call_bad_uuid(self):
+        """
+        Test bad uuid
+        """
+        command_line = self._MENU + ["--uuid", str(uuid4())]
+        self.check_error(DbusClientUniqueResultError, command_line, _ERROR)
+
+    def test_call_bad_uuid_stopped(self):
+        """
+        Test bad uuid for stopped pools.
+        """
+        command_line = self._MENU + ["--stopped", "--uuid", str(uuid4())]
+        self.check_error(StratisCliResourceNotFoundError, command_line, _ERROR)
+
+    def test_report_bad_name(self):
+        """
+        Test bad name.
+        """
+        command_line = self._MENU + [
+            "--name",
+            "noone",
+        ]
+        self.check_error(DbusClientUniqueResultError, command_line, _ERROR)
+
+    def test_report_bad_name_stopped(self):
+        """
+        Test bad name for stopped pools.
+        """
+        command_line = self._MENU + [
+            "--stopped",
+            "--name",
+            "noone",
+        ]
+        self.check_error(StratisCliResourceNotFoundError, command_line, _ERROR)