Add tests for UpstreamSource

* tar and zipfile unpacking
* (filtered) tar packing

2 files changed, 141 insertions, 11 deletions

diff --git a/tests/02_test_import.py b/tests/02_test_import.py
index d31533f0..33628ea7 100644
--- a/tests/02_test_import.py
+++ b/tests/02_test_import.py
@@ -10,16 +10,28 @@ import gbp.deb
 
 class TestUnpack:
     """Make sure we unpack gzip and bzip2 archives correctly"""
-    def _createArchive(self, comp):
-        archive = "archive"
-        name = "%s_0.1.tar.%s" % (archive, comp)
+    archive_prefix = "archive"
+
+    def _unpack_dir(self, compression):
+        return "%s-%s" % (self.archive_prefix, compression)
+
+    def _check_files(self, files, comp):
+        """Check if files exist in the unpacked dir"""
+        for f in files:
+            target = os.path.join(self._unpack_dir(comp), f)
+            assert os.path.exists(target), "%s does not exist" % target
+
+    def _create_archive(self, comp):
+        filelist = [ os.path.basename(f) for f in
+                     glob.glob(os.path.join(self.top, "g*-*")) ]
+
+        name = "%s_0.1.tar.%s" % (self.archive_prefix, comp)
         t = tarfile.open(name= name, mode='w:%s' % comp)
-        for f in glob.glob(os.path.join(self.top, "*.py")):
-            t.add(os.path.join(self.top,f),
-                  os.path.join("%s-%s" % (archive, comp),
-                               os.path.basename(f)))
+        for f in filelist:
+            t.add(os.path.join(self.top, f),
+                  os.path.join(self._unpack_dir(comp), f))
         t.close()
-        return name
+        return name, filelist
 
     def setUp(self):
         self.dir = tempfile.mkdtemp(prefix='gbp_%s_' % __name__, dir='.')
@@ -27,15 +39,47 @@ class TestUnpack:
         os.chdir(self.dir)
         self.archives = {}
         for ext in [ "gz", "bz2" ]:
-            self.archives[ext] = self._createArchive(ext)
+            self.archives[ext] = self._create_archive(ext)
 
     def tearDown(self):
         os.chdir(self.top)
         if not os.getenv("GBP_TESTS_NOCLEAN"):
             shutil.rmtree(self.dir)
 
-    def testUnpack(self):
+    def test_unpack(self):
+        for (comp, archive) in self.archives.iteritems():
+            gbp.deb.unpack_orig(archive[0], ".", [])
+
+    def test_upstream_source_type(self):
+        for (comp, archive) in self.archives.iteritems():
+            source = gbp.deb.UpstreamSource(archive[0])
+            assert source.is_orig == True
+            assert source.is_dir == False
+            assert source.unpacked == None
+            source.unpack(".")
+            assert source.is_orig == True
+            assert source.is_dir == False
+            assert type(source.unpacked) == str
+
+    def test_upstream_source_unpack(self):
+        for (comp, archive) in self.archives.iteritems():
+            source = gbp.deb.UpstreamSource(archive[0])
+            source.unpack(".")
+            self._check_files(archive[1], comp)
+
+    def test_upstream_source_unpack_no_filter(self):
+        for (comp, archive) in self.archives.iteritems():
+            source = gbp.deb.UpstreamSource(archive[0])
+            source.unpack(".", [])
+            self._check_files(archive[1], comp)
+
+    def test_upstream_source_unpack_filtered(self):
+        exclude = "git-buildpackage"
+
         for (comp, archive) in self.archives.iteritems():
-            gbp.deb.unpack_orig(archive, ".", [])
+            source = gbp.deb.UpstreamSource(archive[0])
+            source.unpack(".", [exclude])
+            archive[1].remove(exclude)
+            self._check_files(archive[1], comp)
 
 # vim:et:ts=4:sw=4:et:sts=4:ai:set list listchars=tab\:»·,trail\:·:
diff --git a/tests/06_test_upstream_source.py b/tests/06_test_upstream_source.py
new file mode 100644
index 00000000..13e0e95f
--- /dev/null
+++ b/tests/06_test_upstream_source.py
@@ -0,0 +1,86 @@
+# Test the UpstreamSource class
+
+import glob
+import os
+import shutil
+import tarfile
+import tempfile
+import unittest
+import zipfile
+
+from gbp.deb import UpstreamSource
+
+class TestDir(unittest.TestCase):
+    def test_directory(self):
+        """Upstream source is a directory"""
+        source = UpstreamSource('.')
+        self.assertEqual(source.is_orig, False)
+        self.assertEqual(source.path, '.')
+        self.assertEqual(source.unpacked, '.')
+
+
+class TestTar(unittest.TestCase):
+    def _check_tar(self, us, positive=[], negative=[]):
+        t = tarfile.open(name=us.path, mode="r:bz2")
+        for f in positive:
+            i = t.getmember(f)
+            self.assertEqual(type(i), tarfile.TarInfo)
+
+        for f in negative:
+            try:
+                t.getmember(f)
+                self.fail("Found %s in archive" % f)
+            except KeyError:
+                pass
+        t.close()
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix='gbp_%s_' % __name__, dir='.')
+
+    def tearDown(self):
+        if not os.getenv("GBP_TESTS_NOCLEAN"):
+            shutil.rmtree(self.tmpdir)
+
+    def test_pack_tar(self):
+        """Check if packing tar archives works"""
+        source = UpstreamSource(os.path.abspath("gbp/"))
+        target = os.path.join(self.tmpdir,
+                     "gbp_0.1.tar.bz2")
+        repacked = source.pack(target)
+        self.assertEqual(repacked.is_orig, True)
+        self.assertEqual(repacked.is_dir, False)
+        self._check_tar(repacked, ["gbp/deb.py", "gbp/__init__.py"])
+
+    def test_pack_filtered(self):
+        """Check if filtering out files works"""
+        source = UpstreamSource(os.path.abspath("gbp/"))
+        target = os.path.join(self.tmpdir,
+                     "gbp_0.1.tar.bz2")
+        repacked = source.pack(target, ["__init__.py"])
+        self.assertEqual(repacked.is_orig, True)
+        self.assertEqual(repacked.is_dir, False)
+        self._check_tar(repacked, ["gbp/deb.py"],
+                                  ["gbp/__init__.py"])
+
+
+class TestZip(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix='gbp_%s_' % __name__, dir='.')
+        self.zipfile = os.path.join(self.tmpdir, "gbp-0.1.zip")
+        z = zipfile.ZipFile(os.path.join(self.tmpdir, "gbp-0.1.zip"), "w")
+        for f in glob.glob("gbp/*.py"):
+            z.write(f, f, zipfile.ZIP_DEFLATED)
+        z.close()
+
+    def tearDown(self):
+        if not os.getenv("GBP_TESTS_NOCLEAN"):
+            shutil.rmtree(self.tmpdir)
+
+    def test_unpack(self):
+        source = UpstreamSource(self.zipfile)
+        self.assertEqual(source.is_orig, False)
+        self.assertEqual(source.is_dir, False)
+        self.assertEqual(source.unpacked, None)
+        source.unpack(self.tmpdir)
+        self.assertNotEqual(source.unpacked, None)
+