view hglib/context.py @ 142:fe74d5599539

hglib: wrap all application string literals in util.b() (issue4520) Conversion also included changing use of string interpolation to string concatenation as bytes interpolation does not exist in Python 3. Indexing related to bytes was also changed to length-1 bytes through slicing as Python 3 returns an int in this instance. Tests have not been switched to using util.b() so that the change to application code can be independently verified as not being broken.
author Brett Cannon <brett@python.org>
date Sun, 08 Mar 2015 13:08:37 -0400
parents a05cdc1579df
children c1b966866ed7
line wrap: on
line source

from hglib.error import CommandError
import client, util, templates
from hglib.util import b

_nullcset = [b('-1'), b('0000000000000000000000000000000000000000'), b(''),
             b(''), b(''), b(''), b('')]

class changectx(object):
    """A changecontext object makes access to data related to a particular
    changeset convenient."""
    def __init__(self, repo, changeid=b('')):
        """changeid is a revision number, node, or tag"""
        if changeid == b(''):
            changeid = b('.')
        self._repo = repo
        if isinstance(changeid, client.revision):
            cset = changeid
        elif changeid == -1:
            cset = _nullcset
        else:
            if isinstance(changeid, (long, int)):
                changeid = b('rev(') + str(changeid).encode('latin-1') + b(')')

            notfound = False
            try:
                cset = self._repo.log(changeid)
                # hg bbf4f3dfd700 gave a null result for tip+1
                if (cset and cset[0][1] == _nullcset[1]
                    and cset[0][0] != _nullcset[0]):
                    notfound = True
            except CommandError:
                notfound = True

            if notfound or not len(cset):
                raise ValueError('changeid %r not found in repo' % changeid)
            if len(cset) > 1:
                raise ValueError('changeid must yield a single changeset')
            cset = cset[0]

        self._rev, self._node, self._tags = cset[:3]
        self._branch, self._author, self._description, self._date = cset[3:]

        self._rev = int(self._rev)

        self._tags = self._tags.split()
        try:
            self._tags.remove(b('tip'))
        except ValueError:
            pass

        self._ignored = None
        self._clean = None

    def __str__(self):
        return self._node[:12]

    def __int__(self):
        return self._rev

    def __repr__(self):
        return "<changectx %s>" % str(self)

    def __hash__(self):
        try:
            return hash(self._rev)
        except AttributeError:
            return id(self)

    def __eq__(self, other):
        try:
            return self._rev == other._rev
        except AttributeError:
            return False

    def __ne__(self, other):
        return not (self == other)

    def __nonzero__(self):
        return self._rev != -1

    def __contains__(self, key):
        return key in self._manifest

    def __iter__(self):
        for f in sorted(self._manifest):
            yield f

    @util.propertycache
    def _status(self):
        return self._parsestatus(self._repo.status(change=self))[:4]

    def _parsestatus(self, stat):
        d = dict((c, [])
                 for c in (b('M'), b('A'), b('R'), b('!'), b('?'), b('I'),
                    b('C'), b(' ')))
        for k, path in stat:
            d[k].append(path)
        return (d[b('M')], d[b('A')], d[b('R')], d[b('!')], d[b('?')],
                d[b('I')], d[b('C')])

    def status(self, ignored=False, clean=False):
        """Explicit status query
        Unless this method is used to query the working copy status, the
        _status property will implicitly read the status using its default
        arguments."""
        stat = self._parsestatus(self._repo.status(change=self, ignored=ignored,
                                                   clean=clean))
        self._unknown = self._ignored = self._clean = None
        if ignored:
            self._ignored = stat[5]
        if clean:
            self._clean = stat[6]
        self._status = stat[:4]
        return stat

    def rev(self):
        return self._rev

    def node(self):
        return self._node

    def tags(self):
        return self._tags

    def branch(self):
        return self._branch

    def author(self):
        return self._author

    def user(self):
        return self._author

    def date(self):
        return self._date

    def description(self):
        return self._description

    def files(self):
        return sorted(self._status[0] + self._status[1] + self._status[2])

    def modified(self):
        return self._status[0]

    def added(self):
        return self._status[1]

    def removed(self):
        return self._status[2]

    def ignored(self):
        if self._ignored is None:
            self.status(ignored=True)
        return self._ignored

    def clean(self):
        if self._clean is None:
            self.status(clean=True)
        return self._clean

    @util.propertycache
    def _manifest(self):
        d = {}
        for node, p, e, s, path in self._repo.manifest(rev=self):
            d[path] = node
        return d

    def manifest(self):
        return self._manifest

    def hex(self):
        return hex(self._node)

    @util.propertycache
    def _parents(self):
        """return contexts for each parent changeset"""
        par = self._repo.parents(rev=self)
        if not par:
            return [changectx(self._repo, -1)]
        return [changectx(self._repo, int(cset.rev)) for cset in par]

    def parents(self):
        return self._parents

    def p1(self):
        return self._parents[0]

    def p2(self):
        if len(self._parents) == 2:
            return self._parents[1]
        return changectx(self._repo, -1)

    @util.propertycache
    def _bookmarks(self):
        books = [bm for bm in self._repo.bookmarks()[0] if bm[1] == self._rev]

        bms = []
        for name, r, n in books:
            bms.append(name)
        return bms

    def bookmarks(self):
        return self._bookmarks

    def hidden(self):
        """return True if the changeset is hidden, else False"""
        return bool(self._repo.log(revrange=self._node + b(' and hidden()'),
                                   hidden=True))

    def phase(self):
        """return the phase of the changeset (public, draft or secret)"""
        return self._repo.phase(str(self._rev))[0][1]

    def children(self):
        """return contexts for each child changeset"""
        for c in self._repo.log(b('children(') + self._node + b(')')):
            yield changectx(self._repo, c)

    def ancestors(self):
        for a in self._repo.log(b('ancestors(') + self._node + b(')')):
            yield changectx(self._repo, a)

    def descendants(self):
        for d in self._repo.log(b('descendants(') + self._node + b(')')):
            yield changectx(self._repo, d)

    def ancestor(self, c2):
        """
        return the ancestor context of self and c2
        """
        return changectx(self._repo,
                         b('ancestor(') + self + b(', ') + c2 + b(')'))