import os
import logging
import subprocess
import warnings
from ConfigParser import ConfigParser
from ConfigParser import NoOptionError
from ConfigParser import NoSectionError
from zc.buildout import UserError
from .base import BaseRepo
from .base import SUBPROCESS_ENV
from .base import update_check_call
from ..utils import check_output
logger = logging.getLogger(__name__)
[docs]class HgRepo(BaseRepo):
vcs_control_dir = '.hg'
vcs_official_name = 'Mercurial'
[docs] def update_hgrc_paths(self):
"""Update hgrc paths section if needed.
Old paths are kept in renamed form: buildout_save_%d."""
parser = ConfigParser()
hgrc_path = os.path.join(self.target_dir, '.hg', 'hgrc')
parser.read(hgrc_path) # does not fail if file does not exist
previous = None
try:
previous = parser.get('paths', 'default')
except NoOptionError:
logger.info("No 'default' value for [paths] in %s, will set one",
hgrc_path)
except NoSectionError:
logger.info("Creating [paths] section in %s", hgrc_path)
parser.add_section('paths')
if previous == self.url:
return
if previous is not None:
count = 1
while True:
save = 'buildout_save_%d' % count
try:
parser.get('paths', save)
except NoOptionError:
break
count += 1
parser.set('paths', save, previous)
logger.info("Change of origin URL, saving previous value as %r in "
"[paths] section of %s", save, hgrc_path)
parser.set('paths', 'default', self.url)
f = open(hgrc_path, 'w')
parser.write(f)
f.close()
[docs] def uncommitted_changes(self):
"""True if we have uncommitted changes."""
return bool(check_output(['hg', '--cwd', self.target_dir, 'status'],
env=SUBPROCESS_ENV))
[docs] def parents(self, pip_compatible=False):
"""Return full hash of parent nodes.
:param pip_compatible: ignored, all Hg revspecs are pip compatible
"""
return check_output(['hg', '--cwd', self.target_dir, 'parents',
'--template={node}'],
env=SUBPROCESS_ENV).split()
[docs] def have_fixed_revision(self, revstr):
warnings.warn("have_fixed_revision() is deprecated and has been "
"renamed to is_local_fixed_revision()",
DeprecationWarning)
return self.is_local_fixed_revision(revstr)
[docs] def is_local_fixed_revision(self, revstr):
"""True if revstr is a fixed revision that we already have.
Check is done for known tags (except tip) and known nodes identified
by a long enough (12 char) prefix of their hexadecimal hash.
Summary of collision cases for hg up:
- a revision number has precedence over a tag identically named
- a tag that is a strict prefix of a full hexadecimal node hash wins
over that node.
- a full hexadecimal node hash wins over a tag that would be
identically named (someone would need to be really disturbed to do
that in real life).
- a hexadecimal node that coincides with a decimal revision number is
not something I can test :-)
In theory, a 12 char hexadecimal node hash could be shadowed by an
incoming tag. But also, any tag could be overridden. These are
considered to be fixed anyway for convenience in sensible use-cases.
People having CI robots involving tags that do get overridden by a
third party upstream should complain to upstream for utterly bad
practices.
"""
revstr = revstr.strip()
if revstr == 'tip' or not revstr:
return False
try:
out = check_output(['hg', '--cwd', self.target_dir, 'log',
'-r', revstr,
'--template={node}\n{tags}\n{rev}'],
env=SUBPROCESS_ENV)
except subprocess.CalledProcessError:
return False
node, tags, rev = out.split(os.linesep)
if node.startswith(revstr):
# if too short, can be superseded by an incoming tag
if len(revstr) >= 12:
logger.info("[hg] Found requested revision %r in %s",
revstr, self.target_dir)
return True
if revstr == rev:
logger.warn("[hg] In repo %s, you should not pinpoint revision "
"by a local revision number such as %r",
self.target_dir, revstr)
# but indeed, nothing can change it (unless one day a node has
# exactly that hash code, chances are...)
return True
if revstr in tags.split():
logger.info("[hg] In repo %s, found tag %r as %s",
self.target_dir, revstr, node)
return True
return False
[docs] def clean(self):
if not os.path.isdir(self.target_dir):
return
try:
subprocess.check_call(['hg', 'purge', '--cwd', self.target_dir])
except subprocess.CalledProcessError as exc:
if exc.returncode == 255:
# fallback to default implementation
logger.warn("The 'purge' Mercurial extension is not "
"activated. Do 'hg help purge' for more details "
"Falling back to default cleaning implementation")
super(HgRepo, self).clean()
[docs] def get_update(self, revision):
"""Ensure that target_dir is a clone of url at specified revision.
If target_dir already exists, does a simple pull.
Offline-mode: no clone nor pull, but update.
"""
target_dir = self.target_dir
url = self.url
offline = self.offline
if not os.path.exists(target_dir):
# TODO case of local url ?
if offline:
raise UserError(
"hg repository %r does not exist; "
"cannot clone it from %r (offline mode)" % (target_dir,
url))
logger.info("Cloning %s ...", url)
clone_cmd = ['hg', 'clone']
if revision:
clone_cmd.extend(['-r', revision])
clone_cmd.extend([url, target_dir])
subprocess.check_call(clone_cmd, env=SUBPROCESS_ENV)
else:
self.update_hgrc_paths()
# TODO what if remote repo is actually local fs ?
if self.is_local_fixed_revision(revision):
self._update(revision)
return
if not offline:
self._pull()
self._update(revision)
def _pull(self):
logger.info("Pull for hg repo %r ...", self.target_dir)
subprocess.check_call(['hg', '--cwd', self.target_dir, 'pull'],
env=SUBPROCESS_ENV)
def _update(self, revision):
target_dir = self.target_dir
logger.info("Updating %s to revision %s", target_dir, revision)
up_cmd = ['hg', '--cwd', target_dir, 'up']
if revision:
up_cmd.extend(['-r', revision])
update_check_call(up_cmd, env=SUBPROCESS_ENV)
[docs] def archive(self, target_path):
subprocess.check_call(['hg', '--cwd', self.target_dir,
'archive', target_path])
