dulwich/tests/test_pack.py

   1 # test_pack.py -- Tests for the handling of git packs.
   2 # Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
   3 # Copyright (C) 2008 Jelmer Vernooij <jelmer@samba.org>
   4 #
   5 # This program is free software; you can redistribute it and/or
   6 # modify it under the terms of the GNU General Public License
   7 # as published by the Free Software Foundation; version 2
   8 # of the License, or (at your option) any later version of the license.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with this program; if not, write to the Free Software
  17 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  18 # MA  02110-1301, USA.
  19
  20 import os
  21 import unittest
  22
  23 from dulwich.objects import (
  24         Tree,
  25         )
  26 from dulwich.pack import (
  27         Pack,
  28         PackIndex,
  29         PackData,
  30         hex_to_sha,
  31         sha_to_hex,
  32         write_pack_index_v1,
  33         write_pack_index_v2,
  34         write_pack,
  35         apply_delta,
  36         create_delta,
  37         )
  38
  39 pack1_sha = 'bc63ddad95e7321ee734ea11a7a62d314e0d7481'
  40
  41 a_sha = '6f670c0fb53f9463760b7295fbb814e965fb20c8'
  42 tree_sha = 'b2a2766a2879c209ab1176e7e778b81ae422eeaa'
  43 commit_sha = 'f18faa16531ac570a3fdc8c7ca16682548dafd12'
  44
  45 class PackTests(unittest.TestCase):
  46   """Base class for testing packs"""
  47
  48   datadir = os.path.join(os.path.dirname(__file__), 'data/packs')
  49
  50   def get_pack_index(self, sha):
  51     """Returns a PackIndex from the datadir with the given sha"""
  52     return PackIndex(os.path.join(self.datadir, 'pack-%s.idx' % sha))
  53
  54   def get_pack_data(self, sha):
  55     """Returns a PackData object from the datadir with the given sha"""
  56     return PackData(os.path.join(self.datadir, 'pack-%s.pack' % sha))
  57
  58   def get_pack(self, sha):
  59     return Pack(os.path.join(self.datadir, 'pack-%s' % sha))
  60
  61
  62 class PackIndexTests(PackTests):
  63   """Class that tests the index of packfiles"""
  64
  65   def test_object_index(self):
  66     """Tests that the correct object offset is returned from the index."""
  67     p = self.get_pack_index(pack1_sha)
  68     self.assertEqual(p.object_index(pack1_sha), None)
  69     self.assertEqual(p.object_index(a_sha), 178)
  70     self.assertEqual(p.object_index(tree_sha), 138)
  71     self.assertEqual(p.object_index(commit_sha), 12)
  72
  73   def test_index_len(self):
  74     p = self.get_pack_index(pack1_sha)
  75     self.assertEquals(3, len(p))
  76
  77   def test_get_stored_checksum(self):
  78     p = self.get_pack_index(pack1_sha)
  79     self.assertEquals("\xf2\x84\x8e*\xd1o2\x9a\xe1\xc9.;\x95\xe9\x18\x88\xda\xa5\xbd\x01", str(p.get_stored_checksums()[1]))
  80     self.assertEquals( 'r\x19\x80\xe8f\xaf\x9a_\x93\xadgAD\xe1E\x9b\x8b\xa3\xe7\xb7' , str(p.get_stored_checksums()[0]))
  81
  82   def test_index_check(self):
  83     p = self.get_pack_index(pack1_sha)
  84     self.assertEquals(True, p.check())
  85
  86
  87   def test_iterentries(self):
  88     p = self.get_pack_index(pack1_sha)
  89     self.assertEquals([('og\x0c\x0f\xb5?\x94cv\x0br\x95\xfb\xb8\x14\xe9e\xfb \xc8', 178, None), ('\xb2\xa2vj(y\xc2\t\xab\x11v\xe7\xe7x\xb8\x1a\xe4"\xee\xaa', 138, None), ('\xf1\x8f\xaa\x16S\x1a\xc5p\xa3\xfd\xc8\xc7\xca\x16h%H\xda\xfd\x12', 12, None)], list(p.iterentries()))
  90
  91   def test_iter(self):
  92     p = self.get_pack_index(pack1_sha)
  93     self.assertEquals(set([tree_sha, commit_sha, a_sha]), set(p))
  94
  95
  96 class TestPackDeltas(unittest.TestCase):
  97
  98   test_string1 = "The answer was flailing in the wind"
  99   test_string2 = "The answer was falling down the pipe"
 100   test_string3 = "zzzzz"
 101
 102   test_string_empty = ""
 103   test_string_big = "Z" * 8192
 104
 105   def _test_roundtrip(self, base, target):
 106     self.assertEquals(target,
 107       apply_delta(base, create_delta(base, target)))
 108
 109   def test_nochange(self):
 110     self._test_roundtrip(self.test_string1, self.test_string1)
 111
 112   def test_change(self):
 113     self._test_roundtrip(self.test_string1, self.test_string2)
 114
 115   def test_rewrite(self):
 116     self._test_roundtrip(self.test_string1, self.test_string3)
 117
 118   def test_overflow(self):
 119     self._test_roundtrip(self.test_string_empty, self.test_string_big)
 120
 121
 122 class TestPackData(PackTests):
 123   """Tests getting the data from the packfile."""
 124
 125   def test_create_pack(self):
 126     p = self.get_pack_data(pack1_sha)
 127
 128   def test_pack_len(self):
 129     p = self.get_pack_data(pack1_sha)
 130     self.assertEquals(3, len(p))
 131
 132   def test_index_check(self):
 133     p = self.get_pack_data(pack1_sha)
 134     self.assertEquals(True, p.check())
 135
 136   def test_iterobjects(self):
 137     p = self.get_pack_data(pack1_sha)
 138     self.assertEquals([(12, 1, 'tree b2a2766a2879c209ab1176e7e778b81ae422eeaa\nauthor James Westby <jw+debian@jameswestby.net> 1174945067 +0100\ncommitter James Westby <jw+debian@jameswestby.net> 1174945067 +0100\n\nTest commit\n'), (138, 2, '100644 a\x00og\x0c\x0f\xb5?\x94cv\x0br\x95\xfb\xb8\x14\xe9e\xfb \xc8'), (178, 3, 'test 1\n')], list(p.iterobjects()))
 139
 140   def test_iterentries(self):
 141     p = self.get_pack_data(pack1_sha)
 142     self.assertEquals(set([('og\x0c\x0f\xb5?\x94cv\x0br\x95\xfb\xb8\x14\xe9e\xfb \xc8', 178, -1718046665), ('\xb2\xa2vj(y\xc2\t\xab\x11v\xe7\xe7x\xb8\x1a\xe4"\xee\xaa', 138, -901046474), ('\xf1\x8f\xaa\x16S\x1a\xc5p\xa3\xfd\xc8\xc7\xca\x16h%H\xda\xfd\x12', 12, 1185722901)]), set(p.iterentries()))
 143
 144   def test_create_index_v1(self):
 145     p = self.get_pack_data(pack1_sha)
 146     p.create_index_v1("v1test.idx")
 147     idx1 = PackIndex("v1test.idx")
 148     idx2 = self.get_pack_index(pack1_sha)
 149     self.assertEquals(idx1, idx2)
 150
 151   def test_create_index_v2(self):
 152     p = self.get_pack_data(pack1_sha)
 153     p.create_index_v2("v2test.idx")
 154     idx1 = PackIndex("v2test.idx")
 155     idx2 = self.get_pack_index(pack1_sha)
 156     self.assertEquals(idx1, idx2)
 157
 158
 159
 160 class TestPack(PackTests):
 161
 162     def test_len(self):
 163         p = self.get_pack(pack1_sha)
 164         self.assertEquals(3, len(p))
 165
 166     def test_contains(self):
 167         p = self.get_pack(pack1_sha)
 168         self.assertTrue(tree_sha in p)
 169
 170     def test_get(self):
 171         p = self.get_pack(pack1_sha)
 172         self.assertEquals(type(p[tree_sha]), Tree)
 173
 174     def test_iter(self):
 175         p = self.get_pack(pack1_sha)
 176         self.assertEquals(set([tree_sha, commit_sha, a_sha]), set(p))
 177
 178     def test_get_object_at(self):
 179         """Tests random access for non-delta objects"""
 180         p = self.get_pack(pack1_sha)
 181         obj = p[a_sha]
 182         self.assertEqual(obj._type, 'blob')
 183         self.assertEqual(obj.sha().hexdigest(), a_sha)
 184         obj = p[tree_sha]
 185         self.assertEqual(obj._type, 'tree')
 186         self.assertEqual(obj.sha().hexdigest(), tree_sha)
 187         obj = p[commit_sha]
 188         self.assertEqual(obj._type, 'commit')
 189         self.assertEqual(obj.sha().hexdigest(), commit_sha)
 190
 191     def test_copy(self):
 192         p = self.get_pack(pack1_sha)
 193         write_pack("Elch", p.iterobjects(), len(p))
 194         self.assertEquals(p, Pack("Elch"))
 195
 196     def test_commit_obj(self):
 197         p = self.get_pack(pack1_sha)
 198         commit = p[commit_sha]
 199         self.assertEquals("James Westby <jw+debian@jameswestby.net>", commit.author)
 200         self.assertEquals([], commit.parents)
 201
 202     def test_name(self):
 203         p = self.get_pack(pack1_sha)
 204         self.assertEquals(pack1_sha, p.name())
 205
 206
 207 class TestHexToSha(unittest.TestCase):
 208
 209     def test_simple(self):
 210         self.assertEquals('\xab\xcd' * 10, hex_to_sha("abcd" * 10))
 211
 212     def test_reverse(self):
 213         self.assertEquals("abcd" * 10, sha_to_hex('\xab\xcd' * 10))
 214
 215
 216 class BaseTestPackIndexWriting(object):
 217
 218     def test_empty(self):
 219         pack_checksum = 'r\x19\x80\xe8f\xaf\x9a_\x93\xadgAD\xe1E\x9b\x8b\xa3\xe7\xb7'
 220         self._write_fn("empty.idx", [], pack_checksum)
 221         idx = PackIndex("empty.idx")
 222         self.assertTrue(idx.check())
 223         self.assertEquals(idx.get_stored_checksums()[0], pack_checksum)
 224         self.assertEquals(0, len(idx))
 225
 226     def test_single(self):
 227         pack_checksum = 'r\x19\x80\xe8f\xaf\x9a_\x93\xadgAD\xe1E\x9b\x8b\xa3\xe7\xb7'
 228         my_entries = [('og\x0c\x0f\xb5?\x94cv\x0br\x95\xfb\xb8\x14\xe9e\xfb \xc8', 178, 42)]
 229         my_entries.sort()
 230         self._write_fn("single.idx", my_entries, pack_checksum)
 231         idx = PackIndex("single.idx")
 232         self.assertEquals(idx.version, self._expected_version)
 233         self.assertTrue(idx.check())
 234         self.assertEquals(idx.get_stored_checksums()[0], pack_checksum)
 235         self.assertEquals(1, len(idx))
 236         actual_entries = list(idx.iterentries())
 237         self.assertEquals(len(my_entries), len(actual_entries))
 238         for a, b in zip(my_entries, actual_entries):
 239             self.assertEquals(a[0], b[0])
 240             self.assertEquals(a[1], b[1])
 241             if self._has_crc32_checksum:
 242                 self.assertEquals(a[2], b[2])
 243             else:
 244                 self.assertTrue(b[2] is None)
 245
 246
 247 class TestPackIndexWritingv1(unittest.TestCase, BaseTestPackIndexWriting):
 248
 249     def setUp(self):
 250         unittest.TestCase.setUp(self)
 251         self._has_crc32_checksum = False
 252         self._expected_version = 1
 253         self._write_fn = write_pack_index_v1
 254
 255
 256 class TestPackIndexWritingv2(unittest.TestCase, BaseTestPackIndexWriting):
 257
 258     def setUp(self):
 259         unittest.TestCase.setUp(self)
 260         self._has_crc32_checksum = True
 261         self._expected_version = 2
 262         self._write_fn = write_pack_index_v2