import os
import logging
import subprocess
import warnings
try: # Python 2
from configparser import ConfigParser
from configparser import NoOptionError
from configparser import NoSectionError
except ImportError: # Python 3
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.
"""
if isinstance(revstr, bytes):
revstr = revstr.decode()
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])
