class PoolMaintenanceErrorCode(IntEnum):
    """
    Maintenance error codes for the pool.
    """

    NO_IPC_REQUESTS = 1
    NO_POOL_CHANGES = 2
    READ_ONLY = 3

    def __str__(self):
        return "EM%s" % str(self.value).zfill(3)

    @staticmethod
    def from_str(code_str):
        """
        Discover the code, if any, from the code string.

        :returns: the code if it finds a match, otherwise None
        :rtype: PoolMaintenanceErrorCode or NoneType
        """
        for code in PoolMaintenanceErrorCode:
            if code_str == str(code):
                return code
        return None

    def explain(self):
        """
        Return an explanation of the error return code.
        """
        if self is PoolMaintenanceErrorCode.NO_IPC_REQUESTS:
            return (
                "The pool will return an error on any IPC request that could "
                "cause a change in the pool state, for example, a request to "
                "rename a filesystem. It will still be able to respond to "
                "purely informational requests."
            )

        if self is PoolMaintenanceErrorCode.NO_POOL_CHANGES:
            return (
                "The pool is unable to manage itself by reacting to events, "
                "such as devicemapper events, that might require it to take "
                "any maintenance operations."
            )

        if self is PoolMaintenanceErrorCode.READ_ONLY:  # pragma: no cover
            return "The pool is in read-only mode."

        assert False, "impossible error code reached"  # pragma: no cover class StratisCliAggregateError(StratisCliRuntimeError):
    """
    Raised when multiple errors have occured in a looping operation.
    """

    def __init__(self, operation, type_, errors):
        """
        Initializer.
        :param str operation: the looping operation that failed.
        :param str type_: the type of the resource for which the operation failed.
        :param errors: list of all errors that occurred during the looping operation.
        :type errors: list of Exception
        """
        # pylint: disable=super-init-not-called
        self.operation = operation
        self.type = type_
        self.errors = errors

    def __str__(self):
        return (
            "The operation '%s' on a resource of type %s failed. The following "
            "errors occurred:\n%s"
        ) % (
            self.operation,
            self.type,
            os.linesep.join([str(error) for error in self.errors]),
        ) class StratisCliEngineError(StratisCliRuntimeError):
    """
    Raised if there was a failure due to an error in stratisd's engine.
    """

    def __init__(self, rc, message):
        """
        Initializer.

        :param rc int: the error code returned by the engine
        :param str message: whatever message accompanied the error code
        """
        # pylint: disable=super-init-not-called
        self.error_code = rc
        self.message = message

    def __str__(self):
        return "%s: %s" % (STRATISD_ERROR_TO_NAME(self.error_code), self.message) class StratisCliIncoherenceError(StratisCliRuntimeError):
    """
    Raised if there was a disagreement about state between the CLI
    and the daemon. This can happen if the information derived from
    the GetManagedObjects() result does not correspond with the daemon's
    state. This is a very unlikely event, but could occur if two processes
    were issuing instructions to stratisd at the same time.
    """ class StratisCliInUseOtherTierError(StratisCliInUseError):
    """
    Raised if a request made of stratisd must result in a device being
    included in both data and cache tiers.
    """

    def __init__(self, pools_to_blockdevs, added_as):
        """
        Initializer.

        :param pools_to_blockdevs: pools mapped to the blockdevs they own
        :type pools_to_blockdevs: dict of str * frozenset of str
        :param added_as: what tier the devices were to be added to
        :type added_as: _stratisd_constants.BlockDevTiers

        Precondition: pools_to_blockdevs != {}
        Precondition: all frozenset of str in pools_to_blockdevs have at least one item
        """
        # pylint: disable=super-init-not-called
        self.pools_to_blockdevs = pools_to_blockdevs
        self.added_as = added_as

    def __str__(self):
        (target_blockdev_tier, already_blockdev_tier) = (
            BLOCK_DEV_TIER_TO_NAME(self.added_as),
            BLOCK_DEV_TIER_TO_NAME(
                BlockDevTiers.DATA
                if self.added_as == BlockDevTiers.CACHE
                else BlockDevTiers.CACHE
            ),
        )

        msg = (
            "At least one of the provided devices is already owned by an "
            "existing pool's %s tier" % (already_blockdev_tier,)
        )

        for pool_name, blockdevs in self.pools_to_blockdevs.items():
            if len(blockdevs) > 1:
                msg += (
                    "; devices %s would be added to the %s tier but are "
                    "already in use in the %s tier of pool %s"
                    % (
                        list(blockdevs),
                        target_blockdev_tier,
                        already_blockdev_tier,
                        pool_name,
                    )
                )
            else:
                msg += (
                    "; device %s would be added to the %s tier but is "
                    "already in use in the %s tier of pool %s"
                    % (
                        list(blockdevs)[0],
                        target_blockdev_tier,
                        already_blockdev_tier,
                        pool_name,
                    )
                )

        return msg class StratisCliInUseSameTierError(StratisCliInUseError):
    """
    Raised if a request made of stratisd must result in a device being
    included in the same tier in two different pools.
    """

    def __init__(self, pools_to_blockdevs, added_as):
        """
        Initializer.

        :param pools_to_blockdevs: pools mapped to the blockdevs they own
        :type pools_to_blockdevs: dict of str * frozenset of str
        :param added_as: what tier the devices were to be added to
        :type added_as: _stratisd_constants.BlockDevTiers

        Precondition: pools_to_blockdevs != {}
        Precondition: all frozenset of str in pools_to_blockdevs have at least one item
        """
        # pylint: disable=super-init-not-called
        self.pools_to_blockdevs = pools_to_blockdevs
        self.added_as = added_as

    def __str__(self):
        blockdev_tier = BLOCK_DEV_TIER_TO_NAME(self.added_as)

        msg = (
            "At least one of the provided devices is already owned by an "
            "existing pool's %s tier" % (blockdev_tier,)
        )

        for pool_name, blockdevs in self.pools_to_blockdevs.items():
            if len(blockdevs) > 1:
                msg += (
                    "; devices %s would be added to the %s tier but are "
                    "already in use in the %s tier of pool %s"
                    % (list(blockdevs), blockdev_tier, blockdev_tier, pool_name)
                )
            else:
                msg += (
                    "; device %s would be added to the %s tier but is "
                    "already in use in the %s tier of pool %s"
                    % (list(blockdevs)[0], blockdev_tier, blockdev_tier, pool_name)
                )

        return msg class StratisCliNameConflictError(StratisCliUserError):
    """
    Raised if an item of the same name already exists.
    """

    def __init__(self, object_type, name):
        """
        Initializer.

        :param str object_type: the type of the object, pool, filesystem, etc.
        :param str name: the conflicting name
        """
        # pylint: disable=super-init-not-called
        self.object_type = object_type
        self.name = name

    def __str__(self):
        return "A %s named %s already exists" % (self.object_type, self.name) class StratisCliNoChangeError(StratisCliPartialChangeError):
    """
    Raised if a request made of stratisd must result in no change since the
    post-condition of the request is already entirely satisfied. This is
    a special case of StratisCliPartialChangeError, where the change requested
    is so simple that it can only succeed or fail.
    """

    def __init__(self, command, resource):
        """
        Initializer.

        :param str command: the executed command
        :param str resource: the target resource
        """
        StratisCliPartialChangeError.__init__(
            self, command, frozenset(), frozenset([resource])
        ) class StratisCliPartialChangeError(StratisCliUserError):
    """
    Raised if a request made of stratisd must result in a partial or no change
    since some or all of the post-condition for the request already holds.

    Invariant: self.unchanged_resources != frozenset()
    """

    def __init__(self, command, changed_resources, unchanged_resources):
        """
        Initializer.

        :param str command: the command run that caused the error
        :param changed_resources: the target resources that would change
        :type changed_resources: frozenset of str
        :param unchanged_resources: the target resources that would not change
        :type unchanged_resources: frozenset of str

        Precondition: unchanged_resources != frozenset()
        """
        # pylint: disable=super-init-not-called
        self.command = command
        self.changed_resources = changed_resources
        self.unchanged_resources = unchanged_resources

    def __str__(self):
        if len(self.unchanged_resources) > 1:
            msg = "The '%s' action has no effect for resources %s" % (
                self.command,
                list(self.unchanged_resources),
            )
        else:
            msg = "The '%s' action has no effect for resource %s" % (
                self.command,
                list(self.unchanged_resources)[0],
            )

        if self.changed_resources != frozenset():
            if len(self.changed_resources) > 1:
                msg += " but does for resources %s" % list(self.changed_resources)
            else:
                msg += " but does for resource %s" % list(self.changed_resources)[0]

        return msg class StratisCliPartialFailureError(StratisCliRuntimeError):
    """
    A non-fatal error to be reported at the end as part of a StratisCliAggregateError.
    """

    def __init__(self, action, identifier_string, error_message=None):
        """
        Initializer.
        :param str action: the action that failed
        :param str identifier_string: a string that uniquely identifies the resource
                                      for which the action failed
        :param error_message: an optional error message for the cause of the partial
                              failure
        :type error_message: NoneType or str
        """
        # pylint: disable=super-init-not-called
        self.action = action
        self.identifier_string = identifier_string
        self.error_message = error_message

    def __str__(self):
        fmt_str = 'Partial action "%s" failed for %s' % (
            self.action,
            self.identifier_string,
        )
        if self.error_message is not None:
            fmt_str += ": %s" % self.error_message
        return fmt_str class BlockDevTiers(IntEnum):
    """
    Tier to which a blockdev device belongs.
    """

    DATA = 0
    CACHE = 1 class PoolActionAvailability(IntEnum):
    """
    What category of interactions a pool is enabled for.
    """

    FULLY_OPERATIONAL = 0
    NO_IPC_REQUESTS = 1
    NO_POOL_CHANGES = 2
    NO_WRITE_IO = 3

    def __str__(self):
        if self is PoolActionAvailability.FULLY_OPERATIONAL:
            return "fully_operational"
        if self is PoolActionAvailability.NO_IPC_REQUESTS:
            return "no_ipc_requests"
        if self is PoolActionAvailability.NO_POOL_CHANGES:
            return "no_pool_changes"
        if self is PoolActionAvailability.NO_WRITE_IO:  # pragma: no cover
            return "no_write_io"

        assert False, "impossible value reached"  # pragma: no cover

    @staticmethod
    def from_str(code_str):
        """
        Get ActionAvailability object from a string.
        :param str code_str: a code string
        :rtype: str or NoneType
        """
        for item in list(PoolActionAvailability):
            if code_str == str(item):
                return item
        return None

    def pool_maintenance_error_codes(self):
        """
        Return the list of PoolMaintenanceErrorCodes for this availability.

        :rtype: list of PoolMaintenanceErrorCode
        """
        codes = []
        if self >= PoolActionAvailability.NO_IPC_REQUESTS:
            codes.append(PoolMaintenanceErrorCode.NO_IPC_REQUESTS)

        if self >= PoolActionAvailability.NO_POOL_CHANGES:
            codes.append(PoolMaintenanceErrorCode.NO_POOL_CHANGES)

        if self >= PoolActionAvailability.NO_WRITE_IO:
            codes.append(PoolMaintenanceErrorCode.READ_ONLY)

        return codes class StratisdErrors(IntEnum):
    """
    Stratisd Errors
    """

    OK = 0
    ERROR = 1 def get_object(object_path):
    """
    Get an object from an object path.

    :param str object_path: an object path with a valid format
    :returns: the proxy object corresponding to the object path
    :rtype: ProxyObject
    """
    return Bus.get_bus().get_object(SERVICE, object_path, introspect=False) TOP_OBJECT = "/org/storage/stratis3" def get_property(prop, to_repr, default):
    """
    Get a representation of an optional D-Bus property. An optional
    D-Bus property is one that may be unknown to stratisd.

    :param prop: the property value
    :type prop: a pair of bool * T
    :param to_repr: conversion function
    :type to_repr: object -> object
    :param object default: default for display
    :returns: object produced by to_repr or default
    :rtype: object
    """
    (valid, value) = prop
    return to_repr(value) if valid else default def print_table(column_headings, row_entries, alignment, file=sys.stdout):
    """
    Given the column headings and the row_entries, print a table.
    Align according to alignment specification and always pad with 2 spaces.

    :param column_headings: the column headings
    :type column_headings: list of str
    :param row_entries: a list of the row entries
    :type row_entries: list of list of str
    :param alignment: the alignment indicator for each key, '<', '>', '^', '='
    :type alignment: list of str
    :param file: file to print too
    :type file: writeable stream

    Precondition: len(column_headings) == len(alignment) == len of each entry
    in row_entries.

    Precondition: all(wcswidth(h) != -1 for h in column_headings)
                  all(wcswidth(i) != -1 for row in rows for item in row)
                  (i.e., no items to be printed contain unprintable characters)
    """
    column_widths = [0] * len(column_headings)
    cell_widths = []

    # Column header isn't different than any other row, insert into rows.
    row_entries.insert(0, column_headings)

    for row_index, row in enumerate(row_entries):
        cell_widths.append([])
        for column_index, cell in enumerate(row):
            cell_width = wcswidth(cell)
            cell_widths[row_index].append(cell_width)
            column_widths[column_index] = max(column_widths[column_index], cell_width)

    for row, row_widths in zip(row_entries, cell_widths):
        _print_row(file, row, row_widths, column_widths, alignment)
        print(file=file) def size_triple(size, used):
    """
    Given size and used, return a properly formatted string Total/ Used / Free

    :param size: total size
    :type size: Range or NoneType
    :param used: total amount used
    :type used: Range or NoneType
    :rtype: str
    :returns: formatted string for display
    """
    free = None if size is None or used is None else size - used

    return "%s / %s / %s" % (
        TABLE_FAILURE_STRING if size is None else size,
        TABLE_FAILURE_STRING if used is None else used,
        TABLE_FAILURE_STRING if free is None else free,
    ) def to_hyphenated(mo_uuid):
    """
    Convert uuid string to hyphenated format
    """
    return str(uuid.UUID(mo_uuid)) def get_clevis_info(namespace):
    """
    Get clevis info, if any, from the namespace.

    :param namespace: namespace set up by the parser

    :returns: clevis info or None
    :rtype: pair of str or NoneType
    """
    clevis_info = None
    if namespace.clevis is not None:
        if namespace.clevis in ("nbde", "tang"):
            if namespace.tang_url is None:
                raise StratisCliMissingClevisTangURLError()

            if not namespace.trust_url and namespace.thumbprint is None:
                raise StratisCliMissingClevisThumbprintError()

            clevis_config = {CLEVIS_KEY_URL: namespace.tang_url}
            if namespace.trust_url:
                clevis_config[CLEVIS_KEY_TANG_TRUST_URL] = True
            else:
                assert namespace.thumbprint is not None
                clevis_config[CLEVIS_KEY_THP] = namespace.thumbprint

            clevis_info = (CLEVIS_PIN_TANG, clevis_config)

        elif namespace.clevis == "tpm2":
            clevis_info = (CLEVIS_PIN_TPM2, {})

        else:
            raise AssertionError(
                "unexpected value %s for clevis option" % namespace.clevis
            )  # pragma: no cover

    return (
        clevis_info
        if clevis_info is None
        else (clevis_info[0], json.dumps(clevis_info[1]))
    )
