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

from __future__ import absolute_import

import collections

import contextlib

import errno

import os

import stat

from .i18n import _

from .pycompat import delattr

from hgdemandimport import tracing

from . import (

    dirstatemap,

    encoding,

    error,

    match as matchmod,

    pathutil,

    policy,

    pycompat,

    scmutil,

    sparse,

    util,

)

from .interfaces import (

    dirstate as intdirstate,

    util as interfaceutil,

)

parsers = policy.importmod('parsers')

rustmod = policy.importrust('dirstate')

SUPPORTS_DIRSTATE_V2 = rustmod is not None

propertycache = util.propertycache

filecache = scmutil.filecache

_rangemask = dirstatemap.rangemask

DirstateItem = parsers.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 _getfsnow(vfs):

    '''Get "now" timestamp on filesystem'''

    tmpfd, tmpname = vfs.mkstemp()

    try:

        return os.fstat(tmpfd)[stat.ST_MTIME]

    finally:

        os.close(tmpfd)

        vfs.unlink(tmpname)

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(object):

    def __init__(

        self,

        opener,

        ui,

        root,

        validate,

        sparsematchfn,

        nodeconstants,

        use_dirstate_v2,

    ):

        """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._nodeconstants = nodeconstants

        self._opener = opener

        self._validate = validate

        self._root = root

        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)

        self._dirty = False

        self._lastnormaltime = 0

        self._ui = ui

        self._filecache = {}

        self._parentwriters = 0

        self._filename = b'dirstate'

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

        self._plchangecallbacks = {}

        self._origpl = None

        self._updatedfiles = set()

        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.

        """

        # 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 IOError as inst:

            if inst.errno != errno.ENOENT:

                raise

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

        if self._checklink and self._checkexec:

            def f(x):

                try:

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

                    if util.statislink(st):

                        return b'l'

                    if util.statisexec(st):

                        return b'x'

                except OSError:

                    pass

                return b''

            return f

        fallback = buildfallback()

        if self._checklink:

            def f(x):

                if os.path.islink(self._join(x)):

                    return b'l'

                if b'x' in fallback(x):

                    return b'x'

                return b''

            return f

        if self._checkexec:

            def f(x):

                if b'l' in fallback(x):

                    return b'l'

                if util.isexec(self._join(x)):

                    return b'x'

                return b''

            return f

        else:

            return fallback

    @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 __getitem__(self, key):

        """Return the current state of key (a filename) in the dirstate.

        States are:

          n  normal

          m  needs merging

          r  marked for removal

          a  marked for addition

          ?  not tracked

        XXX The "state" is a bit obscure to be in the "public" API. we should

        consider migrating all user of this to going through the dirstate entry

        instead.

        """

        entry = self._map.get(key)

        if entry is not None:

            return entry.state

        return b'?'

    def __contains__(self, key):

        return key in self._map

    def __iter__(self):

        return iter(sorted(self._map))

    def items(self):

        return pycompat.iteritems(self._map)

    iteritems = items

    def directories(self):

        return self._map.directories()

    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

        self._map.setparents(p1, p2)

        copies = {}

        if (

            oldp2 != self._nodeconstants.nullid

            and p2 == self._nodeconstants.nullid

        ):

            candidatefiles = self._map.non_normal_or_other_parent_paths()

            for f in candidatefiles:

                s = self._map.get(f)

                if s is None:

                    continue

                # Discard "merged" markers when moving away from a merge state

                if s.merged:

                    source = self._map.copymap.get(f)

                    if source:

                        copies[f] = source

                    self.normallookup(f)

                # Also fix up otherparent markers

                elif s.from_p2:

                    source = self._map.copymap.get(f)

                    if source:

                        copies[f] = source

                    self._add(f)

        return copies

    def setbranch(self, branch):

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

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

        try:

            f.write(self._branch + b'\n')

            f.close()

            # make sure filecache has the correct stat info for _branch after

            # replacing the underlying file

            ce = self._filecache[b'_branch']

            if ce:

                ce.refresh()

        except:  # re-raises

            f.discard()

            raise

    def invalidate(self):

        """Causes the next access to reread the dirstate.

        This is different from localrepo.invalidatedirstate() because it always

        rereads the dirstate. Use localrepo.invalidatedirstate() if you want to

        check whether the dirstate has changed before rereading it."""

        for a in ("_map", "_branch", "_ignore"):

            if a in self.__dict__:

                delattr(self, a)

        self._lastnormaltime = 0

        self._dirty = False

        self._updatedfiles.clear()

        self._parentwriters = 0

        self._origpl = None

    def copy(self, source, dest):

        """Mark dest as a copy of source. Unmark dest if source is None."""

        if source == dest:

            return

        self._dirty = True

        if source is not None:

            self._map.copymap[dest] = source

            self._updatedfiles.add(source)

            self._updatedfiles.add(dest)

        elif self._map.copymap.pop(dest, None):

            self._updatedfiles.add(dest)

    def copied(self, file):

        return self._map.copymap.get(file, None)

    def copies(self):

        return self._map.copymap

    @requires_no_parents_change

    def set_tracked(self, filename):

        """a "public" method for generic code to mark a file as tracked

        This function is to be called outside of "update/merge" case. For

        example by a command like `hg add X`.

        return True the file was previously untracked, False otherwise.

        """

        entry = self._map.get(filename)

        if entry is None:

            self._add(filename)

            return True

        elif not entry.tracked: