# dirstate.py - working directory tracking for mercurial

#

# Copyright 2005-2007 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.

import collections

import contextlib

import os

import stat

import uuid

from .i18n import _

from .pycompat import delattr

from hgdemandimport import tracing

from . import (

    dirstatemap,

    encoding,

    error,

    match as matchmod,

    node,

    pathutil,

    policy,

    pycompat,

    scmutil,

    util,

)

from .dirstateutils import (

    docket as docketmod,

    timestamp,

)

from .interfaces import (

    dirstate as intdirstate,

    util as interfaceutil,

)

parsers = policy.importmod('parsers')

rustmod = policy.importrust('dirstate')

HAS_FAST_DIRSTATE_V2 = rustmod is not None

propertycache = util.propertycache

filecache = scmutil.filecache

_rangemask = dirstatemap.rangemask

DirstateItem = dirstatemap.DirstateItem

class repocache(filecache):

    """filecache for files in .hg/"""

    def join(self, obj, fname):

        return obj._opener.join(fname)

class rootcache(filecache):

    """filecache for files in the repository root"""

    def join(self, obj, fname):

        return obj._join(fname)

def requires_changing_parents(func):

    def wrap(self, *args, **kwargs):

        if not self.is_changing_parents:

            msg = 'calling `%s` outside of a changing_parents context'

            msg %= func.__name__

            raise error.ProgrammingError(msg)

        if self._invalidated_context:

            msg = 'calling `%s` after the dirstate was invalidated'

            raise error.ProgrammingError(msg)

        return func(self, *args, **kwargs)

    return wrap

def requires_not_changing_parents(func):

    def wrap(self, *args, **kwargs):

        if self.is_changing_parents:

            msg = 'calling `%s` inside of a changing_parents context'

            msg %= func.__name__

            raise error.ProgrammingError(msg)

        return func(self, *args, **kwargs)

    return wrap

CHANGE_TYPE_PARENTS = "parents"

@interfaceutil.implementer(intdirstate.idirstate)

