Escape '%' when writing to hgrc (issue1199)

Fixes clone with paths containing '%'.
Patch provided by mjc.

mercurial/hg.py

@@ -1,312 +1,313 b''
 # hg.py - repository classes for mercurial
 #
 # Copyright 2005-2007 Matt Mackall <[e-mail]>
 # Copyright 2006 Vadim Gelfer <[e-mail]>
 #
 # This software may be used and distributed according to the terms
 # of the GNU General Public License, incorporated herein by reference.
 
 from i18n import _
 import localrepo, bundlerepo, httprepo, sshrepo, statichttprepo
 import errno, lock, os, shutil, util, extensions
 import merge as _merge
 import verify as _verify
 
 def _local(path):
     return (os.path.isfile(util.drop_scheme('file', path)) and
             bundlerepo or localrepo)
 
 def parseurl(url, revs):
     '''parse url#branch, returning url, branch + revs'''
 
     if '#' not in url:
         return url, (revs or None), None
 
     url, rev = url.split('#', 1)
     return url, revs + [rev], rev
 
 schemes = {
     'bundle': bundlerepo,
     'file': _local,
     'http': httprepo,
     'https': httprepo,
     'ssh': sshrepo,
     'static-http': statichttprepo,
 }
 
 def _lookup(path):
     scheme = 'file'
     if path:
         c = path.find(':')
         if c > 0:
             scheme = path[:c]
     thing = schemes.get(scheme) or schemes['file']
     try:
         return thing(path)
     except TypeError:
         return thing
 
 def islocal(repo):
     '''return true if repo or path is local'''
     if isinstance(repo, str):
         try:
             return _lookup(repo).islocal(repo)
         except AttributeError:
             return False
     return repo.local()
 
 def repository(ui, path='', create=False):
     """return a repository object for the specified path"""
     repo = _lookup(path).instance(ui, path, create)
     ui = getattr(repo, "ui", ui)
     for name, module in extensions.extensions():
         hook = getattr(module, 'reposetup', None)
         if hook:
             hook(ui, repo)
     return repo
 
 def defaultdest(source):
     '''return default destination of clone if none is given'''
     return os.path.basename(os.path.normpath(source))
 
 def clone(ui, source, dest=None, pull=False, rev=None, update=True,
           stream=False):
     """Make a copy of an existing repository.
 
     Create a copy of an existing repository in a new directory.  The
     source and destination are URLs, as passed to the repository
     function.  Returns a pair of repository objects, the source and
     newly created destination.
 
     The location of the source is added to the new repository's
     .hg/hgrc file, as the default to be used for future pulls and
     pushes.
 
     If an exception is raised, the partly cloned/updated destination
     repository will be deleted.
 
     Arguments:
 
     source: repository object or URL
 
     dest: URL of destination repository to create (defaults to base
     name of source repository)
 
     pull: always pull from source repository, even in local case
 
     stream: stream raw data uncompressed from repository (fast over
     LAN, slow over WAN)
 
     rev: revision to clone up to (implies pull=True)
 
     update: update working directory after clone completes, if
     destination is local repository
     """
 
     if isinstance(source, str):
         origsource = ui.expandpath(source)
         source, rev, checkout = parseurl(origsource, rev)
         src_repo = repository(ui, source)
     else:
         src_repo = source
         origsource = source = src_repo.url()
         checkout = None
 
     if dest is None:
         dest = defaultdest(source)
         ui.status(_("destination directory: %s\n") % dest)
 
     def localpath(path):
         if path.startswith('file://localhost/'):
             return path[16:]
         if path.startswith('file://'):
             return path[7:]
         if path.startswith('file:'):
             return path[5:]
         return path
 
     dest = localpath(dest)
     source = localpath(source)
 
     if os.path.exists(dest):
         raise util.Abort(_("destination '%s' already exists") % dest)
 
     class DirCleanup(object):
         def __init__(self, dir_):
             self.rmtree = shutil.rmtree
             self.dir_ = dir_
         def close(self):
             self.dir_ = None
         def __del__(self):
             if self.dir_:
                 self.rmtree(self.dir_, True)
 
     src_lock = dest_lock = dir_cleanup = None
     try:
         if islocal(dest):
             dir_cleanup = DirCleanup(dest)
 
         abspath = origsource
         copy = False
         if src_repo.cancopy() and islocal(dest):
             abspath = os.path.abspath(util.drop_scheme('file', origsource))
             copy = not pull and not rev
 
         if copy:
             try:
                 # we use a lock here because if we race with commit, we
                 # can end up with extra data in the cloned revlogs that's
                 # not pointed to by changesets, thus causing verify to
                 # fail
                 src_lock = src_repo.lock()
             except lock.LockException:
                 copy = False
 
         if copy:
             def force_copy(src, dst):
                 if not os.path.exists(src):
                     # Tolerate empty source repository and optional files
                     return
                 util.copyfiles(src, dst)
 
             src_store = os.path.realpath(src_repo.spath)
             if not os.path.exists(dest):
                 os.mkdir(dest)
             try:
                 dest_path = os.path.realpath(os.path.join(dest, ".hg"))
                 os.mkdir(dest_path)
             except OSError, inst:
                 if inst.errno == errno.EEXIST:
                     dir_cleanup.close()
                     raise util.Abort(_("destination '%s' already exists")
                                      % dest)
                 raise
             if src_repo.spath != src_repo.path:
                 # XXX racy
                 dummy_changelog = os.path.join(dest_path, "00changelog.i")
                 # copy the dummy changelog
                 force_copy(src_repo.join("00changelog.i"), dummy_changelog)
                 dest_store = os.path.join(dest_path, "store")
                 os.mkdir(dest_store)
             else:
                 dest_store = dest_path
             # copy the requires file
             force_copy(src_repo.join("requires"),
                        os.path.join(dest_path, "requires"))
             # we lock here to avoid premature writing to the target
             dest_lock = lock.lock(os.path.join(dest_store, "lock"))
 
             files = ("data",
                      "00manifest.d", "00manifest.i",
                      "00changelog.d", "00changelog.i")
             for f in files:
                 src = os.path.join(src_store, f)
                 dst = os.path.join(dest_store, f)
                 force_copy(src, dst)
 
             # we need to re-init the repo after manually copying the data
             # into it
             dest_repo = repository(ui, dest)
 
         else:
             try:
                 dest_repo = repository(ui, dest, create=True)
             except OSError, inst:
                 if inst.errno == errno.EEXIST:
                     dir_cleanup.close()
                     raise util.Abort(_("destination '%s' already exists")
                                      % dest)
                 raise
 
             revs = None
             if rev:
                 if 'lookup' not in src_repo.capabilities:
                     raise util.Abort(_("src repository does not support revision "
                                        "lookup and so doesn't support clone by "
                                        "revision"))
                 revs = [src_repo.lookup(r) for r in rev]
 
             if dest_repo.local():
                 dest_repo.clone(src_repo, heads=revs, stream=stream)
             elif src_repo.local():
                 src_repo.push(dest_repo, revs=revs)
             else:
                 raise util.Abort(_("clone from remote to remote not supported"))
 
         if dir_cleanup:
             dir_cleanup.close()
 
         if dest_repo.local():
             fp = dest_repo.opener("hgrc", "w", text=True)
             fp.write("[paths]\n")
