# scmutil.py - Mercurial core utility functions

#

#  Copyright Olivia Mackall <olivia@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.

from __future__ import annotations

import binascii

import errno

import glob

import os

import posixpath

import re

import subprocess

import typing

import weakref

from typing import (

    Callable,

    Dict,

    Iterable,

    Iterator,

    List,

    Optional,

    Set,

    Tuple,

)

from .i18n import _

from .node import (

    bin,

    hex,

    nullrev,

    short,

    wdirrev,

)

from .thirdparty import attr

# Force pytype to use the non-vendored package

if typing.TYPE_CHECKING:

    # noinspection PyPackageRequirements

    import attr

from . import (

    copies as copiesmod,

    encoding,

    error,

    match as matchmod,

    obsolete,

    obsutil,

    pathutil,

    phases,

    policy,

    pycompat,

    requirements as requirementsmod,

    revsetlang,

    similar,

    smartset,

    typelib,

    url,

    util,

    vfs,

)

from .interfaces import status as istatus

from .utils import (

    hashutil,

    procutil,

    stringutil,

)

if pycompat.iswindows:

    from . import scmwindows as scmplatform

else:

    from . import scmposix as scmplatform

if typing.TYPE_CHECKING:

    from . import (

        ui as uimod,

    )

parsers = policy.importmod('parsers')

rustrevlog = policy.importrust('revlog')

termsize = scmplatform.termsize

@attr.s(slots=True, repr=False)

class status(istatus.Status):

    """Struct with a list of files per status.

    The 'deleted', 'unknown' and 'ignored' properties are only

    relevant to the working copy.

    """

    modified = attr.ib(default=attr.Factory(list), type=List[bytes])

    added = attr.ib(default=attr.Factory(list), type=List[bytes])

    removed = attr.ib(default=attr.Factory(list), type=List[bytes])

    deleted = attr.ib(default=attr.Factory(list), type=List[bytes])

    unknown = attr.ib(default=attr.Factory(list), type=List[bytes])

    ignored = attr.ib(default=attr.Factory(list), type=List[bytes])

    clean = attr.ib(default=attr.Factory(list), type=List[bytes])

    def __iter__(self) -> Iterator[List[bytes]]:

        yield self.modified

        yield self.added

        yield self.removed

        yield self.deleted

        yield self.unknown

        yield self.ignored

        yield self.clean

    def __repr__(self) -> str:

        return (

            r'<status modified=%s, added=%s, removed=%s, deleted=%s, '

            r'unknown=%s, ignored=%s, clean=%s>'

        ) % tuple(pycompat.sysstr(stringutil.pprint(v)) for v in self)

def itersubrepos(ctx1, ctx2):

    """find subrepos in ctx1 or ctx2"""

    # Create a (subpath, ctx) mapping where we prefer subpaths from

    # ctx1. The subpaths from ctx2 are important when the .hgsub file

    # has been modified (in ctx2) but not yet committed (in ctx1).

    subpaths = dict.fromkeys(ctx2.substate, ctx2)

    subpaths.update(dict.fromkeys(ctx1.substate, ctx1))

    missing = set()

    for subpath in ctx2.substate:

        if subpath not in ctx1.substate:

            del subpaths[subpath]

            missing.add(subpath)

    for subpath, ctx in sorted(subpaths.items()):

        yield subpath, ctx.sub(subpath)

    # Yield an empty subrepo based on ctx1 for anything only in ctx2.  That way,

    # status and diff will have an accurate result when it does

    # 'sub.{status|diff}(rev2)'.  Otherwise, the ctx2 subrepo is compared

    # against itself.

    for subpath in missing:

        yield subpath, ctx2.nullsub(subpath, ctx1)

