# 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 (

    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_parents_change(func):

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

        if not self.pendingparentchange():

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

            msg %= func.__name__

            raise error.ProgrammingError(msg)

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

    return wrap

def requires_no_parents_change(func):

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

        if self.pendingparentchange():

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

            msg %= func.__name__

            raise error.ProgrammingError(msg)

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

    return wrap

@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 = {}

        self._parentwriters = 0

        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 parentchange(self):

        """Context manager for handling dirstate parents.

        If an exception occurs in the scope of the context manager,

        the incoherent dirstate won't be written when wlock is

        released.

        """

        self._parentwriters += 1

        yield

        # Typically we want the "undo" step of a context manager in a

        # finally block so it happens even when an exception

        # occurs. In this case, however, we only want to decrement

        # parentwriters if the code in the with statement exits

        # normally, so we don't have a try/finally here on purpose.

        self._parentwriters -= 1

    def pendingparentchange(self):

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

        that modify the dirstate parent.

        """

        return self._parentwriters > 0

    @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._parentwriters == 0:

            raise ValueError(

                b"cannot set dirstate parent outside of "

                b"dirstate.parentchange 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)