# error.py - Mercurial exceptions

#

# Copyright 2005-2008 Matt Mackall <mpm@selenic.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

"""Mercurial exceptions.

This allows us to catch exceptions at higher levels without forcing

imports.

"""

from __future__ import absolute_import

import difflib

# Do not import anything but pycompat here, please

from . import pycompat

if pycompat.TYPE_CHECKING:

    from typing import (

        Optional,

    )

def _tobytes(exc):

    """Byte-stringify exception in the same way as BaseException_str()"""

    if not exc.args:

        return b''

    if len(exc.args) == 1:

        return pycompat.bytestr(exc.args[0])

    return b'(%s)' % b', '.join(b"'%s'" % pycompat.bytestr(a) for a in exc.args)

class Hint(object):

    """Mix-in to provide a hint of an error

    This should come first in the inheritance list to consume a hint and

    pass remaining arguments to the exception class.

    """

    def __init__(self, *args, **kw):

        self.hint = kw.pop('hint', None)

        super(Hint, self).__init__(*args, **kw)

class StorageError(Hint, Exception):

    """Raised when an error occurs in a storage layer.

    Usually subclassed by a storage-specific exception.

    """

    __bytes__ = _tobytes

class RevlogError(StorageError):

    pass

class SidedataHashError(RevlogError):

    def __init__(self, key, expected, got):

        self.hint = None

        self.sidedatakey = key

        self.expecteddigest = expected

        self.actualdigest = got

class FilteredIndexError(IndexError):

    __bytes__ = _tobytes

class LookupError(RevlogError, KeyError):

    def __init__(self, name, index, message):

        self.name = name

        self.index = index

        # this can't be called 'message' because at least some installs of

        # Python 2.6+ complain about the 'message' property being deprecated

        self.lookupmessage = message

        if isinstance(name, bytes) and len(name) == 20:

            from .node import hex

            name = hex(name)

        # if name is a binary node, it can be None

        RevlogError.__init__(

            self, b'%s@%s: %s' % (index, pycompat.bytestr(name), message)

        )

    def __bytes__(self):

        return RevlogError.__bytes__(self)

    def __str__(self):

        return RevlogError.__str__(self)

class AmbiguousPrefixLookupError(LookupError):

    pass

class FilteredLookupError(LookupError):

    pass

class ManifestLookupError(LookupError):

    pass

class CommandError(Exception):

    """Exception raised on errors in parsing the command line."""

    def __init__(self, command, message):

        self.command = command

        self.message = message

        super(CommandError, self).__init__()

    __bytes__ = _tobytes

class UnknownCommand(Exception):

    """Exception raised if command is not in the command table."""

    def __init__(self, command, all_commands=None):

        self.command = command

        self.all_commands = all_commands

        super(UnknownCommand, self).__init__()

    __bytes__ = _tobytes

class AmbiguousCommand(Exception):

    """Exception raised if command shortcut matches more than one command."""

    def __init__(self, prefix, matches):

        self.prefix = prefix

        self.matches = matches

        super(AmbiguousCommand, self).__init__()

    __bytes__ = _tobytes

class WorkerError(Exception):

    """Exception raised when a worker process dies."""

    def __init__(self, status_code):

        self.status_code = status_code

        # Pass status code to superclass just so it becomes part of __bytes__

        super(WorkerError, self).__init__(status_code)

    __bytes__ = _tobytes

class InterventionRequired(Hint, Exception):

    """Exception raised when a command requires human intervention."""

    __bytes__ = _tobytes

class ConflictResolutionRequired(InterventionRequired):

    """Exception raised when a continuable command required merge conflict resolution."""

    def __init__(self, opname):

        from .i18n import _

        self.opname = opname

        InterventionRequired.__init__(

            self,

            _(

                b"unresolved conflicts (see 'hg resolve', then 'hg %s --continue')"

            )

            % opname,

        )

class Abort(Hint, Exception):

    """Raised if a command needs to print an error and exit."""

    def __init__(self, message, hint=None):

        # type: (bytes, Optional[bytes]) -> None

        self.message = message

        self.hint = hint

        # Pass the message into the Exception constructor to help extensions

        # that look for exc.args[0].

        Exception.__init__(self, message)

    def __bytes__(self):

        return self.message

    if pycompat.ispy3:

        def __str__(self):

            # the output would be unreadable if the message was translated,

            # but do not replace it with encoding.strfromlocal(), which

            # may raise another exception.

            return pycompat.sysstr(self.__bytes__())

    def format(self):

        from .i18n import _

        message = _(b"abort: %s\n") % self.message

        if self.hint:

            message += _(b"(%s)\n") % self.hint

        return message

