# mq.py - patch queues for mercurial## Copyright 2005, 2006 Chris Mason <mason@suse.com>## This software may be used and distributed according to the terms of the# GNU General Public License version 2, incorporated herein by reference. '''manage a stack of patches This extension lets you work with a stack of patches in a Mercurialrepository. It manages two stacks of patches - all known patches, andapplied patches (subset of known patches). Known patches are represented as patch files in the .hg/patchesdirectory. Applied patches are both patch files and changesets. Common tasks (use "hg help command" for more details)::   prepare repository to work with patches   qinit  create new patch                          qnew  import existing patch                     qimport   print patch series                        qseries  print applied patches                     qapplied  print name of top applied patch           qtop   add known patch to applied stack          qpush  remove patch from applied stack           qpop  refresh contents of top applied patch     qrefresh''' from mercurial.i18n import _from mercurial.node import bin, hex, short, nullid, nullrevfrom mercurial.lock import releasefrom mercurial import commands, cmdutil, hg, patch, utilfrom mercurial import repair, extensions, url, errorimport os, sys, re, errno commands.norepo += " qclone" # Patch names looks like unix-file names.# They must be joinable with queue directory and result in the patch path.normname = util.normpath class statusentry(object):    def __init__(self, rev, name=None):        if not name:            fields = rev.split(':', 1)            if len(fields) == 2:                self.rev, self.name = fields            else:                self.rev, self.name = None, None        else:            self.rev, self.name = rev, name     def __str__(self):        return self.rev + ':' + self.name class patchheader(object):    def __init__(self, pf):        def eatdiff(lines):            while lines:                l = lines[-1]                if (l.startswith("diff -") or                    l.startswith("Index:") or                    l.startswith("===========")):                    del lines[-1]                else:                    break        def eatempty(lines):            while lines:                l = lines[-1]                if re.match('\s*$', l):                    del lines[-1]                else:                    break         message = []        comments = []        user = None        date = None        format = None        subject = None        diffstart = 0         for line in file(pf):            line = line.rstrip()            if line.startswith('diff --git'):                diffstart = 2                break            if diffstart:                if line.startswith('+++ '):                    diffstart = 2                break            if line.startswith("--- "):                diffstart = 1                continue            elif format == "hgpatch":                # parse values when importing the result of an hg export                if line.startswith("# User "):                    user = line[7:]                elif line.startswith("# Date "):                    date = line[7:]                elif not line.startswith("# ") and line:                    message.append(line)                    format = None            elif line == '# HG changeset patch':                message = []                format = "hgpatch"            elif (format != "tagdone" and (line.startswith("Subject: ") or                                           line.startswith("subject: "))):                subject = line[9:]                format = "tag"            elif (format != "tagdone" and (line.startswith("From: ") or                                           line.startswith("from: "))):                user = line[6:]                format = "tag"            elif format == "tag" and line == "":                # when looking for tags (subject: from: etc) they                # end once you find a blank line in the source                format = "tagdone"            elif message or line:                message.append(line)            comments.append(line)         eatdiff(message)        eatdiff(comments)        eatempty(message)        eatempty(comments)         # make sure message isn't empty        if format and format.startswith("tag") and subject:            message.insert(0, "")            message.insert(0, subject)         self.message = message        self.comments = comments        self.user = user        self.date = date        self.haspatch = diffstart > 1     def setuser(self, user):        if not self.updateheader(['From: ', '# User '], user):            try:                patchheaderat = self.comments.index('# HG changeset patch')                self.comments.insert(patchheaderat + 1, '# User ' + user)            except ValueError:                if self._hasheader(['Date: ']):                    self.comments = ['From: ' + user] + self.comments                else:                    tmp = ['# HG changeset patch', '# User ' + user, '']                    self.comments = tmp + self.comments        self.user = user     def setdate(self, date):        if not self.updateheader(['Date: ', '# Date '], date):            try:                patchheaderat = self.comments.index('# HG changeset patch')                self.comments.insert(patchheaderat + 1, '# Date ' + date)            except ValueError:                if self._hasheader(['From: ']):                    self.comments = ['Date: ' + date] + self.comments                else:                    tmp = ['# HG changeset patch', '# Date ' + date, '']                    self.comments = tmp + self.comments        self.date = date     def setmessage(self, message):        if self.comments:            self._delmsg()        self.message = [message]        self.comments += self.message     def updateheader(self, prefixes, new):        '''Update all references to a field in the patch header.        Return whether the field is present.'''        res = False        for prefix in prefixes:            for i in xrange(len(self.comments)):                if self.comments[i].startswith(prefix):                    self.comments[i] = prefix + new                    res = True                    break        return res     def _hasheader(self, prefixes):        '''Check if a header starts with any of the given prefixes.'''        for prefix in prefixes:            for comment in self.comments:                if comment.startswith(prefix):                    return True        return False     def __str__(self):        if not self.comments:            return ''        return '\n'.join(self.comments) + '\n\n'     def _delmsg(self):        '''Remove existing message, keeping the rest of the comments fields.        If comments contains 'subject: ', message will prepend        the field and a blank line.'''        if self.message:            subj = 'subject: ' + self.message[0].lower()            for i in xrange(len(self.comments)):                if subj == self.comments[i].lower():                    del self.comments[i]                    self.message = self.message[2:]                    break        ci = 0        for mi in self.message:            while mi != self.comments[ci]:                ci += 1            del self.comments[ci] class queue(object):    def __init__(self, ui, path, patchdir=None):        self.basepath = path        self.path = patchdir or os.path.join(path, "patches")        self.opener = util.opener(self.path)        self.ui = ui        self.applied_dirty = 0        self.series_dirty = 0        self.series_path = "series"        self.status_path = "status"        self.guards_path = "guards"        self.active_guards = None        self.guards_dirty = False        self._diffopts = None     @util.propertycache    def applied(self):        if os.path.exists(self.join(self.status_path)):            lines = self.opener(self.status_path).read().splitlines()            return [statusentry(l) for l in lines]        return []     @util.propertycache    def full_series(self):        if os.path.exists(self.join(self.series_path)):            return self.opener(self.series_path).read().splitlines()        return []     @util.propertycache    def series(self):        self.parse_series()        return self.series     @util.propertycache    def series_guards(self):        self.parse_series()        return self.series_guards     def invalidate(self):        for a in 'applied full_series series series_guards'.split():            if a in self.__dict__:                delattr(self, a)        self.applied_dirty = 0        self.series_dirty = 0        self.guards_dirty = False        self.active_guards = None     def diffopts(self):        if self._diffopts is None:            self._diffopts = patch.diffopts(self.ui)        return self._diffopts     def join(self, *p):        return os.path.join(self.path, *p)     def find_series(self, patch):        pre = re.compile("(\s*)([^#]+)")        index = 0        for l in self.full_series:            m = pre.match(l)            if m:                s = m.group(2)                s = s.rstrip()                if s == patch:                    return index            index += 1        return None     guard_re = re.compile(r'\s?#([-+][^-+# \t\r\n\f][^# \t\r\n\f]*)')     def parse_series(self):        self.series = []        self.series_guards = []        for l in self.full_series:            h = l.find('#')            if h == -1:                patch = l                comment = ''            elif h == 0:                continue            else:                patch = l[:h]                comment = l[h:]            patch = patch.strip()            if patch:                if patch in self.series:                    raise util.Abort(_('%s appears more than once in %s') %                                     (patch, self.join(self.series_path)))                self.series.append(patch)                self.series_guards.append(self.guard_re.findall(comment))     def check_guard(self, guard):        if not guard:            return _('guard cannot be an empty string')        bad_chars = '# \t\r\n\f'        first = guard[0]        if first in '-+':            return (_('guard %r starts with invalid character: %r') %                      (guard, first))        for c in bad_chars:            if c in guard:                return _('invalid character in guard %r: %r') % (guard, c)     def set_active(self, guards):        for guard in guards:            bad = self.check_guard(guard)            if bad:                raise util.Abort(bad)        guards = sorted(set(guards))        self.ui.debug(_('active guards: %s\n') % ' '.join(guards))        self.active_guards = guards        self.guards_dirty = True     def active(self):        if self.active_guards is None:            self.active_guards = []            try:                guards = self.opener(self.guards_path).read().split()            except IOError, err:                if err.errno != errno.ENOENT: raise                guards = []            for i, guard in enumerate(guards):                bad = self.check_guard(guard)                if bad:                    self.ui.warn('%s:%d: %s\n' %                                 (self.join(self.guards_path), i + 1, bad))                else:                    self.active_guards.append(guard)        return self.active_guards     def set_guards(self, idx, guards):        for g in guards:            if len(g) < 2:                raise util.Abort(_('guard %r too short') % g)            if g[0] not in '-+':                raise util.Abort(_('guard %r starts with invalid char') % g)            bad = self.check_guard(g[1:])            if bad:                raise util.Abort(bad)        drop = self.guard_re.sub('', self.full_series[idx])        self.full_series[idx] = drop + ''.join([' #' + g for g in guards])        self.parse_series()        self.series_dirty = True     def pushable(self, idx):        if isinstance(idx, str):            idx = self.series.index(idx)        patchguards = self.series_guards[idx]        if not patchguards:            return True, None        guards = self.active()        exactneg = [g for g in patchguards if g[0] == '-' and g[1:] in guards]        if exactneg:            return False, exactneg[0]        pos = [g for g in patchguards if g[0] == '+']        exactpos = [g for g in pos if g[1:] in guards]        if pos:            if exactpos:                return True, exactpos[0]            return False, pos        return True, ''     def explain_pushable(self, idx, all_patches=False):        write = all_patches and self.ui.write or self.ui.warn        if all_patches or self.ui.verbose:            if isinstance(idx, str):                idx = self.series.index(idx)            pushable, why = self.pushable(idx)            if all_patches and pushable:                if why is None:                    write(_('allowing %s - no guards in effect\n') %                          self.series[idx])                else:                    if not why:                        write(_('allowing %s - no matching negative guards\n') %                              self.series[idx])                    else:                        write(_('allowing %s - guarded by %r\n') %                              (self.series[idx], why))            if not pushable:                if why:                    write(_('skipping %s - guarded by %r\n') %                          (self.series[idx], why))                else:                    write(_('skipping %s - no matching guards\n') %                          self.series[idx])     def save_dirty(self):        def write_list(items, path):            fp = self.opener(path, 'w')            for i in items:                fp.write("%s\n" % i)            fp.close()        if self.applied_dirty: write_list(map(str, self.applied), self.status_path)        if self.series_dirty: write_list(self.full_series, self.series_path)        if self.guards_dirty: write_list(self.active_guards, self.guards_path)     def removeundo(self, repo):        undo = repo.sjoin('undo')        if not os.path.exists(undo):            return        try:            os.unlink(undo)        except OSError, inst:            self.ui.warn(_('error removing undo: %s\n') % str(inst))     def printdiff(self, repo, node1, node2=None, files=None,                  fp=None, changes=None, opts={}):        m = cmdutil.match(repo, files, opts)        chunks = patch.diff(repo, node1, node2, m, changes, self.diffopts())        write = fp is None and repo.ui.write or fp.write        for chunk in chunks:            write(chunk)     def mergeone(self, repo, mergeq, head, patch, rev):        # first try just applying the patch        (err, n) = self.apply(repo, [ patch ], update_status=False,                              strict=True, merge=rev)         if err == 0:            return (err, n)         if n is None:            raise util.Abort(_("apply failed for patch %s") % patch)         self.ui.warn(_("patch didn't work out, merging %s\n") % patch)         # apply failed, strip away that rev and merge.        hg.clean(repo, head)        self.strip(repo, n, update=False, backup='strip')         ctx = repo[rev]        ret = hg.merge(repo, rev)        if ret:            raise util.Abort(_("update returned %d") % ret)        n = repo.commit(ctx.description(), ctx.user(), force=True)        if n is None:            raise util.Abort(_("repo commit failed"))        try:            ph = patchheader(mergeq.join(patch))        except:            raise util.Abort(_("unable to read %s") % patch)         patchf = self.opener(patch, "w")        comments = str(ph)        if comments:            patchf.write(comments)        self.printdiff(repo, head, n, fp=patchf)        patchf.close()        self.removeundo(repo)        return (0, n)     def qparents(self, repo, rev=None):        if rev is None:            (p1, p2) = repo.dirstate.parents()            if p2 == nullid:                return p1            if len(self.applied) == 0:                return None            return bin(self.applied[-1].rev)        pp = repo.changelog.parents(rev)        if pp[1] != nullid:            arevs = [ x.rev for x in self.applied ]            p0 = hex(pp[0])            p1 = hex(pp[1])            if p0 in arevs:                return pp[0]            if p1 in arevs:                return pp[1]        return pp[0]     def mergepatch(self, repo, mergeq, series):        if len(self.applied) == 0:            # each of the patches merged in will have two parents.  This            # can confuse the qrefresh, qdiff, and strip code because it            # needs to know which parent is actually in the patch queue.            # so, we insert a merge marker with only one parent.  This way            # the first patch in the queue is never a merge patch            #            pname = ".hg.patches.merge.marker"            n = repo.commit('[mq]: merge marker', force=True)            self.removeundo(repo)            self.applied.append(statusentry(hex(n), pname))            self.applied_dirty = 1 

Lines 1–500 of 2653. Load more