comparison mercurial/phases.py @ 46644:77e129be10de

typing: add some type annotations to mercurial/phases.py

Some of these were helpful in typing other modules, and then I typed the
easy-ish ones.  Black forces the long `Phasedefaults` definition to be wrapped,
which pytype seems OK with (as shown with `reveal_type()`), but it does seem to
confuse PyCharm a bit.

Differential Revision: https://phab.mercurial-scm.org/D10126

46643:eef13b940887

    smartset,
    txnutil,
    util,
)

_fphasesentry = struct.Struct(b'>i20s')

# record phase index
public, draft, secret = range(3)
archived = 32  # non-continuous for compatibility
internal = 96  # non-continuous for compatibility
allphases = (public, draft, secret, archived, internal)
trackedphases = (draft, secret, archived, internal)
# record phase names

remotehiddenphases = (secret, archived, internal)
localhiddenphases = (internal, archived)


def supportinternal(repo):
    """True if the internal phase can be used on a repository"""
    return requirements.INTERNAL_PHASE_REQUIREMENT in repo.requirements


def _readroots(repo, phasedefaults=None):
    """Read phase roots from disk

    phasedefaults is a list of fn(repo, roots) callable, which are
    executed if the phase roots file does not exist. When phases are
    being initialized on an existing repository, this could be used to

        dirty = True
    return roots, dirty


def binaryencode(phasemapping):
    """encode a 'phase -> nodes' mapping into a binary stream

    The revision lists are encoded as (phase, root) pairs.
    """
    binarydata = []

            binarydata.append(_fphasesentry.pack(phase, head))
    return b''.join(binarydata)


def binarydecode(stream):
    """decode a binary stream into a 'phase -> nodes' mapping

    The (phase, root) pairs are turned back into a dictionary with
    the phase as index and the aggregated roots of that phase as value."""
    headsbyphase = {i: [] for i in allphases}

        data.insert(low + 1, (pycompat.xrange(rev, rev + 1), t))


class phasecache(object):
    def __init__(self, repo, phasedefaults, _load=True):
        if _load:
            # Cheap trick to allow shallow-copy without copy module
            self.phaseroots, self.dirty = _readroots(repo, phasedefaults)
            self._loadedrevslen = 0
            self._phasesets = None
            self.filterunknown(repo)
            self.opener = repo.svfs

    def hasnonpublicphases(self, repo):
        """detect if there are revisions with non-public phase"""
        repo = repo.unfiltered()
        cl = repo.changelog
        if len(cl) >= self._loadedrevslen:
            self.invalidate()

            for phase, revs in pycompat.iteritems(self.phaseroots)
            if phase != public
        )

    def nonpublicphaseroots(self, repo):
        """returns the roots of all non-public phases

        The roots are not minimized, so if the secret revisions are
        descendants of draft revisions, their roots will still be present.
        """

                if phase != public
            ]
        )

    def getrevset(self, repo, phases, subset=None):
        """return a smartset for the given phases"""
        self.loadphaserevs(repo)  # ensure phase's sets are loaded
        phases = set(phases)
        publicphase = public in phases

                lowerroots.update(ps)
                self._phasesets[phase] = ps
        self._loadedrevslen = len(cl)

    def loadphaserevs(self, repo):
        """ensure phase information is loaded in the object"""
        if self._phasesets is None:
            try:
                res = self._getphaserevsnative(repo)
                self._loadedrevslen, self._phasesets = res

    def invalidate(self):
        self._loadedrevslen = 0
        self._phasesets = None

    def phase(self, repo, rev):
        # We need a repo argument here to be able to build _phasesets
        # if necessary. The repository instance is not stored in
        # phasecache to avoid reference cycles. The changelog instance
        # is not stored because it is a filecache() property and can
        # be replaced without us being notified.

            )
            return True
        return False

    def filterunknown(self, repo):
        """remove unknown nodes from the phase boundary

        Nothing is lost as unknown nodes only hold data for their descendants.
        """
        filtered = False

    phcache.registernew(repo, tr, targetphase, revs)
    repo._phasecache.replace(phcache)


def listphases(repo):
    """List phases root for serialization over pushkey"""
    # Use ordered dictionary so behavior is deterministic.
    keys = util.sortdict()
    value = b'%i' % draft
    cl = repo.unfiltered().changelog

        keys[b'publishing'] = b'True'
    return keys


def pushphase(repo, nhex, oldphasestr, newphasestr):
    """List phases root for serialization over pushkey"""
    repo = repo.unfiltered()
    with repo.lock():
        currentphase = repo[nhex].phase()
        newphase = abs(int(newphasestr))  # let's avoid negative index surprise

    return pycompat.maplist(cl.node, sorted(new_heads))


def newcommitphase(ui):
    """helper to get the target phase of new commit

    Handle all possible values for the phases.new-commit options.

    """

            _(b"phases.new-commit: not a valid phase name ('%s')") % v
        )


def hassecret(repo):
    """utility function that check if a repo have any secret changeset."""
    return bool(repo._phasecache.phaseroots[secret])


def preparehookargs(node, old, new):
    if old is None:
        old = b''
    else:
        old = phasenames[old]
    return {b'node': node, b'oldphase': old, b'phase': phasenames[new]}

46644:77e129be10de

    smartset,
    txnutil,
    util,
)

if pycompat.TYPE_CHECKING:
    from typing import (
        Any,
        Callable,
        Dict,
        Iterable,
        List,
        Optional,
        Set,
        Tuple,
    )
    from . import (
        localrepo,
        ui as uimod,
    )

    Phaseroots = Dict[int, Set[bytes]]
    Phasedefaults = List[
        Callable[[localrepo.localrepository, Phaseroots], Phaseroots]
    ]


_fphasesentry = struct.Struct(b'>i20s')

# record phase index
public, draft, secret = range(3)  # type: int
archived = 32  # non-continuous for compatibility
internal = 96  # non-continuous for compatibility
allphases = (public, draft, secret, archived, internal)
trackedphases = (draft, secret, archived, internal)
# record phase names

remotehiddenphases = (secret, archived, internal)
localhiddenphases = (internal, archived)


def supportinternal(repo):
    # type: (localrepo.localrepository) -> bool
    """True if the internal phase can be used on a repository"""
    return requirements.INTERNAL_PHASE_REQUIREMENT in repo.requirements


def _readroots(repo, phasedefaults=None):
    # type: (localrepo.localrepository, Optional[Phasedefaults]) -> Tuple[Phaseroots, bool]
    """Read phase roots from disk

    phasedefaults is a list of fn(repo, roots) callable, which are
    executed if the phase roots file does not exist. When phases are
    being initialized on an existing repository, this could be used to

        dirty = True
    return roots, dirty


