import errno
import os
import re
import subprocess
import sys
from gitosis import util
class GitError(Exception):
"""git failed"""
def __str__(self):
return '%s: %s' % (self.__doc__, ': '.join(self.args))
class GitInitError(Exception):
"""git init failed"""
def init(
path,
template=None,
_git=None,
):
"""
Create a git repository at C{path} (if missing).
Leading directories of C{path} must exist.
@param path: Path of repository create.
@type path: str
@param template: Template directory, to pass to C{git init}.
@type template: str
"""
if _git is None:
_git = 'git'
util.mkdir(path, 0750)
args = [
_git,
'--git-dir=.',
'init',
]
if template is not None:
args.append('--template=%s' % template)
returncode = subprocess.call(
args=args,
cwd=path,
stdout=sys.stderr,
close_fds=True,
)
if returncode != 0:
raise GitInitError('exit status %d' % returncode)
class GitFastImportError(GitError):
"""git fast-import failed"""
pass
def fast_import(
git_dir,
commit_msg,
committer,
files,
parent=None,
):
"""
Create an initial commit.
"""
child = subprocess.Popen(
args=[
'git',
'--git-dir=.',
'fast-import',
'--quiet',
'--date-format=now',
],
cwd=git_dir,
stdin=subprocess.PIPE,
close_fds=True,
)
files = list(files)
for index, (path, content) in enumerate(files):
child.stdin.write("""\
blob
mark :%(mark)d
data %(len)d
%(content)s
""" % dict(
mark=index+1,
len=len(content),
content=content,
))
child.stdin.write("""\
commit refs/heads/master
committer %(committer)s now
data %(commit_msg_len)d
%(commit_msg)s
""" % dict(
committer=committer,
commit_msg_len=len(commit_msg),
commit_msg=commit_msg,
))
if parent is not None:
assert not parent.startswith(':')
child.stdin.write("""\
from %(parent)s
""" % dict(
parent=parent,
))
for index, (path, content) in enumerate(files):
child.stdin.write('M 100644 :%d %s\n' % (index+1, path))
child.stdin.close()
returncode = child.wait()
if returncode != 0:
raise GitFastImportError(
'git fast-import failed', 'exit status %d' % returncode)
class GitExportError(GitError):
"""Export failed"""
pass
class GitReadTreeError(GitExportError):
"""git read-tree failed"""
class GitCheckoutIndexError(GitExportError):
"""git checkout-index failed"""
def export(git_dir, path):
try:
os.mkdir(path)
except OSError, e:
if e.errno == errno.EEXIST:
pass
else:
raise
returncode = subprocess.call(
args=[
'git',
'--git-dir=%s' % git_dir,
'read-tree',
'HEAD',
],
close_fds=True,
)
if returncode != 0:
raise GitReadTreeError('exit status %d' % returncode)
env = {}
env.update(os.environ)
env['GIT_WORK_TREE'] = '.'
returncode = subprocess.call(
args=[
'git',
'--git-dir=%s' % os.path.abspath(git_dir),
'checkout-index',
'-a',
'-f',
],
cwd=path,
close_fds=True,
env=env,
)
if returncode != 0:
raise GitCheckoutIndexError('exit status %d' % returncode)
class GitHasInitialCommitError(GitError):
"""Check for initial commit failed"""
class GitRevParseError(GitError):
"""rev-parse failed"""
def has_initial_commit(git_dir):
child = subprocess.Popen(
args=[
'git',
'--git-dir=.',
'rev-parse',
'HEAD',
],
cwd=git_dir,
stdout=subprocess.PIPE,
close_fds=True,
)
got = child.stdout.read()
returncode = child.wait()
if returncode != 0:
raise GitRevParseError('exit status %d' % returncode)
if got == 'HEAD\n':
return False
elif re.match('^[0-9a-f]{40}\n$', got):
return True
else:
raise GitHasInitialCommitError('Unknown git HEAD: %r' % got)
