"""Git fast import class"""
import subprocess
+import time
from gbp.errors import GbpError
class FastImport(object):
- """Invoke git-fast-import"""
+ """Add data to a git repository using I{git fast-import}"""
_bufsize = 1024
m_regular = 644
m_exec = 755
m_symlink = 120000
- def __init__(self):
+ def __init__(self, repo):
+ """
+ @param repo: the git repository L{FastImport} acts on
+ @type repo: L{GitRepository}
+ """
+ self._repo = repo
try:
- self._fi = subprocess.Popen([ 'git', 'fast-import', '--quiet'], stdin=subprocess.PIPE)
+ self._fi = subprocess.Popen([ 'git', 'fast-import', '--quiet'],
+ stdin=subprocess.PIPE, cwd=repo.path)
self._out = self._fi.stdin
except OSError as err:
raise GbpError("Error spawning git fast-import: %s" % err)
except ValueError as err:
- raise GbpError("Invalid argument when spawning git fast-import: %s" % err)
+ raise GbpError(
+ "Invalid argument when spawning git fast-import: %s" % err)
def _do_data(self, fd, size):
self._out.write("data %s\n" % size)
self._out.write("M %d inline %s\n" % (mode, name))
self._do_data(fd, size)
- def add_file(self, filename, fd, size):
- self._do_file(filename, self.m_regular, fd, size)
-
- def add_executable(self, filename, fd, size):
- self._do_file(filename, self.m_exec, fd, size)
+ def add_file(self, filename, fd, size, mode=m_regular):
+ """
+ Add a file
+
+ @param filename: the name of the file to add
+ @type filename: C{str}
+ @param fd: stream to read data from
+ @type fd: C{File} like object
+ @param size: size of the file to add
+ @type size: C{int}
+ @param mode: file mode, default is L{FastImport.m_regular}.
+ @type mode: C{int}
+ """
+ self._do_file(filename, mode, fd, size)
+
+ def add_symlink(self, linkname, linktarget):
+ """
+ Add a symlink
+
+ @param linkname: the symbolic link's name
+ @param linkname: C{str}
+ @param linktarget: the target the symlink points to
+ @type linktarget: C{str}
+ """
+ self._out.write("M %d inline %s\n" % (self.m_symlink, linkname))
+ self._out.write("data %s\n" % len(linktarget))
+ self._out.write("%s\n" % linktarget)
+
+ def start_commit(self, branch, committer, msg):
+ """
+ Start a fast import commit
+
+ @param branch: branch to commit on
+ @type branch: C{str}
+ @param committer: the committer information
+ @type committer: L{GitModifier}
+ @param msg: the commit message
+ @type msg: C{str}
+ """
+ length = len(msg)
+ if not committer.date:
+ committer.date = "%d %s" % (time.time(),
+ time.strftime("%z"))
- def add_symlink(self, filename, linkname):
- name = "/".join(filename.split('/')[1:])
- self._out.write("M %d inline %s\n" % (self.m_symlink, name))
- self._out.write("data %s\n" % len(linkname))
- self._out.write("%s\n" % linkname)
+ if self._repo.has_branch(branch):
+ from_ = "from refs/heads/%(branch)s^0\n"
+ else:
+ from_ = ''
- def start_commit(self, branch, committer, email, time, msg):
- length = len(msg)
self._out.write("""commit refs/heads/%(branch)s
-committer %(committer)s <%(email)s> %(time)s
+committer %(name)s <%(email)s> %(time)s
data %(length)s
-%(msg)s
-from refs/heads/%(branch)s^0
-""" % locals())
-
- def do_deleteall(self):
+%(msg)s%(from)s""" %
+ { 'branch': branch,
+ 'name': committer.name,
+ 'email': committer.email,
+ 'time': committer.date,
+ 'length': length,
+ 'msg': msg,
+ 'from': from_,
+ })
+
+ def deleteall(self):
+ """
+ Issue I{deleteall} to fastimport so we start from a empty tree
+ """
self._out.write("deleteall\n")
def close(self):
+ """
+ Close fast-import issuing all pending actions
+ """
if self._out:
self._out.close()
if self._fi:
def __del__(self):
self.close()
-
-
--- /dev/null
+# vim: set fileencoding=utf-8 :
+
+import os
+import shutil
+import tarfile
+import tempfile
+
+import gbp.log
+import gbp.git
+
+repo = None
+fastimport = None
+tmpdir = None
+tf_name = 'testfile'
+tl_name = 'a_testlink'
+
+def setup():
+ global repo, tmpdir
+
+ gbp.log.setup(False, False)
+ top = os.path.abspath(os.curdir)
+ tmpdir = os.path.join(top,'gbp_%s_repo' % __name__)
+ os.mkdir(tmpdir)
+
+ repodir = os.path.join(tmpdir, 'test_repo')
+ repo = gbp.git.GitRepository.create(repodir)
+
+def teardown():
+ if not os.getenv("GBP_TESTS_NOCLEAN") and tmpdir:
+ shutil.rmtree(tmpdir)
+
+def test_init_fastimport():
+ """Create a fastimport object"""
+ global fastimport
+ fastimport = gbp.git.FastImport(repo)
+ assert fastimport, "Failed to init FastImport"
+
+def test_add_file():
+ """Add a file via fastimport"""
+ author = repo.get_author_info()
+ fastimport.start_commit('master', author, "a commit")
+ fastimport.deleteall()
+ testfile = os.path.join(repo.path, '.git', 'description')
+ fastimport.add_file('./testfile',
+ file(testfile),
+ os.path.getsize(testfile))
+
+def test_add_symlink():
+ """Add a symbolic link via fastimport"""
+ author = repo.get_author_info()
+ fastimport.start_commit('master', author, "a 2nd commit")
+ fastimport.add_symlink(tl_name, tf_name)
+
+def test_close():
+ fastimport.close()
+
+def test_result():
+ repo.force_head('master', hard=True)
+
+ testfile = os.path.join(repo.path, tf_name)
+ testlink = os.path.join(repo.path, tl_name)
+
+ assert os.path.exists(testfile), "%s doesn't exist" % testfile
+ assert os.path.lexists(testlink), "%s doesn't exist" % testlink
+ assert os.readlink(testlink) == tf_name
+