def nochangesfound(ui: "uimod.ui", repo, excluded=None) -> None:

    """Report no changes for push/pull, excluded is None or a list of

    nodes excluded from the push/pull.

    """

    secretlist = []

    if excluded:

        for n in excluded:

            ctx = repo[n]

            if ctx.phase() >= phases.secret and not ctx.extinct():

                secretlist.append(n)

    if secretlist:

        ui.status(

            _(b"no changes found (ignored %d secret changesets)\n")

            % len(secretlist)

        )

    else:

        ui.status(_(b"no changes found\n"))

def callcatch(ui: "uimod.ui", func: Callable[[], int]) -> int:

    """call func() with global exception handling

    return func() if no exception happens. otherwise do some error handling

    and return an exit code accordingly. does not handle all exceptions.

    """

    coarse_exit_code = -1

    detailed_exit_code = -1

    try:

        try:

            return func()

        except:  # re-raises

            ui.traceback()

            raise

    # Global exception handling, alphabetically

    # Mercurial-specific first, followed by built-in and library exceptions

    except error.LockHeld as inst:

        detailed_exit_code = 20

        if inst.errno == errno.ETIMEDOUT:

            reason = _(b'timed out waiting for lock held by %r') % (

                pycompat.bytestr(inst.locker)

            )

        else:

            reason = _(b'lock held by %r') % inst.locker

        ui.error(

            _(b"abort: %s: %s\n")

            % (inst.desc or stringutil.forcebytestr(inst.filename), reason)

        )

        if not inst.locker:

            ui.error(_(b"(lock might be very busy)\n"))

    except error.LockUnavailable as inst:

        detailed_exit_code = 20

        ui.error(

            _(b"abort: could not lock %s: %s\n")

            % (

                inst.desc or stringutil.forcebytestr(inst.filename),

                encoding.strtolocal(inst.strerror),

            )

        )

    except error.RepoError as inst:

        if isinstance(inst, error.RepoLookupError):

            detailed_exit_code = 10

        ui.error(_(b"abort: %s\n") % inst)

        if inst.hint:

            ui.error(_(b"(%s)\n") % inst.hint)

    except error.ResponseError as inst:

        ui.error(_(b"abort: %s") % inst.args[0])

        msg = inst.args[1]

        if isinstance(msg, type(u'')):

            msg = pycompat.sysbytes(msg)

        if msg is None:

            ui.error(b"\n")

        elif not isinstance(msg, bytes):

            ui.error(b" %r\n" % (msg,))

        elif not msg:

            ui.error(_(b" empty string\n"))

        else:

            ui.error(b"\n%r\n" % pycompat.bytestr(stringutil.ellipsis(msg)))

    except error.CensoredNodeError as inst:

        ui.error(_(b"abort: file censored %s\n") % inst)

    except error.WdirUnsupported:

        ui.error(_(b"abort: working directory revision cannot be specified\n"))

    except error.Error as inst:

        if inst.detailed_exit_code is not None:

            detailed_exit_code = inst.detailed_exit_code

        if inst.coarse_exit_code is not None:

            coarse_exit_code = inst.coarse_exit_code

        ui.error(inst.format())

    except error.WorkerError as inst:

        # Don't print a message -- the worker already should have

        return inst.status_code

    except ImportError as inst:

        ui.error(_(b"abort: %s\n") % stringutil.forcebytestr(inst))

        m = stringutil.forcebytestr(inst).split()[-1]

        if m in b"mpatch bdiff".split():

            ui.error(_(b"(did you forget to compile extensions?)\n"))

        elif m in b"zlib".split():

            ui.error(_(b"(is your Python install correct?)\n"))

    except util.urlerr.httperror as inst:

        detailed_exit_code = 100

        ui.error(_(b"abort: %s\n") % stringutil.forcebytestr(inst))

    except util.urlerr.urlerror as inst:

        detailed_exit_code = 100

        try:  # usually it is in the form (errno, strerror)

            reason = inst.reason.args[1]

        except (AttributeError, IndexError):

            # it might be anything, for example a string

            reason = inst.reason

        if isinstance(reason, str):

            # SSLError of Python 2.7.9 contains a unicode

            reason = encoding.unitolocal(reason)

        ui.error(_(b"abort: error: %s\n") % stringutil.forcebytestr(reason))

    except OSError as inst:

        if hasattr(inst, "args") and inst.args and inst.args[0] == errno.EPIPE:

            pass

        elif getattr(inst, "strerror", None):  # common IOError or OSError

            if getattr(inst, "filename", None) is not None:

                ui.error(

                    _(b"abort: %s: '%s'\n")

                    % (

                        encoding.strtolocal(inst.strerror),

                        stringutil.forcebytestr(inst.filename),

                    )

                )

            else:

                ui.error(_(b"abort: %s\n") % encoding.strtolocal(inst.strerror))

        else:  # suspicious IOError

            raise

    except MemoryError:

        ui.error(_(b"abort: out of memory\n"))

    except SystemExit as inst:

        # Commands shouldn't sys.exit directly, but give a return code.

        # Just in case catch this and and pass exit code to caller.

        detailed_exit_code = 254

        coarse_exit_code = inst.code

    if ui.configbool(b'ui', b'detailed-exit-code'):

        return detailed_exit_code

    else:

        return coarse_exit_code

