import gbp.command_wrappers as gbpc
import gbp.log
-from gbp.pkg import UpstreamSource
+from gbp.pkg import parse_archive_filename
from gbp.errors import GbpError
from gbp.deb.upstreamsource import DebianUpstreamSource
os.unlink(target)
return DebianUpstreamSource(target)
+
+
+def prepare_pristine_tar(source, pkg_name, pkg_version, pristine_commit_name,
+ filters=None, prefix=None, tmpdir=None):
+ """
+ Prepare the upstream sources for pristine-tar import
+
+ @param source: original upstream sources
+ @type source: C{UpstreamSource}
+ @param pkg_name: package name
+ @type pkg_name: C{str}
+ @param pkg_version: upstream version of the package
+ @type pkg_version: C{str}
+ @param pristine_commit_name: archive filename to commit to pristine-tar
+ @type pristine_commit_name: C{str} or C{None}
+ @param filters: filter to exclude files
+ @type filters: C{list} of C{str} or C{None}
+ @param prefix: prefix (i.e. leading directory of files) to use in
+ pristine-tar, set to C{None} to not mangle orig archive
+ @type prefix: C{str} or C{None}
+ @param tmpdir: temporary working dir (cleanup left to caller)
+ @type tmpdir: C{str}
+ @return: prepared source archive
+ @rtype: C{UpstreamSource}
+ """
+ need_repack = False
+ if source.is_dir():
+ if prefix is None:
+ prefix = '%s-%s' % (pkg_name, pkg_version)
+ gbp.log.info("Using guessed prefix '%s/' for pristine-tar" % prefix)
+ need_repack = True
+ else:
+ if prefix is not None and prefix == source.prefix:
+ prefix = None
+ comp = parse_archive_filename(pristine_commit_name)[2]
+ if filters or prefix is not None or source.compression != comp:
+ if not source.unpacked:
+ unpack_dir = tempfile.mkdtemp(prefix='pristine_unpack_',
+ dir=tmpdir)
+ source.unpack(unpack_dir)
+ need_repack = True
+ pristine_path = os.path.join(tmpdir, pristine_commit_name)
+ if need_repack:
+ gbp.log.debug("Packing '%s' from '%s' for pristine-tar" %
+ (pristine_path, source.unpacked))
+ pristine = source.pack(pristine_path, filters, prefix)
+ else:
+ # Just create symlink for mangling the pristine tarball name
+ os.symlink(source.path, pristine_path)
+ pristine = source.__class__(pristine_path)
+
+ return pristine
+
+
+def prepare_sources(source, pkg_name, pkg_version, pristine_commit_name,
+ filters, filter_pristine, prefix, tmpdir):
+ """
+ Prepare upstream sources for importing
+
+ Unpack, filter and repack sources for importing to git and to pristine-tar.
+
+ @param source: original upstream sources
+ @type source: C{UpstreamSource}
+ @param pkg_name: package name
+ @type pkg_name: C{str}
+ @param pkg_version: upstream version of the package
+ @type pkg_version: C{str}
+ @param pristine_commit_name: archive filename to commit to pristine-tar
+ @type pristine_commit_name: C{str} or C{None}
+ @param filters: filter to exclude files
+ @type filters: C{list} of C{str}
+ @param filter_pristine: filter pristine-tar, too
+ @type filter_pristine: C{bool}
+ @param prefix: prefix (i.e. leading directory of files) to use in
+ pristine-tar, set to C{None} to not mangle orig archive
+ @type prefix: C{str} or C{None}
+ @param tmpdir: temporary working dir (cleanup left to caller)
+ @type tmpdir: C{str}
+ @return: path to prepared source tree and tarball to commit to pristine-tar
+ @rtype: C{tuple} of C{str}
+ """
+ pristine = None
+ # Determine parameters for pristine tar
+ pristine_filters = filters if filters and filter_pristine else None
+ pristine_prefix = None
+ if prefix is not None and prefix != 'auto':
+ prefix_subst = {'name': pkg_name,
+ 'version': pkg_version,
+ 'upstreamversion': pkg_version}
+ pristine_prefix = prefix % prefix_subst
+ # Handle unpacked sources, i.e. importing a directory
+ if source.is_dir():
+ if pristine_commit_name:
+ gbp.log.warn('Preparing unpacked sources for pristine-tar')
+ pristine = prepare_pristine_tar(source, pkg_name, pkg_version,
+ pristine_commit_name,
+ pristine_filters, pristine_prefix,
+ tmpdir)
+ if filters:
+ # Re-use sources packed for pristine-tar, if available
+ if pristine:
+ packed = pristine
+ else:
+ packed_fn = tempfile.mkstemp(prefix="packed_", dir=tmpdir,
+ suffix='.tar')[1]
+ gbp.log.debug("Packing '%s' to '%s'" % (source.path, packed_fn))
+ packed = source.pack(packed_fn)
+ unpack_dir = tempfile.mkdtemp(prefix='filtered_', dir=tmpdir)
+ filtered = packed.unpack(unpack_dir, filters)
+ else:
+ filtered = source
+ # Handle source archives
+ else:
+ unpack_dir = tempfile.mkdtemp(prefix='filtered_', dir=tmpdir)
+ gbp.log.debug("Unpacking '%s' to '%s'" % (source.path, unpack_dir))
+ filtered = source.unpack(unpack_dir, filters)
+ if pristine_commit_name:
+ pristine = prepare_pristine_tar(source, pkg_name, pkg_version,
+ pristine_commit_name,
+ pristine_filters, pristine_prefix,
+ tmpdir)
+ pristine_path = pristine.path if pristine else ''
+ return (filtered.unpacked, pristine_path)
+
"""Test import-orig functions"""
from . import context
+import glob
import os
import tarfile
+import tempfile
import unittest
from gbp.errors import GbpError
+from gbp.pkg import UpstreamSource
+from gbp.scripts.common.import_orig import prepare_sources
from gbp.scripts.import_orig import find_source
+from tests.testutils import ls_dir, ls_tar
class TestImportOrigBase(unittest.TestCase):
tarfile.open(tar_fn, 'w' ).close()
self.assertEqual(os.path.abspath(tar_fn),
find_source(False, [tar_fn]).path)
+
+
+class TestPrepareSources(TestImportOrigBase):
+ """Test the prepare_sources() function"""
+ test_pkg_name = 'test'
+ test_pkg_ver = '1.0'
+
+ @staticmethod
+ def _create_test_sources(destdir):
+ """Create dummy source archives"""
+ destdir = os.path.abspath(destdir)
+ origs = {}
+
+ # "Normall" gzipped tarball
+ archive_fn = os.path.join(destdir, 'test-1.0.tar.gz')
+ src_dir = os.path.join(context.projectdir, 'gbp')
+ tarobj = tarfile.open(archive_fn, mode='w:gz')
+ for fname in (glob.glob('%s/*.py' % src_dir) +
+ glob.glob('%s/pkg/*.py' % src_dir)):
+ arcname = 'test-1.0/' + os.path.relpath(fname, src_dir)
+ tarobj.add(fname, arcname=arcname)
+ tarobj.close()
+ origs['tar'] = archive_fn
+
+ # Unpacked sources
+ tarobj = tarfile.open(origs['tar'], 'r')
+ tarobj.extractall(destdir)
+ tarobj.close()
+ origs['dir'] = os.path.join(destdir,'test-1.0')
+ return origs
+
+ @classmethod
+ def setup_class(cls):
+ """Class set-up, run only once"""
+ super(TestPrepareSources, cls).setup_class()
+ # Different source archives
+ cls._origs = cls._create_test_sources(cls._tmpdir)
+
+ def test_dir(self):
+ """Basic test for unpacked sources, no filtering etc"""
+ tmpdir = tempfile.mkdtemp(dir=self._tmpdir, prefix='dir_basic_')
+ source = UpstreamSource(self._origs['dir'])
+ orig, prist = prepare_sources(source, 'test', '1.0', None,
+ None, False, None, tmpdir)
+ self.assertEqual(ls_dir(self._origs['dir']), ls_dir(orig))
+ self.assertEqual(prist, '')
+
+ def test_dir_filter(self):
+ """Test filtering of unpacked sources"""
+ tmpdir = tempfile.mkdtemp(dir=self._tmpdir, prefix='dir_filter_')
+ source = UpstreamSource(self._origs['dir'])
+ orig, prist = prepare_sources(source, 'test', '1.0', None,
+ ['pkg'], False, None, tmpdir)
+ orig_filt_ref = set([fname for fname in ls_dir(self._origs['dir'])
+ if not fname.startswith('pkg')])
+ self.assertEqual(orig_filt_ref, ls_dir(orig))
+ self.assertEqual(prist, '')
+
+ def test_dir_pristine_nofilter(self):
+ """Test filtering of unpacked sources, not filtering pristine-tar"""
+ tmpdir = tempfile.mkdtemp(dir=self._tmpdir, prefix='dir_filter2_')
+ source = UpstreamSource(self._origs['dir'])
+ orig, prist = prepare_sources(source, 'test', '1.0', 'test.tar.gz',
+ ['pkg'], False, None, tmpdir)
+ src_ls = ls_dir(self._origs['dir'])
+ orig_filt_ref = set([fname for fname in src_ls
+ if not fname.startswith('pkg')])
+ prist_ref = set(['test-1.0/%s' % fname for fname in src_ls] +
+ ['test-1.0'])
+ self.assertEqual(orig_filt_ref, ls_dir(orig))
+ self.assertEqual(prist_ref, ls_tar(prist))
+
+ def test_dir_pristine_filter(self):
+ """Test filtering pristine-tar and mangling prefix"""
+ tmpdir = tempfile.mkdtemp(dir=self._tmpdir, prefix='dir_filter3_')
+ source = UpstreamSource(self._origs['dir'])
+ orig, prist = prepare_sources(source, 'test', '1.0', 'test.tar.gz',
+ ['pkg'], True, 'newpref', tmpdir)
+ src_ls = ls_dir(self._origs['dir'])
+ orig_filt_ref = set([fname for fname in src_ls
+ if not fname.startswith('pkg')])
+ prist_ref = set(['newpref/%s' % fname for fname in orig_filt_ref] +
+ ['newpref'])
+ self.assertEqual(orig_filt_ref, ls_dir(orig))
+ self.assertEqual(prist_ref, ls_tar(prist))
+
+ def test_tar(self):
+ """Basic test for tarball sources, with pristine-tar"""
+ tmpdir = tempfile.mkdtemp(dir=self._tmpdir, prefix='tar_basic_')
+ source = UpstreamSource(self._origs['tar'])
+ orig, prist = prepare_sources(source, 'test', '1.0', 'test.tgz',
+ None, False, 'test-1.0', tmpdir)
+ src_ls = ls_tar(self._origs['tar'])
+ orig_ref = set([fname.replace('test-1.0/', '') for fname in src_ls
+ if fname != 'test-1.0'])
+ self.assertEqual(orig_ref, ls_dir(orig))
+ self.assertEqual(src_ls, ls_tar(prist))
+
+ def test_tar_pristine_prefix(self):
+ """Test tarball import with prefix mangling"""
+ tmpdir = tempfile.mkdtemp(dir=self._tmpdir, prefix='tar_prefix_')
+ source = UpstreamSource(self._origs['tar'])
+ _orig, prist = prepare_sources(source, 'test', '1.0', 'test.tgz',
+ None, False, 'np', tmpdir)
+ src_ls = ls_tar(self._origs['tar'])
+ prist_ref = set([fname.replace('test-1.0', 'np') for fname in src_ls])
+ self.assertEqual(prist_ref, ls_tar(prist))
+
+ def test_tar_filter_pristine_prefix(self):
+ """Filter tarball, pristine-tar prefix mangling but not filter"""
+ tmpdir = tempfile.mkdtemp(dir=self._tmpdir, prefix='tar_filter_')
+ source = UpstreamSource(self._origs['tar'])
+ orig, prist = prepare_sources(source, 'test', '1.0', 'test.tgz',
+ ['pkg'], False, 'newp', tmpdir)
+ src_ls = ls_tar(self._origs['tar'])
+ orig_ref = set([fname.replace('test-1.0/', '') for fname in src_ls
+ if fname != 'test-1.0' and not fname.startswith('test-1.0/pkg')])
+ prist_ref = set([fname.replace('test-1.0', 'newp') for fname in src_ls])
+ self.assertEqual(orig_ref, ls_dir(orig))
+ self.assertEqual(prist_ref, ls_tar(prist))
+