# 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_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_changing_files(func):

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

        if not self.is_changing_files:

            msg = 'calling `%s` outside of a `changing_files`'

            msg %= func.__name__

            raise error.ProgrammingError(msg)

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

    return wrap

def requires_changing_any(func):

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

        if not self.is_changing_any:

            msg = 'calling `%s` outside of a changing 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"

CHANGE_TYPE_FILES = "files"

@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)

        has_tr = repo.currenttransaction() is not None

        # 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:  # re-raises

            self.invalidate()

            raise

        finally:

            tr = repo.currenttransaction()

            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.

                        self.write(tr)

            if has_tr != (tr is not None):

                if has_tr:

                    m = "transaction vanished while changing dirstate"

                else:

                    m = "transaction appeared while changing dirstate"

                raise error.ProgrammingError(m)

    @contextlib.contextmanager

    def changing_parents(self, repo):

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

            yield c

    @contextlib.contextmanager

    def changing_files(self, repo):

        with self._changing(repo, CHANGE_TYPE_FILES) 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

    @property

    def is_changing_files(self):

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

        that modify the files tracked or their sources.

        """

        if self._changing_level <= 0:

            return False

        return self._change_type == CHANGE_TYPE_FILES

    @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: