/usr/lib/python2.7/dist-packages/deployer/vcs.py is in juju-deployer 0.3.6-0ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | import subprocess
import os
from bzrlib.workingtree import WorkingTree
from .utils import ErrorExit
class Vcs(object):
err_update = (
"Could not update branch %(path)s from %(branch_url)s\n\n %(output)s")
err_branch = "Could not branch %(branch_url)s to %(path)s\n\n %(output)s"
err_is_mod = "Couldn't determine if %(path)s was modified\n\n %(output)s"
err_pull = (
"Could not pull branch @ %(branch_url)s to %(path)s\n\n %(output)s")
err_cur_rev = (
"Could not determine current revision %(path)s\n\n %(output)s")
def __init__(self, path, origin, log):
self.path = path
self.log = log
self.origin = origin
def _call(self, args, error_msg, cwd=None, stderr=()):
try:
if stderr is not None and not stderr:
stderr = subprocess.STDOUT
output = subprocess.check_output(
args, cwd=cwd or self.path, stderr=stderr)
except subprocess.CalledProcessError, e:
#print "vcs err", " ".join(args), "[dir: %s]" % cwd
self.log.error(error_msg % self.get_err_msg_ctx(e))
raise ErrorExit()
return output.strip()
def get_err_msg_ctx(self, e):
return {
'path': self.path,
'branch_url': self.origin,
'exit_code': e.returncode,
'output': e.output,
'vcs': self.__class__.__name__.lower()}
def get_cur_rev(self):
raise NotImplementedError()
def update(self, rev=None):
raise NotImplementedError()
def branch(self):
raise NotImplementedError()
def pull(self):
raise NotImplementedError()
def is_modified(self):
raise NotImplementedError()
# upstream missing revisions?
class Bzr(Vcs):
def get_cur_rev(self):
params = ["bzr", "revno", "--tree"]
return self._call(params, self.err_cur_rev, stderr=None)
def update(self, rev=None):
params = ["bzr", "up"]
if rev:
params.extend(["-r", str(rev)])
self._call(params, self.err_update)
def branch(self):
params = ["bzr", "co", "--lightweight", self.origin, self.path]
cwd = os.path.dirname(os.path.dirname(self.path))
if not cwd:
cwd = "."
self._call(params, self.err_branch, cwd)
def is_modified(self):
# To replace with bzr cli, we need to be able to detect
# changes to a wc @ a rev or @ trunk.
tree = WorkingTree.open(self.path)
return tree.has_changes()
class Git(Vcs):
def get_cur_rev(self):
params = ["git", "rev-parse", "HEAD"]
return self._call(params, self.err_cur_rev)
def update(self, rev=None):
params = ["git", "reset", "--merge"]
if rev:
params.append(rev)
self._call(params, self.err_update)
def branch(self):
params = ["git", "clone", "--depth", "1", self.branch]
self._call(params, self.err_branch, os.path.dirname(self.path))
def is_modified(self):
params = ["git", "stat", "-s"]
return bool(self._call(params, self.err_is_mod).strip())
def get_origin(self):
params = ["git", "config", "--get", "remote.origin.url"]
return self._call(params, "")
|