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

    """check we func is called a non-invalidated dirstate

    The dirstate is in an "invalidated state" after an error occured during its

    modification and remains so until we exited the top level scope that framed

    such change.

    """

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

        if self._invalidated_context:

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

            msg %= func.__name__

            raise error.ProgrammingError(msg)

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

    return wrap

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)

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

    return check_invalidated(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 check_invalidated(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)

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

    return check_invalidated(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 check_invalidated(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._attached_to_a_transaction = 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

    @check_invalidated

    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)

        has_tr = repo.currenttransaction() is not None

        if not has_tr and self._changing_level == 0 and self._dirty:

            msg = "entering a changing context, but dirstate is already dirty"

            raise error.ProgrammingError(msg)

        assert self._changing_level >= 0

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

        should_write = False

        self._changing_level += 1

        try:

            yield

        except:  # re-raises

            self.invalidate()  # this will set `_invalidated_context`

            raise

        finally:

            assert self._changing_level > 0

            self._changing_level -= 1

            # If the dirstate is being invalidated, call invalidate again.

            # This will throw away anything added by a upper context and

            # reset the `_invalidated_context` flag when relevant

            if self._changing_level <= 0:

                self._change_type = None

                assert self._changing_level == 0

            if self._invalidated_context:

                # make sure we invalidate anything an upper context might

                # have changed.

                self.invalidate()

            else:

                should_write = self._changing_level <= 0

        tr = repo.currenttransaction()

        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)

        if should_write:

            self.write(tr)

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