def checknewlabel(repo, lbl: bytes, kind) -> None:

    # Do not use the "kind" parameter in ui output.

    # It makes strings difficult to translate.

    if lbl in [b'tip', b'.', b'null']:

        raise error.InputError(_(b"the name '%s' is reserved") % lbl)

    for c in (b':', b'\0', b'\n', b'\r'):

        if c in lbl:

            raise error.InputError(

                _(b"%r cannot be used in a name") % pycompat.bytestr(c)

            )

    try:

        int(lbl)

        if b'_' in lbl:

            # If label contains underscores, Python might consider it an

            # integer (with "_" as visual separators), but we do not.

            # See PEP 515 - Underscores in Numeric Literals.

            raise ValueError

        raise error.InputError(_(b"cannot use an integer as a name"))

    except ValueError:

        pass

    if lbl.strip() != lbl:

        raise error.InputError(

            _(b"leading or trailing whitespace in name %r") % lbl

        )

def checkfilename(f: bytes) -> None:

    '''Check that the filename f is an acceptable filename for a tracked file'''

    if b'\r' in f or b'\n' in f:

        raise error.InputError(

            _(b"'\\n' and '\\r' disallowed in filenames: %r")

            % pycompat.bytestr(f)

        )

def checkportable(ui: "uimod.ui", f: bytes) -> None:

    '''Check if filename f is portable and warn or abort depending on config'''

    checkfilename(f)

    abort, warn = checkportabilityalert(ui)

    if abort or warn:

        msg = util.checkwinfilename(f)

        if msg:

            msg = b"%s: %s" % (msg, procutil.shellquote(f))

            if abort:

                raise error.InputError(msg)

            ui.warn(_(b"warning: %s\n") % msg)

def checkportabilityalert(ui: "uimod.ui") -> Tuple[bool, bool]:

    """check if the user's config requests nothing, a warning, or abort for

    non-portable filenames"""

    val = ui.config(b'ui', b'portablefilenames')

    lval = val.lower()

    bval = stringutil.parsebool(val)

    abort = pycompat.iswindows or lval == b'abort'

    warn = bval or lval == b'warn'

    if bval is None and not (warn or abort or lval == b'ignore'):

        raise error.ConfigError(

            _(b"ui.portablefilenames value is invalid ('%s')") % val

        )

    return abort, warn

class casecollisionauditor:

    def __init__(self, ui: "uimod.ui", abort: bool, dirstate) -> None:

        self._ui = ui

        self._abort = abort

        allfiles = b'\0'.join(dirstate)

        self._loweredfiles = set(encoding.lower(allfiles).split(b'\0'))

        self._dirstate = dirstate

        # The purpose of _newfiles is so that we don't complain about

        # case collisions if someone were to call this object with the

        # same filename twice.

        self._newfiles = set()

    def __call__(self, f: bytes) -> None:

        if f in self._newfiles:

            return

        fl = encoding.lower(f)

        if fl in self._loweredfiles and f not in self._dirstate:

            msg = _(b'possible case-folding collision for %s') % f

            if self._abort:

                raise error.StateError(msg)

            self._ui.warn(_(b"warning: %s\n") % msg)

        self._loweredfiles.add(fl)

        self._newfiles.add(f)