import os
from collections import defaultdict
from justbytes import Range
from .._constants import PoolMaintenanceErrorCode
from .._errors import (
    StratisCliAggregateError,
    StratisCliEngineError,
    StratisCliIncoherenceError,
    StratisCliInUseOtherTierError,
    StratisCliInUseSameTierError,
    StratisCliNameConflictError,
    StratisCliNoChangeError,
    StratisCliPartialChangeError,
    StratisCliPartialFailureError,
)
from .._stratisd_constants import BlockDevTiers, PoolActionAvailability, StratisdErrors
from ._connection import get_object
from ._constants import TOP_OBJECT
from ._formatting import get_property, print_table, size_triple, to_hyphenated
from ._utils import get_clevis_info
    from ._data import MODev, MOPool, devs, pools
    from ._data import Manager
        from ._data import Manager, ObjectManager, pools
        from ._data import MODev, ObjectManager, Pool, devs, pools
        from ._data import MOPool, ObjectManager, pools
        from ._data import Manager, ObjectManager, pools
        from ._data import ObjectManager, Pool, pools
        from ._data import MODev, ObjectManager, Pool, devs, pools
        from ._data import MODev, ObjectManager, Pool, devs, pools
        from ._data import Manager
    Check whether specified blockdevs are already in the other tier.

    :param managed_objects: the result of a GetManagedObjects call
    :type managed_objects: dict of str * dict
    :param to_be_added: the blockdevs to be added
    :type to_be_added: frozenset of str
    :param other_tier: the other tier, not the one requested
    :type other_tier: _stratisd_constants.BlockDevTiers
    :raises StratisCliInUseOtherTierError: if blockdevs are used by other tier
    """
    pools_to_blockdevs = _generate_pools_to_blockdevs(
        managed_objects, to_be_added, other_tier
    )

    if pools_to_blockdevs != {}:
        raise StratisCliInUseOtherTierError(
            pools_to_blockdevs,
            BlockDevTiers.DATA
            if other_tier == BlockDevTiers.CACHE
            else BlockDevTiers.CACHE,
        )


def _check_same_tier(pool_name, managed_objects, to_be_added, this_tier):
    """
    Check whether specified blockdevs are already in the tier to which they
    are to be added.

    :param managed_objects: the result of a GetManagedObjects call
    :type managed_objects: dict of str * dict
    :param to_be_added: the blockdevs to be added
    :type to_be_added: frozenset of str
    :param this_tier: the tier requested
    :type this_tier: _stratisd_constants.BlockDevTiers
    :raises StratisCliPartialChangeError: if blockdevs are used by this tier
    :raises StratisCliInUseSameTierError: if blockdevs are used by this tier in another pool
    """
    pools_to_blockdevs = _generate_pools_to_blockdevs(
        managed_objects, to_be_added, this_tier
    )

    owned_by_current_pool = frozenset(pools_to_blockdevs.get(pool_name, []))
    owned_by_other_pools = dict(
        (pool, devnodes)
        for pool, devnodes in pools_to_blockdevs.items()
        if pool_name != pool
    )

    if owned_by_current_pool != frozenset():
        raise StratisCliPartialChangeError(
            "add to cache" if this_tier == BlockDevTiers.CACHE else "add to data",
            to_be_added.difference(owned_by_current_pool),
            to_be_added.intersection(owned_by_current_pool),
        )
    if owned_by_other_pools != {}:
        raise StratisCliInUseSameTierError(owned_by_other_pools, this_tier)


def _fetch_locked_pools_property(proxy):
    """
    Fetch the LockedPools property from stratisd.
    :param proxy: proxy to the top object in stratisd
    :return: a representation of unlocked devices
    :rtype: dict
    :raises StratisCliEngineError:
    """

    # pylint: disable=import-outside-toplevel

    return Manager.Properties.LockedPools.Get(proxy)


class PoolActions:
    """
    Pool actions.
    """

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

        :raises StratisCliEngineError:
        :raises StratisCliIncoherenceError:
        :raises StratisCliNameConflictError:
        """
        # pylint: disable=import-outside-toplevel

        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)