def binaryencode(phasemapping):
    # type: (Dict[int, List[bytes]]) -> bytes
    """encode a 'phase -> nodes' mapping into a binary stream

    The revision lists are encoded as (phase, root) pairs.
    """
    binarydata = []

            binarydata.append(_fphasesentry.pack(phase, head))
    return b''.join(binarydata)


def binarydecode(stream):
    # type: (...) -> Dict[int, List[bytes]]
    """decode a binary stream into a 'phase -> nodes' mapping

    The (phase, root) pairs are turned back into a dictionary with
    the phase as index and the aggregated roots of that phase as value."""
    headsbyphase = {i: [] for i in allphases}

        data.insert(low + 1, (pycompat.xrange(rev, rev + 1), t))


class phasecache(object):
    def __init__(self, repo, phasedefaults, _load=True):
        # type: (localrepo.localrepository, Optional[Phasedefaults], bool) -> None
        if _load:
            # Cheap trick to allow shallow-copy without copy module
            self.phaseroots, self.dirty = _readroots(repo, phasedefaults)
            self._loadedrevslen = 0
            self._phasesets = None
            self.filterunknown(repo)
            self.opener = repo.svfs

    def hasnonpublicphases(self, repo):
        # type: (localrepo.localrepository) -> bool
        """detect if there are revisions with non-public phase"""
        repo = repo.unfiltered()
        cl = repo.changelog
        if len(cl) >= self._loadedrevslen:
            self.invalidate()

            for phase, revs in pycompat.iteritems(self.phaseroots)
            if phase != public
        )

    def nonpublicphaseroots(self, repo):
        # type: (localrepo.localrepository) -> Set[bytes]
        """returns the roots of all non-public phases

        The roots are not minimized, so if the secret revisions are
        descendants of draft revisions, their roots will still be present.
        """

                if phase != public
            ]
        )

    def getrevset(self, repo, phases, subset=None):
        # type: (localrepo.localrepository, Iterable[int], Optional[Any]) -> Any
        # TODO: finish typing this
        """return a smartset for the given phases"""
        self.loadphaserevs(repo)  # ensure phase's sets are loaded
        phases = set(phases)
        publicphase = public in phases

                lowerroots.update(ps)
                self._phasesets[phase] = ps
        self._loadedrevslen = len(cl)

    def loadphaserevs(self, repo):
        # type: (localrepo.localrepository) -> None
        """ensure phase information is loaded in the object"""
        if self._phasesets is None:
            try:
                res = self._getphaserevsnative(repo)
                self._loadedrevslen, self._phasesets = res

    def invalidate(self):
        self._loadedrevslen = 0
        self._phasesets = None

    def phase(self, repo, rev):
        # type: (localrepo.localrepository, int) -> int
        # We need a repo argument here to be able to build _phasesets
        # if necessary. The repository instance is not stored in
        # phasecache to avoid reference cycles. The changelog instance
        # is not stored because it is a filecache() property and can
        # be replaced without us being notified.

            )
            return True
        return False

    def filterunknown(self, repo):
        # type: (localrepo.localrepository) -> None
        """remove unknown nodes from the phase boundary

        Nothing is lost as unknown nodes only hold data for their descendants.
        """
        filtered = False

    phcache.registernew(repo, tr, targetphase, revs)
    repo._phasecache.replace(phcache)


def listphases(repo):
    # type: (localrepo.localrepository) -> Dict[bytes, bytes]
    """List phases root for serialization over pushkey"""
    # Use ordered dictionary so behavior is deterministic.
    keys = util.sortdict()
    value = b'%i' % draft
    cl = repo.unfiltered().changelog

        keys[b'publishing'] = b'True'
    return keys


def pushphase(repo, nhex, oldphasestr, newphasestr):
    # type: (localrepo.localrepository, bytes, bytes, bytes) -> bool
    """List phases root for serialization over pushkey"""
    repo = repo.unfiltered()
    with repo.lock():
        currentphase = repo[nhex].phase()
        newphase = abs(int(newphasestr))  # let's avoid negative index surprise

    return pycompat.maplist(cl.node, sorted(new_heads))


def newcommitphase(ui):
    # type: (uimod.ui) -> int
    """helper to get the target phase of new commit

    Handle all possible values for the phases.new-commit options.

    """

            _(b"phases.new-commit: not a valid phase name ('%s')") % v
        )


def hassecret(repo):
    # type: (localrepo.localrepository) -> bool
    """utility function that check if a repo have any secret changeset."""
    return bool(repo._phasecache.phaseroots[secret])


def preparehookargs(node, old, new):
    # type: (bytes, Optional[int], Optional[int]) -> Dict[bytes, bytes]
    if old is None:
        old = b''
    else:
        old = phasenames[old]
    return {b'node': node, b'oldphase': old, b'phase': phasenames[new]}