class dirstate:

    def __init__(

        self,

        opener,

        ui,

        root,

        validate,

        sparsematchfn,

        nodeconstants,

        use_dirstate_v2,

        use_tracked_hint=False,

    ):

        """Create a new dirstate object.

        opener is an open()-like callable that can be used to open the

        dirstate file; root is the root of the directory tracked by

        the dirstate.

        """

        self._use_dirstate_v2 = use_dirstate_v2

        self._use_tracked_hint = use_tracked_hint

        self._nodeconstants = nodeconstants

        self._opener = opener

        self._validate = validate

        self._root = root

        # Either build a sparse-matcher or None if sparse is disabled

        self._sparsematchfn = sparsematchfn

        # ntpath.join(root, '') of Python 2.7.9 does not add sep if root is

        # UNC path pointing to root share (issue4557)

        self._rootdir = pathutil.normasprefix(root)

        # True is any internal state may be different

        self._dirty = False

        # True if the set of tracked file may be different

        self._dirty_tracked_set = False

        self._ui = ui

        self._filecache = {}

        # nesting level of `changing_parents` context

        self._changing_level = 0

        # the change currently underway

        self._change_type = None

        # True if the current dirstate changing operations have been

        # invalidated (used to make sure all nested contexts have been exited)

        self._invalidated_context = False

        self._filename = b'dirstate'

        self._filename_th = b'dirstate-tracked-hint'

        self._pendingfilename = b'%s.pending' % self._filename

        self._plchangecallbacks = {}

        self._origpl = None

        self._mapcls = dirstatemap.dirstatemap

        # Access and cache cwd early, so we don't access it for the first time

        # after a working-copy update caused it to not exist (accessing it then

        # raises an exception).

        self._cwd

    def prefetch_parents(self):

        """make sure the parents are loaded

        Used to avoid a race condition.

        """

        self._pl

    @contextlib.contextmanager

    def _changing(self, repo, change_type):

        if repo.currentwlock() is None:

            msg = b"trying to change the dirstate without holding the wlock"

            raise error.ProgrammingError(msg)

        if self._invalidated_context:

            msg = "trying to use an invalidated dirstate before it has reset"

            raise error.ProgrammingError(msg)

        # different type of change are mutually exclusive

        if self._change_type is None:

            assert self._changing_level == 0

            self._change_type = change_type

        elif self._change_type != change_type:

            msg = (

                'trying to open "%s" dirstate-changing context while a "%s" is'

                ' already open'

            )

            msg %= (change_type, self._change_type)

            raise error.ProgrammingError(msg)

        self._changing_level += 1

        try:

            yield

        except Exception:

            self.invalidate()

            raise

        finally:

            if self._changing_level > 0:

                if self._invalidated_context:

                    # make sure we invalidate anything an upper context might

                    # have changed.

                    self.invalidate()

                self._changing_level -= 1

                # The invalidation is complete once we exit the final context

                # manager

                if self._changing_level <= 0:

                    self._change_type = None

                    assert self._changing_level == 0

                    if self._invalidated_context:

                        self._invalidated_context = False

                    else:

                        # When an exception occured, `_invalidated_context`

                        # would have been set to True by the `invalidate`

                        # call earlier.

                        #

                        # We don't have more straightforward code, because the

                        # Exception catching (and the associated `invalidate`

                        # calling) might have been called by a nested context

                        # instead of the top level one.

    @contextlib.contextmanager

    def changing_parents(self, repo):

        with self._changing(repo, CHANGE_TYPE_PARENTS) as c:

            yield c

    # here to help migration to the new code

    def parentchange(self):

        msg = (

            "Mercurial 6.4 and later requires call to "

            "`dirstate.changing_parents(repo)`"

        )

        raise error.ProgrammingError(msg)

    @property

    def is_changing_any(self):

        """Returns true if the dirstate is in the middle of a set of changes.

        This returns True for any kind of change.

        """

        return self._changing_level > 0

    def pendingparentchange(self):

        return self.is_changing_parent()

    def is_changing_parent(self):

        """Returns true if the dirstate is in the middle of a set of changes

        that modify the dirstate parent.

        """

        self._ui.deprecwarn(b"dirstate.is_changing_parents", b"6.5")

        return self.is_changing_parents

    @property

    def is_changing_parents(self):

        """Returns true if the dirstate is in the middle of a set of changes

        that modify the dirstate parent.

        """

        if self._changing_level <= 0:

            return False

        return self._change_type == CHANGE_TYPE_PARENTS

    @propertycache

    def _map(self):

        """Return the dirstate contents (see documentation for dirstatemap)."""

        self._map = self._mapcls(

            self._ui,

            self._opener,

            self._root,

            self._nodeconstants,

            self._use_dirstate_v2,

        )

        return self._map

    @property

    def _sparsematcher(self):

        """The matcher for the sparse checkout.

        The working directory may not include every file from a manifest. The

        matcher obtained by this property will match a path if it is to be

        included in the working directory.

        When sparse if disabled, return None.

        """

        if self._sparsematchfn is None:

            return None

        # TODO there is potential to cache this property. For now, the matcher

        # is resolved on every access. (But the called function does use a

        # cache to keep the lookup fast.)

        return self._sparsematchfn()

    @repocache(b'branch')

    def _branch(self):

        try:

            return self._opener.read(b"branch").strip() or b"default"

        except FileNotFoundError:

            return b"default"

    @property

    def _pl(self):

        return self._map.parents()

    def hasdir(self, d):

        return self._map.hastrackeddir(d)

    @rootcache(b'.hgignore')

    def _ignore(self):

        files = self._ignorefiles()

        if not files:

            return matchmod.never()

        pats = [b'include:%s' % f for f in files]

        return matchmod.match(self._root, b'', [], pats, warn=self._ui.warn)

    @propertycache

    def _slash(self):

        return self._ui.configbool(b'ui', b'slash') and pycompat.ossep != b'/'

    @propertycache

    def _checklink(self):

        return util.checklink(self._root)

    @propertycache

    def _checkexec(self):

        return bool(util.checkexec(self._root))

    @propertycache

    def _checkcase(self):

        return not util.fscasesensitive(self._join(b'.hg'))

    def _join(self, f):

        # much faster than os.path.join()

        # it's safe because f is always a relative path

        return self._rootdir + f

    def flagfunc(self, buildfallback):

        """build a callable that returns flags associated with a filename

        The information is extracted from three possible layers:

        1. the file system if it supports the information

        2. the "fallback" information stored in the dirstate if any

        3. a more expensive mechanism inferring the flags from the parents.

        """

        # small hack to cache the result of buildfallback()

        fallback_func = []

        def get_flags(x):

            entry = None

            fallback_value = None

            try:

                st = os.lstat(self._join(x))

            except OSError:

                return b''

            if self._checklink:

                if util.statislink(st):

                    return b'l'

            else:

                entry = self.get_entry(x)

                if entry.has_fallback_symlink:

                    if entry.fallback_symlink:

                        return b'l'

                else:

                    if not fallback_func:

                        fallback_func.append(buildfallback())

                    fallback_value = fallback_func[0](x)

                    if b'l' in fallback_value:

                        return b'l'

            if self._checkexec:

                if util.statisexec(st):

                    return b'x'

            else:

                if entry is None:

                    entry = self.get_entry(x)

                if entry.has_fallback_exec:

                    if entry.fallback_exec:

                        return b'x'

                else:

                    if fallback_value is None:

                        if not fallback_func:

                            fallback_func.append(buildfallback())

                        fallback_value = fallback_func[0](x)

                    if b'x' in fallback_value:

                        return b'x'

            return b''

        return get_flags

    @propertycache

    def _cwd(self):

        # internal config: ui.forcecwd

        forcecwd = self._ui.config(b'ui', b'forcecwd')

        if forcecwd:

            return forcecwd

        return encoding.getcwd()

    def getcwd(self):

        """Return the path from which a canonical path is calculated.

        This path should be used to resolve file patterns or to convert

        canonical paths back to file paths for display. It shouldn't be

        used to get real file paths. Use vfs functions instead.

        """

        cwd = self._cwd

        if cwd == self._root:

            return b''

        # self._root ends with a path separator if self._root is '/' or 'C:\'

        rootsep = self._root

        if not util.endswithsep(rootsep):

            rootsep += pycompat.ossep

        if cwd.startswith(rootsep):

            return cwd[len(rootsep) :]

        else:

            # we're outside the repo. return an absolute path.

            return cwd

    def pathto(self, f, cwd=None):

        if cwd is None:

            cwd = self.getcwd()

        path = util.pathto(self._root, cwd, f)

        if self._slash:

            return util.pconvert(path)

        return path

    def get_entry(self, path):

        """return a DirstateItem for the associated path"""

        entry = self._map.get(path)

        if entry is None:

            return DirstateItem()

        return entry

    def __contains__(self, key):

        return key in self._map

    def __iter__(self):

        return iter(sorted(self._map))

    def items(self):

        return self._map.items()

    iteritems = items

    def parents(self):

        return [self._validate(p) for p in self._pl]

    def p1(self):

        return self._validate(self._pl[0])

    def p2(self):

        return self._validate(self._pl[1])

    @property

    def in_merge(self):

        """True if a merge is in progress"""

        return self._pl[1] != self._nodeconstants.nullid

    def branch(self):

        return encoding.tolocal(self._branch)

    def setparents(self, p1, p2=None):

        """Set dirstate parents to p1 and p2.

        When moving from two parents to one, "merged" entries a

        adjusted to normal and previous copy records discarded and

        returned by the call.

        See localrepo.setparents()

        """

        if p2 is None:

            p2 = self._nodeconstants.nullid

        if self._changing_level == 0:

            raise ValueError(

                b"cannot set dirstate parent outside of "

                b"dirstate.changing_parents context manager"

            )

        self._dirty = True

        oldp2 = self._pl[1]

        if self._origpl is None:

            self._origpl = self._pl

        nullid = self._nodeconstants.nullid

        # True if we need to fold p2 related state back to a linear case

        fold_p2 = oldp2 != nullid and p2 == nullid

        return self._map.setparents(p1, p2, fold_p2=fold_p2)

    def setbranch(self, branch):

        self.__class__._branch.set(self, encoding.fromlocal(branch))

        f = self._opener(b'branch', b'w', atomictemp=True, checkambig=True)