view hglib/context.py @ 221:a2afbf236ca8

hglib tests: remove deprecated constructions

This mostly removes usage of 'assertEquals' (replaced with 'assertEqual'),
as well as opening files without closing them
(fixed using a 'with' statement).

import hglib.client  # Circular dependency.
from hglib import util, templates
from hglib.error import CommandError
from hglib.util import b, strtobytes, integertypes

_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, hglib.client.revision):
            cset = changeid
        elif changeid == -1:
            cset = _nullcset
        else:
            if isinstance(changeid, integertypes):
                changeid = b('rev(') + strtobytes(changeid) + 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].decode('latin-1')

    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 __bool__(self):
        return self.__nonzero__()

    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=strtobytes(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=strtobytes(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=strtobytes(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=strtobytes(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(strtobytes(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(')'))