def combined_filtered_and_obsolete_hash(

    repo, maxrev, needobsolete: bool = False

):

    """build hash of filtered revisions in the current repoview.

    Multiple caches perform up-to-date validation by checking that the

    tiprev and tipnode stored in the cache file match the current repository.

    However, this is not sufficient for validating repoviews because the set

    of revisions in the view may change without the repository tiprev and

    tipnode changing.

    This function hashes all the revs filtered from the view (and, optionally,

    all obsolete revs) up to maxrev and returns that SHA-1 digest.

    """

    cl = repo.changelog

    if needobsolete:

        obsrevs = obsolete.getrevs(repo, b'obsolete')

        if not cl.filteredrevs and not obsrevs:

            return None

        key = (maxrev, hash(cl.filteredrevs), hash(obsrevs))

    else:

        if not cl.filteredrevs:

            return None

        key = maxrev

        obsrevs = frozenset()

    result = cl._filteredrevs_hashcache.get(key)

    if not result:

        revs, obs_revs = _filtered_and_obs_revs(repo, maxrev)

        if needobsolete:

            revs = revs | obs_revs

        revs = sorted(revs)

        if revs:

            result = _hash_revs(revs)

            cl._filteredrevs_hashcache[key] = result

    return result

def filtered_and_obsolete_hash(repo, maxrev):

    """build hashs of filtered and obsolete revisions in the current repoview.

    Multiple caches perform up-to-date validation by checking that the

    tiprev and tipnode stored in the cache file match the current repository.

    However, this is not sufficient for validating repoviews because the set

    of revisions in the view may change without the repository tiprev and

    tipnode changing.

    This function hashes all the revs filtered from the view up to maxrev and

    returns that SHA-1 digest. The obsolete revisions hashed are only the

    non-filtered one.

    """

    cl = repo.changelog

    obs_set = obsolete.getrevs(repo, b'obsolete')

    key = (maxrev, hash(cl.filteredrevs), hash(obs_set))

    result = cl._filteredrevs_hashcache.get(key)

    if result is None:

        filtered_hash = None

        obs_hash = None

        filtered_revs, obs_revs = _filtered_and_obs_revs(repo, maxrev)

        if filtered_revs:

            filtered_hash = _hash_revs(filtered_revs)

        if obs_revs:

            obs_hash = _hash_revs(obs_revs)

        result = (filtered_hash, obs_hash)

        cl._filteredrevs_hashcache[key] = result

    return result

def _filtered_and_obs_revs(repo, max_rev):

    """return the set of filtered and non-filtered obsolete revision"""

    cl = repo.changelog

    obs_set = obsolete.getrevs(repo, b'obsolete')

    filtered_set = cl.filteredrevs

    if cl.filteredrevs:

        obs_set = obs_set - cl.filteredrevs

    if max_rev < (len(cl) - 1):

        # there might be revision to filter out

        filtered_set = set(r for r in filtered_set if r <= max_rev)

        obs_set = set(r for r in obs_set if r <= max_rev)

    return (filtered_set, obs_set)

def _hash_revs(revs: Iterable[int]) -> bytes:

    """return a hash from a list of revision numbers"""

    s = hashutil.sha1()

    for rev in revs:

        s.update(b'%d;' % rev)

    return s.digest()

def walkrepos(

    path,

    followsym: bool = False,

    seen_dirs: Optional[List[bytes]] = None,

    recurse: bool = False,

) -> Iterable[bytes]:

    """yield every hg repository under path, always recursively.

    The recurse flag will only control recursion into repo working dirs"""

    def errhandler(err):

        if err.filename == path: