ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity
community.general/library/hg
Yeukhon Wong e5b64b2f74 Added hg module to the core.
2013-01-26 12:20:46 -05:00

267 lines
8.9 KiB
Python
Raw Blame History

#!/usr/bin/python
#-*- coding: utf-8 -*-
# (c) 2013, Yeukhon Wong <yeukhon@acm.org>
#
# This module was originally inspired by Brad Olson's ansible-module-mercurial
# <https://github.com/bradobro/ansible-module-mercurial>.
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
import os
import shutil
import ConfigParser
from subprocess import Popen, PIPE
DOCUMENTATION = '''
---
module: hg
short_description: Manages Mercurial (hg) repositories.
description:
- Manages Mercurial (hg) repositories. Supports SSH, HTTP/S and local address.
version_added: "1.0"
author: Yeukhon Wong
options:
repo:
description:
- The repository location.
required: true
default: null
dest:
description:
- Absolute path of where the repository should be cloned to.
required: true
default: null
state:
description:
- C(hg clone) is performed when state is set to C(present). C(hg pull) and C(hg update) is
performed when state is set to C(latest). If you want the latest copy of the repository,
just rely on C(present). C(latest) assumes the repository is already on disk.
required: false
default: present
choices: [ "present", "absent", "latest" ]
revision:
description:
- Equivalent C(-r) option in hg command, which can either be a changeset number or a branch
name.
required: false
default: "default"
force:
description:
- Whether to discard uncommitted changes and remove untracked files or not. Basically, it
combines C(hg up -C) and C(hg purge).
required: false
default: "yes"
choices: [ "yes", "no" ]
examples:
- code: "hg: repo=https://bitbucket.org/user/repo_name dest=/home/user/repo_name"
description: Clone the default branch of repo_name.
- code: "hg: repo=https://bitbucket.org/user/repo_name dest=/home/user/repo_name force=yes state=latest"
description: Ensure the repository at dest is latest and discard any uncommitted and/or untracked files.
notes:
- If the task seems to be hanging, first verify remote host is in C(known_hosts).
SSH will prompt user to authorize the first contact with a remote host. One solution is to add
C(StrictHostKeyChecking no) in C(.ssh/config) which will accept and authorize the connection
on behalf of the user. However, if you run as a different user such as setting sudo to True),
for example, root will not look at the user .ssh/config setting.
requirements: [ ]
'''
class HgError(Exception):
""" Custom exception class to report hg command error. """
def __init__(self, msg, stderr=''):
self.msg = msg + \
"\n\nExtra information on this error: \n" + \
stderr
def __str__(self):
return self.msg
def _set_hgrc(hgrc, vals):
# val is a list of triple-tuple of the form [(section, option, value),...]
parser = ConfigParser.SafeConfigParser()
parser.read(hgrc)
for each in vals:
section,option, value = each
if not parser.has_section(section):
parser.add_section(section)
parser.set(section, option, value)
f = open(hgrc, 'w')
parser.write(f)
f.close()
def _undo_hgrc(hgrc, vals):
parser = ConfigParser.SafeConfigParser()
parser.read(hgrc)
for each in vals:
section, option, value = each
if parser.has_section(section):
parser.remove_option(section, option)
f = open(hgrc, 'w')
parser.write(f)
f.close()
def _hg_command(args_list):
cmd = ['hg'] + args_list
p = Popen(cmd, stdout=PIPE, stderr=PIPE)
out, err = p.communicate()
return out, err, p.returncode
def _hg_discard(dest):
out, err, code = _hg_command(['up', '-C', '-R', dest])
if code != 0:
raise HgError(err)
def _hg_purge(dest):
hgrc = os.path.join(dest, '.hg/hgrc')
purge_option = [('extensions', 'purge', '')]
_set_hgrc(hgrc, purge_option)
out, err, code = _hg_command(['purge', '-R', dest])
if code == 0:
_undo_hgrc(hgrc, purge_option)
else:
raise HgError(err)
def _hg_verify(dest):
error1 = "hg verify failed."
error2 = "{dest} is not a repository.".format(dest=dest)
out, err, code = _hg_command(['verify', '-R', dest])
if code == 1:
raise HgError(error1, stderr=err)
elif code == 255:
raise HgError(error2, stderr=err)
elif code == 0:
return True
def _post_op_hg_revision_check(dest, revision):
"""
Verify the tip is the same as `revision`.
This function is usually called after some hg operations
such as `clone`. However, this check is skipped if `revision`
is the string `default` since it will result an error.
Instead, pull is performed.
"""
err1 = "Unable to perform hg tip."
err2 = "tip is different from %s. See below for extended summary." % revision
if revision == 'default':
out, err, code = _hg_command(['pull', '-R', dest])
if "no changes found" in out:
return True
else:
raise HgError(err2, stderr=out)
else:
out, err, code = _hg_command(['tip', '-R', dest])
if revision in out: # revision should be part of the output (changeset: $revision ...)
return True
else:
if code != 0: # something went wrong with hg tip
raise HgError(err1, stderr=err)
else: # hg tip is fine, but tip != revision
raise HgError(err2, stderr=out)
def force_and_clean(dest):
_hg_discard(dest)
_hg_purge(dest)
def pull_and_update(repo, dest, revision, force):
if force == 'yes':
force_and_clean(dest)
if _hg_verify(dest):
cmd1 = ['pull', '-R', dest, '-r', revision]
out, err, code = _hg_command(cmd1)
if code == 1:
raise HgError("Unable to perform pull on %s" % dest, stderr=err)
elif code == 0:
cmd2 = ['update', '-R', dest, '-r', revision]
out, err, code = _hg_command(cmd2)
if code == 1:
raise HgError("There are unresolved files in %s" % dest, stderr=err)
elif code == 0:
# so far pull and update seems to be working, check revision and $revision are equal
_post_op_hg_revision_check(dest, revision)
return True
# when code aren't 1 or 0 in either command
raise HgError("", stderr=err)
def clone(repo, dest, revision, force):
if os.path.exists(dest):
if _hg_verify(dest): # make sure it's a real repo
if _post_op_hg_revision_check(dest, revision): # make sure revision and $revision are equal
if force == 'yes':
force_and_clean(dest)
return False
cmd = ['clone', repo, dest, '-r', revision]
out, err, code = _hg_command(cmd)
if code == 0:
_hg_verify(dest)
_post_op_hg_revision_check(dest, revision)
return True
else:
raise HgError(err, stderr='')
def main():
module = AnsibleModule(
argument_spec = dict(
repo = dict(required=True),
dest = dict(required=True),
state = dict(default='present', choices=['present', 'absent', 'latest']),
revision = dict(default="default"),
force = dict(default='yes', choices=['yes', 'no']),
),
)
repo = module.params['repo']
state = module.params['state']
dest = module.params['dest']
revision = module.params['revision']
force = module.params['force']
try:
if state == 'absent':
if not os.path.exists(dest):
shutil.rmtree(dest)
changed = True
elif state == 'present':
changed = clone(repo, dest, revision, force)
elif state == 'latest':
changed = pull_and_update(repo, dest, revision, force)
module.exit_json(dest=dest, changed=changed)
#except HgError as e:
# module.fail_json(msg=str(e), params=module.params)
#except IOError as e:
# module.fail_json(msg=str(e), params=module.params)
except Exception as e:
module.fail_json(msg=str(e), params=module.params)
# include magic from lib/ansible/module_common.py
#<<INCLUDE_ANSIBLE_MODULE_COMMON>>
main()
View git blame Copy permalink