Mercurial > hgsubversion
view hgsubversion/svnexternals.py @ 882:07234759a3f7 1.4
svnexternals: fix handling of subrepo pinned to HEAD in hg 2.2
hg 2.2 is trying harder to see through our lies but we will prevail. Override
subrepo.basestate() so we can pretend the svn revision number is HEAD, honest.
author | Patrick Mezard <patrick@mezard.eu> |
---|---|
date | Thu, 26 Apr 2012 16:54:03 +0200 |
parents | 9ce00cb1d676 |
children | 91cf81a36236 |
line wrap: on
line source
import cStringIO import os, re, shutil, stat, subprocess from mercurial import util as hgutil from mercurial.i18n import _ try: from mercurial import subrepo # require svnsubrepo and hg >= 1.7.1 subrepo.svnsubrepo hgutil.checknlink except (ImportError, AttributeError), e: subrepo = None passpegrev = True # see svnsubrepo below try: canonpath = hgutil.canonpath except (ImportError, AttributeError): from mercurial import scmutil canonpath = scmutil.canonpath passpegrev = False import util class externalsfile(dict): """Map svn directories to lists of externals entries. """ def __init__(self): super(externalsfile, self).__init__() self.encoding = 'utf-8' def __setitem__(self, key, value): if value is None: value = [] elif isinstance(value, basestring): value = value.splitlines() if key == '.': key = '' if not value: if key in self: del self[key] else: super(externalsfile, self).__setitem__(key, value) def write(self): fp = cStringIO.StringIO() for target in sorted(self): lines = self[target] if not lines: continue if not target: target = '.' fp.write('[%s]\n' % target) for l in lines: l = ' ' + l + '\n' fp.write(l) return fp.getvalue() def read(self, data): self.clear() fp = cStringIO.StringIO(data) target = None for line in fp.readlines(): if not line.strip(): continue if line.startswith('['): line = line.strip() if line[-1] != ']': raise hgutil.Abort('invalid externals section name: %s' % line) target = line[1:-1] if target == '.': target = '' elif line.startswith(' '): line = line.rstrip('\n') if target is None or not line: continue self.setdefault(target, []).append(line[1:]) def diff(ext1, ext2): """Compare 2 externalsfile and return a list of tuples like (dir, value1, value2) where value1 is the external value in ext1 for dir or None, and value2 the same in ext2. """ changes = [] for d in ext1: if d not in ext2: changes.append((d, '\n'.join(ext1[d]), None)) elif ext1[d] != ext2[d]: changes.append((d, '\n'.join(ext1[d]), '\n'.join(ext2[d]))) for d in ext2: if d not in ext1: changes.append((d, None, '\n'.join(ext2[d]))) return changes class BadDefinition(Exception): pass re_defold = re.compile(r'^\s*(.*?)\s+(?:-r\s*(\d+|\{REV\})\s+)?([a-zA-Z+]+://.*)\s*$') re_defnew = re.compile(r'^\s*(?:-r\s*(\d+|\{REV\})\s+)?((?:[a-zA-Z+]+://|\^/).*)\s+(\S+)\s*$') re_scheme = re.compile(r'^[a-zA-Z+]+://') def parsedefinition(line): """Parse an external definition line, return a tuple (path, rev, source) or raise BadDefinition. """ # The parsing is probably not correct wrt path with whitespaces or # potential quotes. svn documentation is not really talkative about # these either. pegrev, revgroup = None, 1 m = re_defnew.search(line) if m: rev, source, path = m.group(1, 2, 3) if '@' in source: source, pegrev = source.rsplit('@', 1) else: m = re_defold.search(line) if not m: raise BadDefinition() revgroup = 2 path, rev, source = m.group(1, 2, 3) try: int(rev) # ensure revision is int()able, so we bail otherwise norevline = line[:m.start(revgroup)] + '{REV}' + line[m.end(revgroup):] except (TypeError, ValueError): norevline = line return (path, rev, source, pegrev, norevline) class RelativeSourceError(Exception): pass def resolvesource(ui, svnroot, source): if re_scheme.search(source): return source if source.startswith('^/'): if svnroot is None: raise RelativeSourceError() return svnroot + source[1:] ui.warn(_('ignoring unsupported non-fully qualified external: %r\n' % source)) return None def parsedefinitions(ui, repo, svnroot, exts): """Return (targetdir, revision, source) tuples. Fail if nested targetdirs are detected. source is an svn project URL. """ defs = [] for base in sorted(exts): for line in exts[base]: if not line.strip() or line.lstrip().startswith('#'): # Ignore comments and blank lines continue try: path, rev, source, pegrev, norevline = parsedefinition(line) except BadDefinition: ui.warn(_('ignoring invalid external definition: %r\n' % line)) continue source = resolvesource(ui, svnroot, source) if source is None: continue wpath = hgutil.pconvert(os.path.join(base, path)) wpath = canonpath(repo.root, '', wpath) defs.append((wpath, rev, source, pegrev, norevline, base)) # Check target dirs are not nested defs.sort() for i, d in enumerate(defs): for d2 in defs[i+1:]: if d2[0].startswith(d[0] + '/'): raise hgutil.Abort(_('external directories cannot nest:\n%s\n%s') % (d[0], d2[0])) return defs def computeactions(ui, repo, svnroot, ext1, ext2): def listdefs(data): defs = {} exts = externalsfile() exts.read(data) for d in parsedefinitions(ui, repo, svnroot, exts): defs[d[0]] = d return defs ext1 = listdefs(ext1) ext2 = listdefs(ext2) for wp1 in ext1: if wp1 in ext2: yield 'u', ext2[wp1] else: yield 'd', ext1[wp1] for wp2 in ext2: if wp2 not in ext1: yield 'u', ext2[wp2] def getsvninfo(svnurl): """Return a tuple (url, root) for supplied svn URL or working directory path. """ # Yes, this is ugly, but good enough for now args = ['svn', 'info', '--xml', svnurl] shell = os.name == 'nt' p = subprocess.Popen(args, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) stdout = p.communicate()[0] if p.returncode: raise hgutil.Abort(_('cannot get information about %s') % svnurl) m = re.search(r'<root>(.*)</root>', stdout, re.S) if not m: raise hgutil.Abort(_('cannot find SVN repository root from %s') % svnurl) root = m.group(1).rstrip('/') m = re.search(r'<url>(.*)</url>', stdout, re.S) if not m: raise hgutil.Abort(_('cannot find SVN repository URL from %s') % svnurl) url = m.group(1) m = re.search(r'<entry[^>]+revision="([^"]+)"', stdout, re.S) if not m: raise hgutil.Abort(_('cannot find SVN revision from %s') % svnurl) rev = m.group(1) return url, root, rev class externalsupdater: def __init__(self, ui, repo): self.repo = repo self.ui = ui def update(self, wpath, rev, source, pegrev): path = self.repo.wjoin(wpath) revspec = [] if rev: revspec = ['-r', rev] if os.path.isdir(path): exturl, _extroot, extrev = getsvninfo(path) # Comparing the source paths is not enough, but I don't # know how to compare path+pegrev. The following update # might fail if the path was replaced by another unrelated # one. It can be fixed manually by deleting the externals # and updating again. if source == exturl: if extrev != rev: self.ui.status(_('updating external on %s@%s\n') % (wpath, rev or 'HEAD')) cwd = os.path.join(self.repo.root, path) self.svn(['update'] + revspec, cwd) return self.delete(wpath) cwd, dest = os.path.split(path) cwd = os.path.join(self.repo.root, cwd) if not os.path.isdir(cwd): os.makedirs(cwd) if not pegrev and rev: pegrev = rev if pegrev: source = '%s@%s' % (source, pegrev) self.ui.status(_('fetching external %s@%s\n') % (wpath, rev or 'HEAD')) self.svn(['co'] + revspec + [source, dest], cwd) def delete(self, wpath): path = self.repo.wjoin(wpath) if os.path.isdir(path): self.ui.status(_('removing external %s\n') % wpath) def onerror(function, path, excinfo): if function is not os.remove: raise # read-only files cannot be unlinked under Windows s = os.stat(path) if (s.st_mode & stat.S_IWRITE) != 0: raise os.chmod(path, stat.S_IMODE(s.st_mode) | stat.S_IWRITE) os.remove(path) shutil.rmtree(path, onerror=onerror) return 1 def svn(self, args, cwd): args = ['svn'] + args self.ui.debug(_('updating externals: %r, cwd=%s\n') % (args, cwd)) shell = os.name == 'nt' p = subprocess.Popen(args, cwd=cwd, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) for line in p.stdout: self.ui.note(line) p.wait() if p.returncode != 0: raise hgutil.Abort("subprocess '%s' failed" % ' '.join(args)) def updateexternals(ui, args, repo, **opts): """update repository externals """ if len(args) > 2: raise hgutil.Abort(_('updateexternals expects at most one changeset')) node = None if len(args) == 2: svnurl = util.normalize_url(repo.ui.expandpath(args[0])) args = args[1:] else: svnurl = util.normalize_url(repo.ui.expandpath('default')) if args: node = args[0] svnroot = getsvninfo(svnurl)[1] # Retrieve current externals status try: oldext = file(repo.join('svn/externals'), 'rb').read() except IOError: oldext = '' newext = '' ctx = repo[node] if '.hgsvnexternals' in ctx: newext = ctx['.hgsvnexternals'].data() updater = externalsupdater(ui, repo) actions = computeactions(ui, repo, svnroot, oldext, newext) for action, ext in actions: if action == 'u': updater.update(ext[0], ext[1], ext[2], ext[3]) elif action == 'd': updater.delete(ext[0]) else: raise hgutil.Abort(_('unknown update actions: %r') % action) file(repo.join('svn/externals'), 'wb').write(newext) def getchanges(ui, repo, parentctx, exts): """Take a parent changectx and the new externals definitions as an externalsfile and return a dictionary mapping the special file hgsubversion needs for externals bookkeeping, to their new content as raw bytes or None if the file has to be removed. """ mode = ui.config('hgsubversion', 'externals', 'svnexternals') if mode == 'svnexternals': files = { '.hgsvnexternals': None, } if exts: files['.hgsvnexternals'] = exts.write() elif mode == 'subrepos': # XXX: clobering the subrepos files is good enough for now files = { '.hgsub': None, '.hgsubstate': None, } if exts: defs = parsedefinitions(ui, repo, '', exts) hgsub, hgsubstate = [], [] for path, rev, _source, _pegrev, norevline, base in sorted(defs): hgsub.append('%s = [hgsubversion] %s:%s\n' % (path, base, norevline)) if rev is None: rev = 'HEAD' hgsubstate.append('%s %s\n' % (rev, path)) files['.hgsub'] = ''.join(hgsub) files['.hgsubstate'] = ''.join(hgsubstate) elif mode == 'ignore': files = {} else: raise hgutil.Abort(_('unknown externals modes: %s') % mode) # Should the really be updated? updates = {} for fn, data in files.iteritems(): if data is not None: if fn not in parentctx or parentctx[fn].data() != data: updates[fn] = data else: if fn in parentctx: updates[fn] = None return updates def parse(ui, ctx): """Return the externals definitions stored in ctx as a (possibly empty) externalsfile(). """ external = externalsfile() mode = ui.config('hgsubversion', 'externals', 'svnexternals') if mode == 'svnexternals': if '.hgsvnexternals' in ctx: external.read(ctx['.hgsvnexternals'].data()) elif mode == 'subrepos': for path in ctx.substate: src, rev = ctx.substate[path][:2] base, norevline = src.split(':', 1) base = base.strip() if rev is None: rev = 'HEAD' line = norevline.replace('{REV}', rev) external.setdefault(base, []).append(line) elif mode == 'ignore': pass else: raise hgutil.Abort(_('unknown externals modes: %s') % mode) return external if subrepo: class svnsubrepo(subrepo.svnsubrepo): def __init__(self, ctx, path, state): state = (state[0].split(':', 1)[1], state[1]) super(svnsubrepo, self).__init__(ctx, path, state) def get(self, state, *args, **kwargs): # Resolve source first line = state[0].split(':', 1)[1] source, pegrev = parsedefinition(line)[2:4] try: # Getting the root SVN repository URL is expensive. # Assume the externals is absolute. source = resolvesource(self._ui, None, source) except RelativeSourceError: svnurl = self._ctx._repo.ui.expandpath('default') svnroot = getsvninfo(util.normalize_url(svnurl))[1] source = resolvesource(self._ui, svnroot, source) # hg < 1.9 svnsubrepo calls "svn checkout" with --rev # only, so peg revisions are correctly used. 1.9 and # higher, append the rev as a peg revision to the source # URL, so we cannot add our own. We assume that "-r10 # url@2" will be similar to "url@10" most of the time. if pegrev is not None and passpegrev: source = source + '@' + pegrev state = (source, state[1]) # hg-1.7.4-c19b9282d3a7 introduced the overwrite argument return super(svnsubrepo, self).get(state, *args, **kwargs) def dirty(self, ignoreupdate=False): # You cannot compare anything with HEAD. Just accept it # can be anything. if hasattr(self, '_wcrevs'): wcrevs = self._wcrevs() else: wcrev = self._wcrev() wcrevs = (wcrev, wcrev) if (('HEAD' in wcrevs or self._state[1] == 'HEAD' or self._state[1] in wcrevs or ignoreupdate) and not self._wcchanged()[0]): return False return True def commit(self, text, user, date): rev = super(svnsubrepo, self).commit(text, user, date) # Keep unversioned externals unversioned if self._state[1] == 'HEAD': rev = 'HEAD' return rev def basestate(self): # basestate() was introduced by bcb973abcc0b in 2.2 if self._state[1] == 'HEAD': return 'HEAD' return super(svnsubrepo, self).basestate()