mercurial/phases.py

51401:68289ed170c7

"""


import struct
import typing

from typing import (
    Any,
    Callable,
    Dict,

            # 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: "localrepo.localrepository") -> bool:
        """detect if there are revisions with non-public phase"""
        repo = repo.unfiltered()
        cl = repo.changelog

        # Shallow copy meant to ensure isolation in
        # advance/retractboundary(), nothing more.
        ph = self.__class__(None, None, _load=False)
        ph._phaseroots = self._phaseroots.copy()
        ph.dirty = self.dirty
        ph.opener = self.opener
        ph._loadedrevslen = self._loadedrevslen
        ph._phasesets = self._phasesets
        return ph

    def replace(self, phcache):
        """replace all values in 'self' with content of phcache"""
        for a in (
            '_phaseroots',
            'dirty',
            'opener',
            '_loadedrevslen',
            '_phasesets',
        ):
            setattr(self, a, getattr(phcache, a))

        for phase in trackedphases:
            if rev in self._phasesets[phase]:
                return phase
        return public

    def write(self):
        if not self.dirty:
            return
        f = self.opener(b'phaseroots', b'w', atomictemp=True, checkambig=True)
        try:
            self._write(f)
        finally:
            f.close()

    def _write(self, fp):
        for phase, roots in self._phaseroots.items():
            for h in sorted(roots):
                fp.write(b'%i %s\n' % (phase, hex(h)))
        self.dirty = False

    def _updateroots(self, phase, newroots, tr):
        self._phaseroots[phase] = newroots
        self.invalidate()
        self.dirty = True

        tr.addfilegenerator(b'phase', (b'phaseroots',), self._write)
        tr.hookargs[b'phases_moved'] = b'1'

    def registernew(self, repo, tr, targetphase, revs):
        repo = repo.unfiltered()
        self._retractboundary(repo, tr, targetphase, [], revs=revs)

            roots = {
                ctx.node()
                for ctx in repo.set(b'roots((%ln::) - %ld)', olds, affected)
            }
            if olds != roots:
                self._updateroots(phase, roots, tr)
                # some roots may need to be declared for lower phases
                delroots.extend(olds - roots)
        if not dryrun:
            # declare deleted root in the target phase
            if targetphase != 0:

            finalroots = {rev for rev in currentroots if rev < minnewroot}
            finalroots.update(updatedroots)
        if finalroots != oldroots:
            self._updateroots(
                targetphase, {tonode(rev) for rev in finalroots}, tr
            )
            return True
        return False

    def register_strip(

        assert repo.filtername is None
        to_rev = repo.changelog.index.rev
        for targetphase, nodes in list(self._phaseroots.items()):
            filtered = {n for n in nodes if to_rev(n) >= strip_rev}
            if filtered:
                self._updateroots(targetphase, nodes - filtered, tr)
        self.invalidate()

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

51402:04111ef08fb0

"""


import struct
import typing
import weakref

from typing import (
    Any,
    Callable,
    Dict,

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

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

        # Shallow copy meant to ensure isolation in
        # advance/retractboundary(), nothing more.
        ph = self.__class__(None, None, _load=False)
        ph._phaseroots = self._phaseroots.copy()
        ph.dirty = self.dirty
        ph._loadedrevslen = self._loadedrevslen
        ph._phasesets = self._phasesets
        return ph

    def replace(self, phcache):
        """replace all values in 'self' with content of phcache"""
        for a in (
            '_phaseroots',
            'dirty',
            '_loadedrevslen',
            '_phasesets',
        ):
            setattr(self, a, getattr(phcache, a))

        for phase in trackedphases:
            if rev in self._phasesets[phase]:
                return phase
        return public

    def write(self, repo):
        if not self.dirty:
            return
        f = repo.svfs(b'phaseroots', b'w', atomictemp=True, checkambig=True)
        try:
            self._write(repo.unfiltered(), f)
        finally:
            f.close()

    def _write(self, repo, fp):
        assert repo.filtername is None
        for phase, roots in self._phaseroots.items():
            for h in sorted(roots):
                fp.write(b'%i %s\n' % (phase, hex(h)))
        self.dirty = False

    def _updateroots(self, repo, phase, newroots, tr):
        self._phaseroots[phase] = newroots
        self.invalidate()
        self.dirty = True

        assert repo.filtername is None
        wrepo = weakref.ref(repo)

        def tr_write(fp):
            repo = wrepo()
            assert repo is not None
            self._write(repo, fp)

        tr.addfilegenerator(b'phase', (b'phaseroots',), tr_write)
        tr.hookargs[b'phases_moved'] = b'1'

    def registernew(self, repo, tr, targetphase, revs):
        repo = repo.unfiltered()
        self._retractboundary(repo, tr, targetphase, [], revs=revs)

            roots = {
                ctx.node()
                for ctx in repo.set(b'roots((%ln::) - %ld)', olds, affected)
            }
            if olds != roots:
                self._updateroots(repo, phase, roots, tr)
                # some roots may need to be declared for lower phases
                delroots.extend(olds - roots)
        if not dryrun:
            # declare deleted root in the target phase
            if targetphase != 0:

            finalroots = {rev for rev in currentroots if rev < minnewroot}
            finalroots.update(updatedroots)
        if finalroots != oldroots:
            self._updateroots(
                repo,
                targetphase,
                {tonode(rev) for rev in finalroots},
                tr,
            )
            return True
        return False

    def register_strip(

        assert repo.filtername is None
        to_rev = repo.changelog.index.rev
        for targetphase, nodes in list(self._phaseroots.items()):
            filtered = {n for n in nodes if to_rev(n) >= strip_rev}
            if filtered:
                self._updateroots(repo, targetphase, nodes - filtered, tr)
        self.invalidate()

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