class InputError(Abort):

    """Indicates that the user made an error in their input.

    Examples: Invalid command, invalid flags, invalid revision.

    """

class StateError(Abort):

    """Indicates that the operation might work if retried in a different state.

    Examples: Unresolved merge conflicts, unfinished operations.

    """

class CanceledError(Abort):

    """Indicates that the user canceled the operation.

    Examples: Close commit editor with error status, quit chistedit.

    """

class SecurityError(Abort):

    """Indicates that some aspect of security failed.

    Examples: Bad server credentials, expired local credentials for network

    filesystem, mismatched GPG signature, DoS protection.

    """

class HookLoadError(Abort):

    """raised when loading a hook fails, aborting an operation

    Exists to allow more specialized catching."""

class HookAbort(Abort):

    """raised when a validation hook fails, aborting an operation

    Exists to allow more specialized catching."""

class ConfigError(Abort):

    """Exception raised when parsing config files"""

    def __init__(self, message, location=None, hint=None):

        super(ConfigError, self).__init__(message, hint=hint)

        self.location = location

    def format(self):

        from .i18n import _

        if self.location is not None:

            message = _(b"config error at %s: %s\n") % (

                pycompat.bytestr(self.location),

                self.message,

            )

        else:

            message = _(b"config error: %s\n") % self.message

        if self.hint:

            message += _(b"(%s)\n") % self.hint

        return message

class UpdateAbort(Abort):

    """Raised when an update is aborted for destination issue"""

class MergeDestAbort(Abort):

    """Raised when an update is aborted for destination issues"""

class NoMergeDestAbort(MergeDestAbort):

    """Raised when an update is aborted because there is nothing to merge"""

class ManyMergeDestAbort(MergeDestAbort):

    """Raised when an update is aborted because destination is ambiguous"""

class ResponseExpected(Abort):

    """Raised when an EOF is received for a prompt"""

    def __init__(self):

        from .i18n import _

        Abort.__init__(self, _(b'response expected'))

class OutOfBandError(Hint, Exception):

    """Exception raised when a remote repo reports failure"""

    __bytes__ = _tobytes

class ParseError(Abort):

    """Raised when parsing config files and {rev,file}sets (msg[, pos])"""

    def __init__(self, message, location=None, hint=None):

        super(ParseError, self).__init__(message, hint=hint)

        self.location = location

    def format(self):

        from .i18n import _

        if self.location is not None:

            message = _(b"hg: parse error at %s: %s\n") % (

                pycompat.bytestr(self.location),

                self.message,

            )

        else:

            message = _(b"hg: parse error: %s\n") % self.message

        if self.hint:

            message += _(b"(%s)\n") % self.hint

        return message

class PatchError(Exception):

    __bytes__ = _tobytes

def getsimilar(symbols, value):

    sim = lambda x: difflib.SequenceMatcher(None, value, x).ratio()

    # The cutoff for similarity here is pretty arbitrary. It should

    # probably be investigated and tweaked.

    return [s for s in symbols if sim(s) > 0.6]

def similarity_hint(similar):

    from .i18n import _

    if len(similar) == 1:

        return _(b"did you mean %s?") % similar[0]

    elif similar:

        ss = b", ".join(sorted(similar))

        return _(b"did you mean one of %s?") % ss

    else:

        return None

class UnknownIdentifier(ParseError):

    """Exception raised when a {rev,file}set references an unknown identifier"""

    def __init__(self, function, symbols):

        from .i18n import _

        similar = getsimilar(symbols, function)

        hint = similarity_hint(similar)

        ParseError.__init__(

            self, _(b"unknown identifier: %s") % function, hint=hint

        )

class RepoError(Hint, Exception):

    __bytes__ = _tobytes

class RepoLookupError(RepoError):

    pass

class FilteredRepoLookupError(RepoLookupError):

    pass

class CapabilityError(RepoError):

    pass

class RequirementError(RepoError):

    """Exception raised if .hg/requires has an unknown entry."""

class StdioError(IOError):

    """Raised if I/O to stdout or stderr fails"""

    def __init__(self, err):

        IOError.__init__(self, err.errno, err.strerror)

    # no __bytes__() because error message is derived from the standard IOError

class UnsupportedMergeRecords(Abort):

    def __init__(self, recordtypes):

        from .i18n import _

        self.recordtypes = sorted(recordtypes)

        s = b' '.join(self.recordtypes)

        Abort.__init__(

            self,

            _(b'unsupported merge state records: %s') % s,

            hint=_(

                b'see https://mercurial-scm.org/wiki/MergeStateRecords for '

                b'more information'

            ),

        )

class UnknownVersion(Abort):

    """generic exception for aborting from an encounter with an unknown version"""