mercurial: comparison mercurial/phases.py

equal deleted inserted replaced

-:eef13b940887
+: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)
+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]}

changeset 46644	77e129be10de
parent 45790	5d65e04b6a80
child 47012	d55b71393907