-            fp.write("default = %s\n" % abspath)
+            # percent needs to be escaped for ConfigParser
+            fp.write("default = %s\n" % abspath.replace('%', '%%'))
             fp.close()
 
             if update:
                 dest_repo.ui.status(_("updating working directory\n"))
                 if not checkout:
                     try:
                         checkout = dest_repo.lookup("default")
                     except:
                         checkout = dest_repo.changelog.tip()
                 _update(dest_repo, checkout)
 
         return src_repo, dest_repo
     finally:
         del src_lock, dest_lock, dir_cleanup
 
 def _showstats(repo, stats):
     stats = ((stats[0], _("updated")),
              (stats[1], _("merged")),
              (stats[2], _("removed")),
              (stats[3], _("unresolved")))
     note = ", ".join([_("%d files %s") % s for s in stats])
     repo.ui.status("%s\n" % note)
 
 def _update(repo, node): return update(repo, node)
 
 def update(repo, node):
     """update the working directory to node, merging linear changes"""
     pl = repo.parents()
     stats = _merge.update(repo, node, False, False, None)
     _showstats(repo, stats)
     if stats[3]:
         repo.ui.status(_("There are unresolved merges with"
                          " locally modified files.\n"))
         if stats[1]:
             repo.ui.status(_("You can finish the partial merge using:\n"))
         else:
             repo.ui.status(_("You can redo the full merge using:\n"))
         # len(pl)==1, otherwise _merge.update() would have raised util.Abort:
         repo.ui.status(_("  hg update %s\n  hg update %s\n")
                        % (pl[0].rev(), repo.changectx(node).rev()))
     return stats[3] > 0
 
 def clean(repo, node, show_stats=True):
     """forcibly switch the working directory to node, clobbering changes"""
     stats = _merge.update(repo, node, False, True, None)
     if show_stats: _showstats(repo, stats)
     return stats[3] > 0
 
 def merge(repo, node, force=None, remind=True):
     """branch merge with node, resolving changes"""
     stats = _merge.update(repo, node, True, force, False)
     _showstats(repo, stats)
     if stats[3]:
         pl = repo.parents()
         repo.ui.status(_("There are unresolved merges,"
                          " you can redo the full merge using:\n"
                          "  hg update -C %s\n"
                          "  hg merge %s\n")
                        % (pl[0].rev(), pl[1].rev()))
     elif remind:
         repo.ui.status(_("(branch merge, don't forget to commit)\n"))
     return stats[3] > 0
 
 def revert(repo, node, choose):
     """revert changes to revision in node without updating dirstate"""
     return _merge.update(repo, node, False, True, choose)[3] > 0
 
 def verify(repo):
     """verify the consistency of a repository"""
     return _verify.verify(repo)

tests/test-hgrc

@@ -1,7 +1,18 b''
 #!/bin/sh
 
 mkdir t
 cd t
 hg init
 echo "invalid" > .hg/hgrc
 hg status 2>&1 |sed -e "s:/.*\(/t/.*\):...\1:"
+
+#issue 1199, escaping
+
+cd ..
+hg init "foo%bar"
+hg clone "foo%bar" foobar
+p=`pwd`
+cd foobar
+cat .hg/hgrc |sed -e "s:$p:...:"
+hg paths |sed -e "s:$p:...:"
+hg showconfig |sed -e "s:$p:...:"

tests/test-hgrc.out

@@ -1,4 +1,16 b''
 abort: Failed to parse .../t/.hg/hgrc
 File contains no section headers.
 file: .../t/.hg/hgrc, line: 1
 'invalid\n'
+updating working directory
+0 files updated, 0 files merged, 0 files removed, 0 files unresolved
+[paths]
+default = .../foo%%bar
+default = .../foo%bar
+bundle.mainreporoot=.../foobar
+defaults.backout=-d "0 0"
+defaults.commit=-d "0 0"
+defaults.debugrawcommit=-d "0 0"
+defaults.tag=-d "0 0"
+paths.default=.../foo%bar
+ui.slash=True