1. camlspotter
  2. hgcachedoutgoing


hgcachedoutgoing / cout.py

# -*- coding: utf-8 -*-

# mercurial extension for cached version of ``hg outgoing``
# Copyright (C) 2011 by Takafumi Arakaki
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
# MA  02110-1301, USA.

"""Cached version of ``hg outgoing``"""

import shelve
import os
from mercurial import hg
from mercurial.node import short

def _checkid(ui, repo, dest):
    ui.debug("connecting to '{0}' to get the tip id.\n".format(dest))
    destrepo = hg.peer(ui, {}, dest)
    desttiprev = destrepo.lookup('tip')
    # get "TypeError: 'sshrepository' object is unsubscriptable"
    # if `destrepo['tip']` is used
    return desttiprev

def _get_cached_tip_id(ui, repo, dest, check):
    dbpath = os.path.join(repo.root, '.hg', 'cout.db')
    codb = shelve.open(dbpath)
    ui.debug("opened DB in '{0}'\n".format(dbpath))
    # get cached newest id on `dest` server (cdid)
    if dest in codb and not check:
        cdid = codb[dest]
        ui.debug("dest '{0}' is in .hg/cout.db.\n".format(dest))
        # check tip id of dest
        cdid = _checkid(ui, repo, dest)
        codb[dest] = cdid
    ui.debug("cdid='{0}' is successfully found.\n".format(short(cdid)))
    ui.debug("closed DB in '{0}'\n".format(dbpath))
    return cdid

def cout(ui, repo, dest=None, check=False, **opts):
    """Cached version of ``hg outgoing``"""

    dest = ui.expandpath(dest or 'default-push', dest or 'default')
    cdid = _get_cached_tip_id(ui, repo, dest, check)
    # check if the newest id on server is the tip on local
    if cdid in repo:
        cdtip = repo[cdid]
            "got context object of tip on dest (rev='{0}', short='{1}').\n"
            .format(cdtip.rev(), short(cdtip.node())))
        ui.debug("short(codb['tip'].node()) = '{0}'\n"
        uncommitted = len(repo.status()[0])  # num. of uncommitted files
        if cdtip == repo['tip'] and uncommitted == 0:
            return 1
        num_changes = repo['tip'].rev() - cdtip.rev()
        if num_changes > 0:
            ui.write("{0} changes to push.\n"
        if uncommitted:
            ui.write("{0} uncommitted files\n" .format(uncommitted))
        return 0
            "the tip ({0}) on dest ({1}) is not in current "
            "local repository.\n"
            "you might want to pull the changes.\n"
            .format(short(cdid), dest))
        return 0

def postpushhook(ui, repo, dest=None, **kwargs):
    """Check the tip of dest after ``hg push``"""
    dest = ui.expandpath(dest or 'default-push', dest or 'default')
    ui.debug("dest = '{0}'\n".format(dest))
    # NOTE:
    # assuming dest as argument might be the wrong way to get dest
    # of push command.

    _get_cached_tip_id(ui, repo, dest, check=True)
    # NOTE:
    # it is better if I can find the revs pushed via some API.
    # but calling _get_cached_tip_id is far more simple.

def uisetup(ui):
    ui.setconfig("hooks", "post-push.hgcachedoutgoing", postpushhook)

cmdtable = {
    # "command-name": (function-call, options-list, help-string)
    "cout": (
        [('c', 'check', False,
          'forcefully check the tip of DEST (ignore cache)'),
        "[options] [DEST]")