class LogicalActions:
    """
    Actions on the logical aspects of a pool.
    """

    @staticmethod
    def create_volumes(namespace):
        """
        Create volumes in a pool.

        :raises StratisCliEngineError:
        :raises StratisCliIncoherenceError:
        :raises StratisCliPartialChangeError:
        """
        # pylint: disable=too-many-locals

        # pylint: disable=import-outside-toplevel
        from ._data import MOFilesystem, ObjectManager, Pool, filesystems, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
        (pool_object_path, _) = next(
            pools(props={"Name": namespace.pool_name})
            .require_unique_match(True)
            .search(managed_objects)
        )

        requested_names = frozenset(namespace.fs_name)

        names = frozenset(
            MOFilesystem(info).Name()
            for (_, info) in filesystems(props={"Pool": pool_object_path}).search(
                managed_objects
            )
        )
        already_names = requested_names.intersection(names)

        if already_names != frozenset():
            raise StratisCliPartialChangeError(
                "create", requested_names.difference(already_names), already_names
            )

        requested_size_arg = (
            (False, "")
            if namespace.size is None
            else (True, str(namespace.size.magnitude))
        )
        requested_specs = [(n, requested_size_arg) for n in requested_names]

        (
            (created, list_created),
            return_code,
            message,
        ) = Pool.Methods.CreateFilesystems(
            get_object(pool_object_path), {"specs": requested_specs}
        )

        if return_code != StratisdErrors.OK:
            raise StratisCliEngineError(return_code, message)

        if not created or len(list_created) < len(requested_names):  # pragma: no cover
            raise StratisCliIncoherenceError(
                (
                    "Expected to create the specified filesystems in pool %s "
                    "but stratisd reports that it did not actually create "
                    "some or all of the filesystems requested"
                )
                % namespace.pool_name
            )

    @staticmethod
    def list_volumes(namespace):
        """
        List the volumes in a pool.
        """
        # pylint: disable=import-outside-toplevel
        from ._data import MOFilesystem, MOPool, ObjectManager, filesystems, pools

        # This method is invoked as the default for "stratis filesystem";
        # the namespace may not have a pool_name field.
        pool_name = getattr(namespace, "pool_name", None)

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})

        filesystems_with_props = [
            MOFilesystem(info)
            for objpath, info in filesystems(
                props=None
                if pool_name is None
                else {
                    "Pool": next(
                        pools(props={"Name": pool_name})
                        .require_unique_match(True)
                        .search(managed_objects)
                    )[0]
                }
            ).search(managed_objects)
        ]

        path_to_name = dict(
            (path, MOPool(info).Name())
            for path, info in pools(
                props=None if pool_name is None else {"Name": pool_name}
            ).search(managed_objects)
        )

        def filesystem_size_triple(dbus_props):
            """
            Calcuate the triple to display for filesystem size.

            :param dbus_props: filesystem D-Bus properties
            :type dbus_props: MOFilesystem

            :returns: a string a formatted string showing all three values
            :rtype: str
            """
            total = Range(dbus_props.Size())
            used = get_property(dbus_props.Used(), Range, None)
            return size_triple(total, used)

        format_uuid = (
            (lambda mo_uuid: mo_uuid) if namespace.unhyphenated_uuids else to_hyphenated
        )

        tables = [
            (
                path_to_name[mofilesystem.Pool()],
                mofilesystem.Name(),
                filesystem_size_triple(mofilesystem),
                date_parser.parse(mofilesystem.Created())
                .astimezone()
                .strftime("%b %d %Y %H:%M"),
                mofilesystem.Devnode(),
                format_uuid(mofilesystem.Uuid()),
            )
            for mofilesystem in filesystems_with_props
        ]

        print_table(
            ["Pool Name", "Name", "Size", "Created", "Device", "UUID"],
            sorted(tables, key=lambda entry: (entry[0], entry[1])),
            ["<", "<", "<", "<", "<", "<"],
        )

    @staticmethod
    def destroy_volumes(namespace):
        """
        Destroy volumes in a pool.

        :raises StratisCliEngineError:
        :raises StratisCliIncoherenceError:
        :raises StratisCliPartialChangeError:
        """
        # pylint: disable=too-many-locals

        # pylint: disable=import-outside-toplevel
        from ._data import MOFilesystem, ObjectManager, Pool, filesystems, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})

        (pool_object_path, _) = next(
            pools(props={"Name": namespace.pool_name})
            .require_unique_match(True)
            .search(managed_objects)
        )

        requested_names = frozenset(namespace.fs_name)

        pool_filesystems = {
            MOFilesystem(info).Name(): op
            for (op, info) in filesystems(props={"Pool": pool_object_path}).search(
                managed_objects
            )
        }
        already_removed = requested_names.difference(frozenset(pool_filesystems.keys()))

        if already_removed != frozenset():
            raise StratisCliPartialChangeError(
                "destroy", requested_names.difference(already_removed), already_removed
            )

        fs_object_paths = [
            op for (name, op) in pool_filesystems.items() if name in requested_names
        ]

        (
            (destroyed, list_destroyed),
            return_code,
            message,
        ) = Pool.Methods.DestroyFilesystems(
            get_object(pool_object_path), {"filesystems": fs_object_paths}
        )

        if return_code != StratisdErrors.OK:  # pragma: no cover
            raise StratisCliEngineError(return_code, message)

        if not destroyed or len(list_destroyed) < len(
            fs_object_paths
        ):  # pragma: no cover
            raise StratisCliIncoherenceError(
                (
                    "Expected to destroy the specified filesystems in pool %s "
                    "but stratisd reports that it did not actually destroy "
                    "some or all of the filesystems requested"
                )
                % namespace.pool_name
            )

    @staticmethod
    def snapshot_filesystem(namespace):
        """
        Snapshot filesystem in a pool.

        :raises StratisCliEngineError:
        :raises StratisCliNoChangeError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import ObjectManager, Pool, filesystems, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})

        (pool_object_path, _) = next(
            pools(props={"Name": namespace.pool_name})
            .require_unique_match(True)
            .search(managed_objects)
        )
        (origin_fs_object_path, _) = next(
            filesystems(props={"Name": namespace.origin_name, "Pool": pool_object_path})
            .require_unique_match(True)
            .search(managed_objects)
        )

        ((changed, _), return_code, message) = Pool.Methods.SnapshotFilesystem(
            get_object(pool_object_path),
            {"origin": origin_fs_object_path, "snapshot_name": namespace.snapshot_name},
        )

        if return_code != StratisdErrors.OK:  # pragma: no cover
            raise StratisCliEngineError(return_code, message)

        if not changed:
            raise StratisCliNoChangeError("snapshot", namespace.snapshot_name)

    @staticmethod
    def rename_fs(namespace):
        """
        Rename a filesystem.

        :raises StratisCliEngineError:
        :raises StratisCliNoChangeError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import Filesystem, ObjectManager, filesystems, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
        (pool_object_path, _) = next(
            pools(props={"Name": namespace.pool_name})
            .require_unique_match(True)
            .search(managed_objects)
        )
        (fs_object_path, _) = next(
            filesystems(props={"Name": namespace.fs_name, "Pool": pool_object_path})
            .require_unique_match(True)
            .search(managed_objects)
        )

        ((changed, _), return_code, message) = Filesystem.Methods.SetName(
            get_object(fs_object_path), {"name": namespace.new_name}
        )

        if return_code != StratisdErrors.OK:  # pragma: no cover
            raise StratisCliEngineError(return_code, message)

        if not changed:
            raise StratisCliNoChangeError("rename", namespace.new_name) class PhysicalActions:
    """
    Actions on the physical aspects of a pool.
    """

    # pylint: disable=too-few-public-methods

    @staticmethod
    def list_devices(namespace):
        """
        List devices. If a pool is specified in the namespace, list devices
        for that pool. Otherwise, list all devices for all pools.
        """
        # pylint: disable=import-outside-toplevel
        from ._data import MODev, MOPool, ObjectManager, devs, pools

        # This method is invoked as the default for "stratis blockdev";
        # the namespace may not have a pool_name field.
        pool_name = getattr(namespace, "pool_name", None)

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})

        modevs = [
            MODev(info)
            for objpath, info in devs(
                props=None
                if pool_name is None
                else {
                    "Pool": next(
                        pools(props={"Name": pool_name})
                        .require_unique_match(True)
                        .search(managed_objects)
                    )[0]
                }
            ).search(managed_objects)
        ]

        path_to_name = dict(
            (path, MOPool(info).Name())
            for path, info in pools(
                props=None if pool_name is None else {"Name": pool_name}
            ).search(managed_objects)
        )

        def paths(modev):
            """
            Return <physical_path> (<metadata_path>) if they are different,
            otherwise, just <metadata_path>.

            physical_path D-Bus Property key is PhysicalPath
            metadata_path D-Bus Property key is Devnode

            :param modev: object containing D-Bus properties
            :returns: the string to print
            :rtype: str
            """
            metadata_path = modev.Devnode()
            physical_path = modev.PhysicalPath()

            return (
                metadata_path
                if metadata_path == physical_path
                else "%s (%s)" % (physical_path, metadata_path)
            )

        tables = [
            [
                path_to_name[modev.Pool()],
                paths(modev),
                str(Range(modev.TotalPhysicalSize())),
                BLOCK_DEV_TIER_TO_NAME(modev.Tier(), True),
            ]
            for modev in modevs
        ]
        print_table(
            ["Pool Name", "Device Node", "Physical Size", "Tier"],
            sorted(tables, key=lambda entry: (entry[0], entry[1])),
            ["<", "<", ">", ">"],
        ) class PoolActions:
    """
    Pool actions.
    """

    @staticmethod
    def create_pool(namespace):
        """
        Create a stratis pool.

        :raises StratisCliEngineError:
        :raises StratisCliIncoherenceError:
        :raises StratisCliNameConflictError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import Manager, ObjectManager, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
        pool_name = namespace.pool_name
        names = pools(props={"Name": pool_name}).search(managed_objects)
        blockdevs = frozenset([os.path.abspath(p) for p in namespace.blockdevs])
        if list(names) != []:
            raise StratisCliNameConflictError("pool", pool_name)

        _check_opposite_tier(managed_objects, blockdevs, BlockDevTiers.CACHE)

        _check_same_tier(pool_name, managed_objects, blockdevs, BlockDevTiers.DATA)

        clevis_info = get_clevis_info(namespace)

        ((changed, (_, _)), return_code, message) = Manager.Methods.CreatePool(
            proxy,
            {
                "name": pool_name,
                "redundancy": (True, 0),
                "devices": blockdevs,
                "key_desc": (
                    (True, namespace.key_desc)
                    if namespace.key_desc is not None
                    else (False, "")
                ),
                "clevis_info": (False, ("", ""))
                if clevis_info is None
                else (True, clevis_info),
            },
        )

        if return_code != StratisdErrors.OK:  # pragma: no cover
            raise StratisCliEngineError(return_code, message)

        if not changed:  # pragma: no cover
            raise StratisCliIncoherenceError(
                (
                    "Expected to create the specified pool %s but stratisd "
                    "reports that it did not actually create the pool"
                )
                % pool_name
            )

    @staticmethod
    def init_cache(namespace):  # pylint: disable=too-many-locals
        """
        Initialize the cache of an existing stratis pool.

        :raises StratisCliEngineError:
        :raises StratisCliIncoherenceError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import MODev, ObjectManager, Pool, devs, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
        pool_name = namespace.pool_name
        (pool_object_path, _) = next(
            pools(props={"Name": pool_name})
            .require_unique_match(True)
            .search(managed_objects)
        )
        blockdevs = frozenset([os.path.abspath(p) for p in namespace.blockdevs])

        _check_opposite_tier(managed_objects, blockdevs, BlockDevTiers.DATA)

        _check_same_tier(pool_name, managed_objects, blockdevs, BlockDevTiers.CACHE)

        ((changed, devs_added), return_code, message) = Pool.Methods.InitCache(
            get_object(pool_object_path), {"devices": blockdevs}
        )

        if return_code != StratisdErrors.OK:
            raise StratisCliEngineError(return_code, message)

        if not changed or len(devs_added) < len(blockdevs):  # pragma: no cover
            devnodes_added = [
                MODev(info).Devnode()
                for (object_path, info) in devs(
                    props={"Pool": pool_object_path}
                ).search(ObjectManager.Methods.GetManagedObjects(proxy, {}))
                if object_path in devs_added
            ]
            raise StratisCliIncoherenceError(
                (
                    "Expected to add the specified blockdevs as cache "
                    "to pool %s but stratisd reports that it did not actually "
                    "add some or all of the blockdevs requested; devices "
                    "added: (%s), devices requested: (%s)"
                )
                % (namespace.pool_name, ", ".join(devnodes_added), ", ".join(blockdevs))
            )

    @staticmethod
    def list_pools(namespace):
        """
        List all stratis pools.
        """
        # pylint: disable=import-outside-toplevel
        from ._data import MOPool, ObjectManager, pools

        proxy = get_object(TOP_OBJECT)

        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
        pools_with_props = [
            MOPool(info) for objpath, info in pools().search(managed_objects)
        ]

        def physical_size_triple(mopool):
            """
            Calculate the triple to display for total physical size.

            The format is total/used/free where the display value for each
            member of the tuple are chosen automatically according to justbytes'
            configuration.

            :param mopool: an object representing all the properties of the pool
            :type mopool: MOPool
            :returns: a string to display in the resulting list output
            :rtype: str
            """
            total_physical_size = Range(mopool.TotalPhysicalSize())
            total_physical_used = get_property(mopool.TotalPhysicalUsed(), Range, None)
            return size_triple(total_physical_size, total_physical_used)

        def properties_string(mopool):
            """
            Make a string encoding some important properties of the pool

            :param mopool: an object representing all the properties of the pool
            :type mopool: MOPool
            :param props_map: a map of properties returned by GetAllProperties
            :type props_map: dict of str * any
            """

            def gen_string(has_property, code):
                """
                Generate the display string for a boolean property

                :param has_property: whether the property is true or false
                :type has_property: bool or NoneType
                :param str code: the code to generate the string for
                :returns: the generated string
                :rtype: str
                """
                if has_property == True:  # pylint: disable=singleton-comparison
                    prefix = " "
                elif has_property == False:  # pylint: disable=singleton-comparison
                    prefix = "~"
                # This is only going to occur if the engine experiences an
                # error while calculating a property or if our code has a bug.
                else:  # pragma: no cover
                    prefix = "?"
                return prefix + code

            props_list = [(mopool.HasCache(), "Ca"), (mopool.Encrypted(), "Cr")]
            return ",".join(gen_string(x, y) for x, y in props_list)

        format_uuid = (
            (lambda mo_uuid: mo_uuid) if namespace.unhyphenated_uuids else to_hyphenated
        )

        def alert_string(mopool):
            """
            Alert information to display, if any

            :param mopool: object to access pool properties

            :returns: string w/ alert information, "" if no alert
            :rtype: str
            """
            action_availability = PoolActionAvailability.from_str(
                mopool.AvailableActions()
            )
            error_codes = action_availability.pool_maintenance_error_codes()
            return ", ".join(sorted(str(code) for code in error_codes))

        tables = [
            (
                mopool.Name(),
                physical_size_triple(mopool),
                properties_string(mopool),
                format_uuid(mopool.Uuid()),
                alert_string(mopool),
            )
            for mopool in pools_with_props
        ]

        print_table(
            ["Name", "Total Physical", "Properties", "UUID", "Alerts"],
            sorted(tables, key=lambda entry: entry[0]),
            ["<", ">", ">", ">", "<"],
        )

    @staticmethod
    def destroy_pool(namespace):
        """
        Destroy a stratis pool.

        If no pool exists, the method succeeds.

        :raises StratisCliEngineError:
        :raises StratisCliIncoherenceError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import Manager, ObjectManager, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
        (pool_object_path, _) = next(
            pools(props={"Name": namespace.pool_name})
            .require_unique_match(True)
            .search(managed_objects)
        )

        ((changed, _), return_code, message) = Manager.Methods.DestroyPool(
            proxy, {"pool": pool_object_path}
        )

        # This branch can be covered, since the engine will return an error
        # if the pool can not be destroyed because it has filesystems.
        if return_code != StratisdErrors.OK:
            raise StratisCliEngineError(return_code, message)

        if not changed:  # pragma: no cover
            raise StratisCliIncoherenceError(
                (
                    "Expected to destroy the specified pool %s but "
                    "stratisd reports that it did not actually "
                    "destroy the pool requested"
                )
                % namespace.pool_name
            )

    @staticmethod
    def rename_pool(namespace):
        """
        Rename a pool.

        :raises StratisCliEngineError:
        :raises StratisCliNoChangeError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import ObjectManager, Pool, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
        (pool_object_path, _) = next(
            pools(props={"Name": namespace.current})
            .require_unique_match(True)
            .search(managed_objects)
        )

        ((changed, _), return_code, message) = Pool.Methods.SetName(
            get_object(pool_object_path), {"name": namespace.new}
        )

        if return_code != StratisdErrors.OK:  # pragma: no cover
            raise StratisCliEngineError(return_code, message)

        if not changed:
            raise StratisCliNoChangeError("rename", namespace.new)

    @staticmethod
    def add_data_devices(namespace):  # pylint: disable=too-many-locals
        """
        Add specified data devices to a pool.

        :raises StratisCliEngineError:
        :raises StratisCliIncoherenceError:
        :raises StratisCliInUseOtherTierError:
        :raises StratisCliInUseSameTierError:
        :raises StratisCliPartialChangeError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import MODev, ObjectManager, Pool, devs, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})

        blockdevs = frozenset([os.path.abspath(p) for p in namespace.blockdevs])

        _check_opposite_tier(managed_objects, blockdevs, BlockDevTiers.CACHE)

        _check_same_tier(
            namespace.pool_name, managed_objects, blockdevs, BlockDevTiers.DATA
        )

        (pool_object_path, _) = next(
            pools(props={"Name": namespace.pool_name})
            .require_unique_match(True)
            .search(managed_objects)
        )

        ((added, devs_added), return_code, message) = Pool.Methods.AddDataDevs(
            get_object(pool_object_path), {"devices": list(blockdevs)}
        )
        if return_code != StratisdErrors.OK:  # pragma: no cover
            raise StratisCliEngineError(return_code, message)

        if not added or len(devs_added) < len(blockdevs):  # pragma: no cover
            devnodes_added = [
                MODev(info).Devnode()
                for (object_path, info) in devs(
                    props={"Pool": pool_object_path}
                ).search(ObjectManager.Methods.GetManagedObjects(proxy, {}))
                if object_path in devs_added
            ]
            raise StratisCliIncoherenceError(
                (
                    "Expected to add the specified blockdevs to the data tier "
                    "in pool %s but stratisd reports that it did not actually "
                    "add some or all of the blockdevs requested; devices "
                    "added: (%s), devices requested: (%s)"
                )
                % (namespace.pool_name, ", ".join(devnodes_added), ", ".join(blockdevs))
            )

    @staticmethod
    def add_cache_devices(namespace):  # pylint: disable=too-many-locals
        """
        Add specified cache devices to a pool.

        :raises StratisCliEngineError:
        :raises StratisCliIncoherenceError:
        :raises StratisCliInUseOtherTierError:
        :raises StratisCliInUseSameTierError:
        :raises StratisCliPartialChangeError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import MODev, ObjectManager, Pool, devs, pools

        proxy = get_object(TOP_OBJECT)
        managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})

        blockdevs = frozenset([os.path.abspath(p) for p in namespace.blockdevs])

        _check_opposite_tier(managed_objects, blockdevs, BlockDevTiers.DATA)

        _check_same_tier(
            namespace.pool_name, managed_objects, blockdevs, BlockDevTiers.CACHE
        )

        (pool_object_path, _) = next(
            pools(props={"Name": namespace.pool_name})
            .require_unique_match(True)
            .search(managed_objects)
        )

        ((added, devs_added), return_code, message) = Pool.Methods.AddCacheDevs(
            get_object(pool_object_path), {"devices": list(blockdevs)}
        )
        if return_code != StratisdErrors.OK:
            raise StratisCliEngineError(return_code, message)

        if not added or len(devs_added) < len(blockdevs):  # pragma: no cover
            devnodes_added = [
                MODev(info).Devnode()
                for (object_path, info) in devs(
                    props={"Pool": pool_object_path}
                ).search(ObjectManager.Methods.GetManagedObjects(proxy, {}))
                if object_path in devs_added
            ]
            raise StratisCliIncoherenceError(
                (
                    "Expected to add the specified blockdevs to the cache tier "
                    "in pool %s but stratisd reports that it did not actually "
                    "add some or all of the blockdevs requested; devices "
                    "added: (%s), devices requested: (%s)"
                )
                % (namespace.pool_name, ", ".join(devnodes_added), ", ".join(blockdevs))
            )

    @staticmethod
    def unlock_pools(namespace):
        """
        Unlock all of the encrypted pools that have been detected by the daemon
        but are still locked.
        :raises StratisCliIncoherenceError:
        :raises StratisCliNoChangeError:
        :raises StratisCliAggregateError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import Manager

        proxy = get_object(TOP_OBJECT)

        locked_pools = _fetch_locked_pools_property(proxy)
        if locked_pools == {}:  # pragma: no cover
            raise StratisCliNoChangeError("unlock", "pools")

        # This block is not covered as the sim engine does not simulate the
        # management of unlocked devices, so locked_pools is always empty.
        errors = []  # pragma: no cover
        for uuid in locked_pools:  # pragma: no cover
            (
                (is_some, unlocked_devices),
                return_code,
                message,
            ) = Manager.Methods.UnlockPool(
                proxy, {"pool_uuid": uuid, "unlock_method": namespace.unlock_method}
            )

            if return_code != StratisdErrors.OK:
                errors.append(
                    StratisCliPartialFailureError(
                        "unlock", "pool with UUID %s" % uuid, error_message=message
                    )
                )

            if is_some and unlocked_devices == []:
                raise StratisCliIncoherenceError(
                    (
                        "stratisd reported that some existing devices are locked but "
                        "no new devices were unlocked during this operation"
                    )
                )

        if errors != []:  # pragma: no cover
            raise StratisCliAggregateError("unlock", "pool", errors)

    @staticmethod
    def explain_code(namespace):
        """
        Print an explanation of pool error code.
        """
        error_code = PoolMaintenanceErrorCode.from_str(namespace.code)
        assert error_code is not None
        print(error_code.explain()) class StratisActions:
    """
    Stratis actions.
    """

    @staticmethod
    def list_stratisd_redundancy(_namespace):
        """
        List the stratisd redundancy designations.
        """
        for code in RedundancyCodes:
            print("%s: %d" % (code.name, code.value))

    @staticmethod
    def list_stratisd_version(_namespace):
        """
        List the stratisd version.
        """
        # pylint: disable=import-outside-toplevel
        from ._data import Manager

        print("%s" % Manager.Properties.Version.Get(get_object(TOP_OBJECT))) def check_stratisd_version():
    """
    Checks that the version of stratisd that is running is compatible with
    this version of the CLI.

    :raises StratisCliStratisdVersionError
    """
    # pylint: disable=import-outside-toplevel
    from ._data import Manager0

    version_spec = SpecifierSet(">=%s" % MINIMUM_STRATISD_VERSION) & SpecifierSet(
        "<%s" % MAXIMUM_STRATISD_VERSION
    )
    version = Manager0.Properties.Version.Get(get_object(TOP_OBJECT))

    if Version(version) not in version_spec:
        raise StratisCliStratisdVersionError(
            version, MINIMUM_STRATISD_VERSION, MAXIMUM_STRATISD_VERSION
        ) class TopActions:
    """
    Top level actions.
    """

    @staticmethod
    def get_report(namespace):
        """
        Get the requested report from stratisd.

        :raises StratisCliEngineError:
        """

        # pylint: disable=import-outside-toplevel
        if namespace.report_name == str(ReportKey.MANAGED_OBJECTS):

            from ._data import ObjectManager

            json_report = ObjectManager.Methods.GetManagedObjects(
                get_object(TOP_OBJECT), {}
            )

        else:

            if namespace.report_name == str(ReportKey.ENGINE_STATE):

                from ._data import Manager

                (report, return_code, message) = Manager.Methods.EngineStateReport(
                    get_object(TOP_OBJECT), {}
                )

            else:

                from ._data import Report

                (report, return_code, message) = Report.Methods.GetReport(
                    get_object(TOP_OBJECT), {"name": namespace.report_name}
                )

            # The only reason that stratisd has for returning an error code is
            # if the report name is unrecognized. However, the parser restricts
            # the list of names to only the ones that stratisd recognizes, so
            # this branch can only be taken due to an unexpected bug in
            # stratisd.
            if return_code != StratisdErrors.OK:  # pragma: no cover
                raise StratisCliEngineError(return_code, message)

            json_report = json.loads(report)

        json.dump(json_report, sys.stdout, indent=4)
        print(file=sys.stdout)

    @staticmethod
    def set_key(namespace):
        """
        Set a key in the kernel keyring.

        :raises StratisCliEngineError:
        :raises StratisCliNameConflictError:
        :raises StratisCliIncoherenceError:
        """
        proxy = get_object(TOP_OBJECT)

        key_list = _fetch_keylist(proxy)
        if namespace.keydesc in key_list:
            raise StratisCliNameConflictError("key", namespace.keydesc)

        ((changed, existing_modified), return_code, message) = _add_update_key(
            proxy,
            namespace.keydesc,
            namespace.capture_key,
            keyfile_path=namespace.keyfile_path,
        )

        if return_code != StratisdErrors.OK:
            raise StratisCliEngineError(return_code, message)

        if not changed and not existing_modified:  # pragma: no cover
            raise StratisCliIncoherenceError(
                (
                    "Key %s was reported to not exist but stratisd reported "
                    "that no change was made to the key"
                )
                % namespace.keydesc
            )

        # A key was updated even though there was no key reported to already exist.
        if changed and existing_modified:  # pragma: no cover
            raise StratisCliIncoherenceError(
                (
                    "Key %s was reported to not exist but stratisd reported "
                    "that it reset an existing key"
                )
                % namespace.keydesc
            )

    @staticmethod
    def reset_key(namespace):
        """
        Reset the key data for an existing key in the kernel keyring.

        :raises StratisCliEngineError:
        :raises StratisCliResourceNotFoundError:
        :raises StratisCliIncoherenceError:
        """
        proxy = get_object(TOP_OBJECT)

        key_list = _fetch_keylist(proxy)
        if namespace.keydesc not in key_list:
            raise StratisCliResourceNotFoundError("reset", namespace.keydesc)

        ((changed, existing_modified), return_code, message) = _add_update_key(
            proxy,
            namespace.keydesc,
            namespace.capture_key,
            keyfile_path=namespace.keyfile_path,
        )

        if return_code != StratisdErrors.OK:
            raise StratisCliEngineError(return_code, message)

        # The key description existed and the key was not changed.
        if not changed and not existing_modified:
            raise StratisCliNoChangeError("reset", namespace.keydesc)

        # A new key was added even though there was a key reported to already exist.
        if changed and not existing_modified:  # pragma: no cover
            raise StratisCliIncoherenceError(
                (
                    "Key %s was reported to already exist but stratisd reported "
                    "that it created a new key"
                )
                % namespace.keydesc
            )

    @staticmethod
    def unset_key(namespace):
        """
        Unset a key in kernel keyring.

        :raises StratisCliEngineError:
        :raises StratisCliNoChangeError:
        :raises StratisCliIncoherenceError:
        """
        # pylint: disable=import-outside-toplevel
        from ._data import Manager

        proxy = get_object(TOP_OBJECT)

        key_list = _fetch_keylist(proxy)
        if namespace.keydesc not in key_list:
            raise StratisCliNoChangeError("remove", namespace.keydesc)

        (changed, return_code, message) = Manager.Methods.UnsetKey(
            proxy, {"key_desc": namespace.keydesc}
        )

        if return_code != StratisdErrors.OK:  # pragma: no cover
            raise StratisCliEngineError(return_code, message)

        if not changed:  # pragma: no cover
            raise StratisCliIncoherenceError(
                (
                    "Key %s was reported to exist but stratisd reported "
                    "that no key was unset"
                )
                % namespace.keydesc
            )

    @staticmethod
    def list_keys(_):
        """
        List keys in kernel keyring.

        :raises StratisCliEngineError:
        """
        proxy = get_object(TOP_OBJECT)

        key_list = [[key_desc] for key_desc in _fetch_keylist(proxy)]

        print_table(
            ["Key Description"], sorted(key_list, key=lambda entry: entry[0]), ["<"]
        ) class ReportKey(Enum):
    """
    Report identifiers.

    Note: "managed_objects_report" is not a key recognized by stratisd.
    However, since the other constants are, and they are all used together,
    this type is defined with the other stratisd constants.
    """

    ENGINE_STATE = "engine_state_report"
    ERRORED_POOL = "errored_pool_report"
    MANAGED_OBJECTS = "managed_objects_report"

    def __str__(self):
        return self.value KEY_SUBCMDS = [
    (
        "set",
        dict(
            help="Set a key in the kernel keyring",
            args=[("keydesc", dict(action="store", help="key description"))],
            mut_ex_args=[
                (
                    True,
                    [
                        (
                            "--keyfile-path",
                            dict(
                                action="store",
                                help=(
                                    "Path to the key file containing a key to set "
                                    "in the keyring"
                                ),
                                dest="keyfile_path",
                            ),
                        ),
                        (
                            "--capture-key",
                            dict(
                                action="store_true",
                                help=(
                                    "Read key from stdin with no terminal echo or "
                                    "userspace buffer storage"
                                ),
                                dest="capture_key",
                            ),
                        ),
                    ],
                )
            ],
            func=TopActions.set_key,
        ),
    ),
    (
        "reset",
        dict(
            help="Reset an existing key in the kernel keyring",
            args=[("keydesc", dict(action="store", help="key description"))],
            mut_ex_args=[
                (
                    True,
                    [
                        (
                            "--keyfile-path",
                            dict(
                                action="store",
                                help=(
                                    "Path to the key file containing a key to reset "
                                    "in the keyring"
                                ),
                                dest="keyfile_path",
                            ),
                        ),
                        (
                            "--capture-key",
                            dict(
                                action="store_true",
                                help=(
                                    "Read key from stdin with no terminal echo or "
                                    "userspace buffer storage"
                                ),
                                dest="capture_key",
                            ),
                        ),
                    ],
                )
            ],
            func=TopActions.reset_key,
        ),
    ),
    (
        "unset",
        dict(
            help="Unset a key in the kernel keyring",
            args=[("keydesc", dict(action="store", help="key description"))],
            func=TopActions.unset_key,
        ),
    ),
    (
        "list",
        dict(help="List Stratis keys in kernel keyring", func=TopActions.list_keys),
    ),
] LOGICAL_SUBCMDS = [
    (
        "create",
        dict(
            help="Create filesystems in a pool",
            args=[
                ("pool_name", dict(action="store", help="pool name")),
                (
                    "fs_name",
                    dict(
                        help="Create filesystems in this pool using the given names",
                        nargs="+",
                    ),
                ),
                (
                    "--size",
                    dict(
                        action=RangeAction,
                        dest="size",
                        help='The size of the filesystems to be created, e.g., "32GiB"',
                    ),
                ),
            ],
            func=LogicalActions.create_volumes,
        ),
    ),
    (
        "snapshot",
        dict(
            help="Snapshot the named filesystem in a pool",
            args=[
                ("pool_name", dict(action="store", help="pool name")),
                ("origin_name", dict(action="store", help="origin name")),
                ("snapshot_name", dict(action="store", help="snapshot name")),
            ],
            func=LogicalActions.snapshot_filesystem,
        ),
    ),
    (
        "list",
        dict(
            help="List filesystems",
            args=[
                (
                    "pool_name",
                    dict(action="store", default=None, nargs="?", help="Pool name"),
                )
            ],
            func=LogicalActions.list_volumes,
        ),
    ),
    (
        "destroy",
        dict(
            help="Destroy filesystems in a pool",
            args=[
                ("pool_name", dict(action="store", help="pool name")),
                (
                    "fs_name",
                    dict(help="Destroy the named filesystems in this pool", nargs="+"),
                ),
            ],
            func=LogicalActions.destroy_volumes,
        ),
    ),
    (
        "rename",
        dict(
            help="Rename a filesystem",
            args=[
                (
                    "pool_name",
                    dict(
                        action="store",
                        help="Name of the pool the filesystem is part of",
                    ),
                ),
                (
                    "fs_name",
                    dict(action="store", help="Name of the filesystem to change"),
                ),
                (
                    "new_name",
                    dict(action="store", help="New name to give that filesystem"),
                ),
            ],
            func=LogicalActions.rename_fs,
        ),
    ),
] PHYSICAL_SUBCMDS = [
    (
        "list",
        dict(
            help="List information about blockdevs in the pool",
            args=[
                (
                    "pool_name",
                    dict(action="store", default=None, nargs="?", help="Pool name"),
                )
            ],
            func=PhysicalActions.list_devices,
        ),
    )
] POOL_SUBCMDS = [
    (
        "create",
        dict(
            help="Create a pool",
            args=[
                ("pool_name", dict(action="store", help="Name of new pool")),
                (
                    "blockdevs",
                    dict(help="Create the pool using these block devs", nargs="+"),
                ),
                (
                    "--redundancy",
                    dict(
                        action="store",
                        choices=["none"],
                        default="none",
                        help="Redundancy level for this pool",
                    ),
                ),
                (
                    "--key-desc",
                    dict(
                        default=None,
                        type=str,
                        help=(
                            "Key description of key in kernel keyring to use "
                            "for encryption"
                        ),
                        dest="key_desc",
                    ),
                ),
                (
                    "--clevis",
                    dict(
                        default=None,
                        type=str,
                        help=("Specification for binding with Clevis."),
                        dest="clevis",
                        choices=["nbde", "tang", "tpm2"],
                    ),
                ),
                (
                    "--tang-url",
                    dict(
                        default=None,
                        type=str,
                        help=(
                            "URL of Clevis tang server (ignored if "
                            "--clevis=[tang|nbde] not set)"
                        ),
                        dest="tang_url",
                    ),
                ),
            ],
            mut_ex_args=[
                (
                    False,
                    [
                        (
                            "--trust-url",
                            dict(
                                action="store_true",
                                help=(
                                    "Omit verification of tang server "
                                    "credentials (ignored if "
                                    "--clevis=[tang|nbde] not set)"
                                ),
                                dest="trust_url",
                            ),
                        ),
                        (
                            "--thumbprint",
                            dict(
                                action="store",
                                help=(
                                    "Thumbprint of tang server at specified "
                                    "URL (ignored if --clevis=[tang|nbde] not "
                                    "set)"
                                ),
                                dest="thumbprint",
                            ),
                        ),
                    ],
                )
            ],
            func=PoolActions.create_pool,
        ),
    ),
    (
        "init-cache",
        dict(
            help="Initialize the cache with block devices",
            args=[
                (
                    "pool_name",
                    dict(
                        action="store",
                        help=("Name of the pool for which to initialize the cache"),
                    ),
                ),
                (
                    "blockdevs",
                    dict(
                        help="Initialize the pool cache using these block devs",
                        nargs="+",
                    ),
                ),
            ],
            func=PoolActions.init_cache,
        ),
    ),
    (
        "list",
        dict(
            help="List pools",
            description="Lists Stratis pools that exist on the system",
            func=PoolActions.list_pools,
        ),
    ),
    (
        "destroy",
        dict(
            help="Destroy a pool",
            args=[("pool_name", dict(action="store", help="pool name"))],
            func=PoolActions.destroy_pool,
        ),
    ),
    (
        "rename",
        dict(
            help="Rename a pool",
            args=[
                ("current", dict(action="store", help="Current pool name")),
                ("new", dict(action="store", help="New pool name")),
            ],
            func=PoolActions.rename_pool,
        ),
    ),
    (
        "add-data",
        dict(
            help="Add one or more blockdevs to an existing pool for use as data storage",
            args=[
                ("pool_name", dict(action="store", help="Pool name")),
                (
                    "blockdevs",
                    dict(
                        help="Block devices to add to the pool",
                        metavar="blockdev",
                        nargs="+",
                    ),
                ),
            ],
            func=PoolActions.add_data_devices,
        ),
    ),
    (
        "add-cache",
        dict(
            help="Add one or more blockdevs to an existing pool for use as cache",
            args=[
                ("pool_name", dict(action="store", help="Pool name")),
                (
                    "blockdevs",
                    dict(
                        help="Block devices to add to the pool as cache",
                        metavar="blockdev",
                        nargs="+",
                    ),
                ),
            ],
            func=PoolActions.add_cache_devices,
        ),
    ),
    (
        "bind",
        dict(
            help="Bind the given pool with an additional encryption facility",
            subcmds=BIND_SUBCMDS,
        ),
    ),
    (
        "rebind",
        dict(
            help=(
                "Rebind the given pool with a currently in use encryption "
                "facility but new credentials"
            ),
            subcmds=REBIND_SUBCMDS,
        ),
    ),
    (
        "unbind",
        dict(
            help="Unbind the given pool, removing support for encryption with Clevis",
            args=[
                (
                    "method",
                    dict(
                        action="store",
                        choices=[str(x) for x in list(EncryptionMethod)],
                        help="Encryption method to unbind",
                    ),
                ),
                ("pool_name", dict(action="store", help="Pool name")),
            ],
            func=BindActions.unbind,
        ),
    ),
    (
        "unlock",
        dict(
            help="Unlock all of the currently locked encrypted pools",
            args=[
                (
                    "unlock_method",
                    dict(
                        default=str(EncryptionMethod.KEYRING),
                        action="store",
                        choices=[str(x) for x in list(EncryptionMethod)],
                        help="Method to use to unlock encrypted pools",
                    ),
                )
            ],
            func=PoolActions.unlock_pools,
        ),
    ),
    (
        "explain",
        dict(
            help="Explain pool alert codes",
            args=[
                (
                    "code",
                    dict(
                        action="store",
                        choices=[str(x) for x in list(PoolMaintenanceErrorCode)],
                        help="Error code to explain",
                    ),
                )
            ],
            func=PoolActions.explain_code,
        ),
    ),
]
import argparse
from .._actions import (
    LogicalActions,
    PhysicalActions,
    PoolActions,
    StratisActions,
    TopActions,
    check_stratisd_version,
)
from .._stratisd_constants import ReportKey
from .._version import __version__
from ._key import KEY_SUBCMDS
from ._logical import LOGICAL_SUBCMDS
from ._physical import PHYSICAL_SUBCMDS
from ._pool import POOL_SUBCMDS
# Copyright 2016 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.
"""
Top level parser for Stratis CLI.
"""

# isort: STDLIB


PRINT_HELP = lambda parser: lambda _: parser.error("missing sub-command")


def _add_args(parser, args):
    """
    Call subcommand.add_argument() based on args list.

    :param parser: the parser being build
    :param list args: a data structure representing the arguments to be added
    """
    for name, arg in args:
        parser.add_argument(name, **arg)


def _add_mut_ex_args(parser, args):
    """
    Add mututally exlusive arguments for a subcommand.

    :param parser: the parser being build
    :param args: a data structure representing sets of mututally exclusive
                 arguments to be added
    :type args: list of bool * (list of dict)
    """
    for (one_is_required, arg_list) in args:
        group = parser.add_mutually_exclusive_group(required=one_is_required)
        for name, arg in arg_list:
            group.add_argument(name, **arg)


def add_subcommand(subparser, cmd):
    """
    Add subcommand to a parser based on a subcommand dict.
    """
    name, info = cmd
    parser = subparser.add_parser(
        name, help=info["help"], aliases=info.get("aliases", [])
    )

    subcmds = info.get("subcmds")
    if subcmds is not None:
        subparsers = parser.add_subparsers(title="subcommands")
        for subcmd in subcmds:
            add_subcommand(subparsers, subcmd)

    _add_args(parser, info.get("args", []))
    _add_mut_ex_args(parser, info.get("mut_ex_args", []))

    def wrap_func(func):
        if func is None:
            return PRINT_HELP(parser)

        def wrapped_func(*args):
            check_stratisd_version()
            func(*args)

        return wrapped_func

    parser.set_defaults(func=wrap_func(info.get("func")))


DAEMON_SUBCMDS = [
    (
        "redundancy",
        dict(
            help="Redundancy designations understood by stratisd daemon",
            func=StratisActions.list_stratisd_redundancy,
        ),
    ),
    (
        "version",
        dict(

            help="version of stratisd daemon", func=StratisActions.list_stratisd_version