comparison hgext/git/dirstate.py @ 51929:93d872a06132

typing: add type annotations to the dirstate classes

The basic procedure here was to use `merge-pyi` to merge the `git/dirstate.pyi`
file in (after renaming the interface class to match), cleaning up the import
statement mess, and then repeating the procedure for `mercurial/dirstate.pyi`.
Surprisingly, git's dirstate had more hints inferred in its *.pyi file.

After that, it was a manual examination of each method in the interface, and how
they were implemented in the core and git classes to verify what was inferred by
pytype, and fill in the missing gaps.  Since this involved jumping around
between three different files, I applied the same type info to all three at the
same time.  Complex types I rolled up into type aliases in the interface module,
and used that as needed.  That way if it changes, there's one place to edit.

There are some hints still missing, and some documentation that doesn't match
the signatures.  They should all be marked with TODOs.  There are also a bunch
of methods on the core class that aren't on the Protocol class that seem like
maybe they should be (like `set_tracked()`).  There are even more methods
missing from the git class.  But that's a project for another time.

51928:3688a984134b

from __future__ import annotations

import contextlib
import os

from mercurial.node import sha1nodeconstants
from mercurial import (
    dirstatemap,
    error,

            sha1nodeconstants,
            self._use_dirstate_v2,
        )
        return self._map

    def p1(self):
        try:
            return self.git.head.peel().id.raw
        except pygit2.GitError:
            # Typically happens when peeling HEAD fails, as in an
            # empty repository.
            return sha1nodeconstants.nullid

    def p2(self):
        # TODO: MERGE_HEAD? something like that, right?
        return sha1nodeconstants.nullid

    def setparents(self, p1, p2=None):
        if p2 is None:
            p2 = sha1nodeconstants.nullid
        assert p2 == sha1nodeconstants.nullid, b'TODO merging support'
        self.git.head.set_target(gitutil.togitnode(p1))

    def identity(self):
        return util.filestat.frompath(
            os.path.join(self._root, b'.git', b'index')
        )

    def branch(self):
        return b'default'

    def parents(self):
        # TODO how on earth do we find p2 if a merge is in flight?
        return [self.p1(), sha1nodeconstants.nullid]

    def __iter__(self):
        return (pycompat.fsencode(f.path) for f in self.git.index)

    def items(self):
        for ie in self.git.index:
            yield ie.path, None  # value should be a DirstateItem

    # py2,3 compat forward
    iteritems = items

            gs = self.git.status_file(filename)
        except KeyError:
            return b'?'
        return _STATUS_MAP[gs]

    def __contains__(self, filename):
        try:
            gs = self.git.status_file(filename)
            return _STATUS_MAP[gs] != b'?'
        except KeyError:
            return False

    def status(self, match, subrepos, ignored, clean, unknown):
        listclean = clean
        # TODO handling of clean files - can we get that from git.status()?
        modified, added, removed, deleted, unknown, ignored, clean = (
            [],
            [],

                modified, added, removed, deleted, unknown, ignored, clean
            ),
            mtime_boundary,
        )

    def flagfunc(self, buildfallback):
        # TODO we can do better
        return buildfallback()

    def getcwd(self):
        # TODO is this a good way to do this?
        return os.path.dirname(
            os.path.dirname(pycompat.fsencode(self.git.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 normalize(self, path, isknown=False, ignoremissing=False):
        normed = util.normcase(path)
        assert normed == path, b"TODO handling of case folding: %s != %s" % (
            normed,
            path,
        )
        return path

    @property
    def _checklink(self):
        return util.checklink(os.path.dirname(pycompat.fsencode(self.git.path)))

    def copies(self):
        # TODO support copies?
        return {}

    # # TODO what the heck is this
    _filecache = set()

    @property
    def is_changing_parents(self):
        # TODO: we need to implement the context manager bits and
        # correctly stage/revert index edits.
        return False

    @property
    def is_changing_any(self):
        # TODO: we need to implement the context manager bits and
        # correctly stage/revert index edits.
        return False

    def write(self, tr):
        # TODO: call parent change callbacks

        if tr:

            def writeinner(category):

            tr.addpending(b'gitdirstate', writeinner)
        else:
            self.git.index.write()

    def pathto(self, f, cwd=None):
        if cwd is None:
            cwd = self.getcwd()
        # TODO core dirstate does something about slashes here
        assert isinstance(f, bytes)
        r = util.pathto(self._root, cwd, f)
        return r

    def matches(self, match):
        for x in self.git.index:
            p = pycompat.fsencode(x.path)
            if match(p):
                yield p

    def set_clean(self, f, parentfiledata):
        """Mark a file normal and clean."""
        # TODO: for now we just let libgit2 re-stat the file. We can
        # clearly do better.

    def set_possibly_dirty(self, f):
        """Mark a file normal, but possibly dirty."""
        # TODO: for now we just let libgit2 re-stat the file. We can
        # clearly do better.

    def walk(self, match, subrepos, unknown, ignored, full=True):
        # TODO: we need to use .status() and not iterate the index,
        # because the index doesn't force a re-walk and so `hg add` of
        # a new file without an intervening call to status will
        # silently do nothing.
        r = {}

        index = self.git.index
        index.read()
        index.remove(pycompat.fsdecode(f))
        index.write()

    def copied(self, path):
        # TODO: track copies?
        return None

    def prefetch_parents(self):
        # TODO

    @contextlib.contextmanager
    def changing_parents(self, repo):
        # TODO: track this maybe?
        yield

    def addparentchangecallback(self, category, callback):
        # TODO: should this be added to the dirstate interface?
        self._plchangecallbacks[category] = callback

    def setbranch(self, branch, transaction):
        raise error.Abort(
            b'git repos do not support branches. try using bookmarks'
        )

51929:93d872a06132

from __future__ import annotations

import contextlib
import os

from typing import (
    Any,
    Dict,
    Iterable,
    Iterator,
    List,
    Optional,
    Tuple,
)

from mercurial.node import sha1nodeconstants
from mercurial import (
    dirstatemap,
    error,

            sha1nodeconstants,
            self._use_dirstate_v2,
        )
        return self._map

    def p1(self) -> bytes:
        try:
            return self.git.head.peel().id.raw
        except pygit2.GitError:
            # Typically happens when peeling HEAD fails, as in an
            # empty repository.
            return sha1nodeconstants.nullid

    def p2(self) -> bytes:
        # TODO: MERGE_HEAD? something like that, right?
        return sha1nodeconstants.nullid

    def setparents(self, p1: bytes, p2: Optional[bytes] = None):
        if p2 is None:
            p2 = sha1nodeconstants.nullid
        assert p2 == sha1nodeconstants.nullid, b'TODO merging support'
        self.git.head.set_target(gitutil.togitnode(p1))

    def identity(self):
        return util.filestat.frompath(
            os.path.join(self._root, b'.git', b'index')
        )

    def branch(self) -> bytes:
        return b'default'

    def parents(self) -> List[bytes]:
        # TODO how on earth do we find p2 if a merge is in flight?
        return [self.p1(), sha1nodeconstants.nullid]

    def __iter__(self) -> Iterator[bytes]:
        return (pycompat.fsencode(f.path) for f in self.git.index)

    def items(self) -> Iterator[Tuple[bytes, intdirstate.DirstateItemT]]:
        for ie in self.git.index:
            yield ie.path, None  # value should be a DirstateItem

    # py2,3 compat forward
    iteritems = items

            gs = self.git.status_file(filename)
        except KeyError:
            return b'?'
        return _STATUS_MAP[gs]

    def __contains__(self, filename: Any) -> bool:
        try:
            gs = self.git.status_file(filename)
            return _STATUS_MAP[gs] != b'?'
        except KeyError:
            return False

    def status(
        self,
        match: matchmod.basematcher,
        subrepos: bool,
        ignored: bool,
        clean: bool,
        unknown: bool,
    ) -> intdirstate.StatusReturnT:
        listclean = clean
        # TODO handling of clean files - can we get that from git.status()?
        modified, added, removed, deleted, unknown, ignored, clean = (
            [],
            [],

                modified, added, removed, deleted, unknown, ignored, clean
            ),
            mtime_boundary,
        )

    def flagfunc(
        self, buildfallback: intdirstate.FlagFuncFallbackT
    ) -> intdirstate.FlagFuncReturnT:
        # TODO we can do better
        return buildfallback()

    def getcwd(self) -> bytes:
        # TODO is this a good way to do this?
        return os.path.dirname(
            os.path.dirname(pycompat.fsencode(self.git.path))
        )

    def get_entry(self, path: bytes) -> intdirstate.DirstateItemT:
        """return a DirstateItem for the associated path"""
        entry = self._map.get(path)
        if entry is None:
            return DirstateItem()
        return entry

    def normalize(
        self, path: bytes, isknown: bool = False, ignoremissing: bool = False
    ) -> bytes:
        normed = util.normcase(path)
        assert normed == path, b"TODO handling of case folding: %s != %s" % (
            normed,
            path,
        )
        return path

    @property
    def _checklink(self) -> bool:
        return util.checklink(os.path.dirname(pycompat.fsencode(self.git.path)))

    def copies(self) -> Dict[bytes, bytes]:
        # TODO support copies?
        return {}

    # # TODO what the heck is this
    _filecache = set()

    @property
    def is_changing_parents(self) -> bool:
        # TODO: we need to implement the context manager bits and
        # correctly stage/revert index edits.
        return False

    @property
    def is_changing_any(self) -> bool:
        # TODO: we need to implement the context manager bits and
        # correctly stage/revert index edits.
        return False

    def write(self, tr: Optional[intdirstate.TransactionT]) -> None:
        # TODO: call parent change callbacks

        if tr:

            def writeinner(category):

            tr.addpending(b'gitdirstate', writeinner)
        else:
            self.git.index.write()

    def pathto(self, f: bytes, cwd: Optional[bytes] = None) -> bytes:
        if cwd is None:
            cwd = self.getcwd()
        # TODO core dirstate does something about slashes here
        assert isinstance(f, bytes)
        r = util.pathto(self._root, cwd, f)
        return r

    def matches(self, match: matchmod.basematcher) -> Iterable[bytes]:
        for x in self.git.index:
            p = pycompat.fsencode(x.path)
            if match(p):
                yield p  # TODO: return list instead of yielding?

    def set_clean(self, f, parentfiledata):
        """Mark a file normal and clean."""
        # TODO: for now we just let libgit2 re-stat the file. We can
        # clearly do better.

    def set_possibly_dirty(self, f):
        """Mark a file normal, but possibly dirty."""
        # TODO: for now we just let libgit2 re-stat the file. We can
        # clearly do better.

    def walk(
        self,
        match: matchmod.basematcher,
        subrepos: Any,
        unknown: bool,
        ignored: bool,
        full: bool = True,
    ) -> intdirstate.WalkReturnT:
        # TODO: we need to use .status() and not iterate the index,
        # because the index doesn't force a re-walk and so `hg add` of
        # a new file without an intervening call to status will
        # silently do nothing.
        r = {}

        index = self.git.index
        index.read()
        index.remove(pycompat.fsdecode(f))
        index.write()

    def copied(self, file: bytes) -> Optional[bytes]:
        # TODO: track copies?
        return None

    def prefetch_parents(self):
        # TODO

    @contextlib.contextmanager
    def changing_parents(self, repo):
        # TODO: track this maybe?
        yield

    def addparentchangecallback(
        self, category: bytes, callback: intdirstate.AddParentChangeCallbackT
    ) -> None:
        # TODO: should this be added to the dirstate interface?
        self._plchangecallbacks[category] = callback

    def setbranch(
        self, branch: bytes, transaction: Optional[intdirstate.TransactionT]
    ) -> None:
        raise error.Abort(
            b'git repos do not support branches. try using bookmarks'
        )