dulwich/tests/test_object_store.py

   1 # test_object_store.py -- tests for object_store.py
   2 # Copyright (C) 2008 Jelmer Vernooij <jelmer@samba.org>
   3 #
   4 # This program is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU General Public License
   6 # as published by the Free Software Foundation; version 2
   7 # or (at your option) any later version of the License.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program; if not, write to the Free Software
  16 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  17 # MA  02110-1301, USA.
  18
  19 """Tests for the object store interface."""
  20
  21
  22 import os
  23 import shutil
  24 import tempfile
  25
  26 from dulwich.index import (
  27     commit_tree,
  28     )
  29 from dulwich.errors import (
  30     NotTreeError,
  31     )
  32 from dulwich.objects import (
  33     object_class,
  34     Blob,
  35     Tag,
  36     Tree,
  37     TreeEntry,
  38     )
  39 from dulwich.object_store import (
  40     DiskObjectStore,
  41     MemoryObjectStore,
  42     tree_lookup_path,
  43     )
  44 from dulwich.pack import (
  45     write_pack_objects,
  46     )
  47 from dulwich.tests import (
  48     TestCase,
  49     )
  50 from dulwich.tests.utils import (
  51     make_object,
  52     )
  53
  54
  55 testobject = make_object(Blob, data="yummy data")
  56
  57
  58 class ObjectStoreTests(object):
  59
  60     def test_determine_wants_all(self):
  61         self.assertEquals(["1" * 40],
  62             self.store.determine_wants_all({"refs/heads/foo": "1" * 40}))
  63
  64     def test_determine_wants_all_zero(self):
  65         self.assertEquals([],
  66             self.store.determine_wants_all({"refs/heads/foo": "0" * 40}))
  67
  68     def test_iter(self):
  69         self.assertEquals([], list(self.store))
  70
  71     def test_get_nonexistant(self):
  72         self.assertRaises(KeyError, lambda: self.store["a" * 40])
  73
  74     def test_contains_nonexistant(self):
  75         self.assertFalse(("a" * 40) in self.store)
  76
  77     def test_add_objects_empty(self):
  78         self.store.add_objects([])
  79
  80     def test_add_commit(self):
  81         # TODO: Argh, no way to construct Git commit objects without
  82         # access to a serialized form.
  83         self.store.add_objects([])
  84
  85     def test_add_object(self):
  86         self.store.add_object(testobject)
  87         self.assertEquals(set([testobject.id]), set(self.store))
  88         self.assertTrue(testobject.id in self.store)
  89         r = self.store[testobject.id]
  90         self.assertEquals(r, testobject)
  91
  92     def test_add_objects(self):
  93         data = [(testobject, "mypath")]
  94         self.store.add_objects(data)
  95         self.assertEquals(set([testobject.id]), set(self.store))
  96         self.assertTrue(testobject.id in self.store)
  97         r = self.store[testobject.id]
  98         self.assertEquals(r, testobject)
  99
 100     def test_tree_changes(self):
 101         blob_a1 = make_object(Blob, data='a1')
 102         blob_a2 = make_object(Blob, data='a2')
 103         blob_b = make_object(Blob, data='b')
 104         for blob in [blob_a1, blob_a2, blob_b]:
 105             self.store.add_object(blob)
 106
 107         blobs_1 = [('a', blob_a1.id, 0100644), ('b', blob_b.id, 0100644)]
 108         tree1_id = commit_tree(self.store, blobs_1)
 109         blobs_2 = [('a', blob_a2.id, 0100644), ('b', blob_b.id, 0100644)]
 110         tree2_id = commit_tree(self.store, blobs_2)
 111         change_a = (('a', 'a'), (0100644, 0100644), (blob_a1.id, blob_a2.id))
 112         self.assertEquals([change_a],
 113                           list(self.store.tree_changes(tree1_id, tree2_id)))
 114         self.assertEquals(
 115           [change_a, (('b', 'b'), (0100644, 0100644), (blob_b.id, blob_b.id))],
 116           list(self.store.tree_changes(tree1_id, tree2_id,
 117                                        want_unchanged=True)))
 118
 119     def test_iter_tree_contents(self):
 120         blob_a = make_object(Blob, data='a')
 121         blob_b = make_object(Blob, data='b')
 122         blob_c = make_object(Blob, data='c')
 123         for blob in [blob_a, blob_b, blob_c]:
 124             self.store.add_object(blob)
 125
 126         blobs = [
 127           ('a', blob_a.id, 0100644),
 128           ('ad/b', blob_b.id, 0100644),
 129           ('ad/bd/c', blob_c.id, 0100755),
 130           ('ad/c', blob_c.id, 0100644),
 131           ('c', blob_c.id, 0100644),
 132           ]
 133         tree_id = commit_tree(self.store, blobs)
 134         self.assertEquals([TreeEntry(p, m, h) for (p, h, m) in blobs],
 135                           list(self.store.iter_tree_contents(tree_id)))
 136
 137     def test_iter_tree_contents_include_trees(self):
 138         blob_a = make_object(Blob, data='a')
 139         blob_b = make_object(Blob, data='b')
 140         blob_c = make_object(Blob, data='c')
 141         for blob in [blob_a, blob_b, blob_c]:
 142             self.store.add_object(blob)
 143
 144         blobs = [
 145           ('a', blob_a.id, 0100644),
 146           ('ad/b', blob_b.id, 0100644),
 147           ('ad/bd/c', blob_c.id, 0100755),
 148           ]
 149         tree_id = commit_tree(self.store, blobs)
 150         tree = self.store[tree_id]
 151         tree_ad = self.store[tree['ad'][1]]
 152         tree_bd = self.store[tree_ad['bd'][1]]
 153
 154         expected = [
 155           TreeEntry('', 0040000, tree_id),
 156           TreeEntry('a', 0100644, blob_a.id),
 157           TreeEntry('ad', 0040000, tree_ad.id),
 158           TreeEntry('ad/b', 0100644, blob_b.id),
 159           TreeEntry('ad/bd', 0040000, tree_bd.id),
 160           TreeEntry('ad/bd/c', 0100755, blob_c.id),
 161           ]
 162         actual = self.store.iter_tree_contents(tree_id, include_trees=True)
 163         self.assertEquals(expected, list(actual))
 164
 165     def make_tag(self, name, obj):
 166         tag = make_object(Tag, name=name, message='',
 167                           tag_time=12345, tag_timezone=0,
 168                           tagger='Test Tagger <test@example.com>',
 169                           object=(object_class(obj.type_name), obj.id))
 170         self.store.add_object(tag)
 171         return tag
 172
 173     def test_peel_sha(self):
 174         self.store.add_object(testobject)
 175         tag1 = self.make_tag('1', testobject)
 176         tag2 = self.make_tag('2', testobject)
 177         tag3 = self.make_tag('3', testobject)
 178         for obj in [testobject, tag1, tag2, tag3]:
 179             self.assertEqual(testobject, self.store.peel_sha(obj.id))
 180
 181
 182 class MemoryObjectStoreTests(ObjectStoreTests, TestCase):
 183
 184     def setUp(self):
 185         TestCase.setUp(self)
 186         self.store = MemoryObjectStore()
 187
 188
 189 class PackBasedObjectStoreTests(ObjectStoreTests):
 190
 191     def tearDown(self):
 192         for pack in self.store.packs:
 193             pack.close()
 194
 195     def test_empty_packs(self):
 196         self.assertEquals([], self.store.packs)
 197
 198     def test_pack_loose_objects(self):
 199         b1 = make_object(Blob, data="yummy data")
 200         self.store.add_object(b1)
 201         b2 = make_object(Blob, data="more yummy data")
 202         self.store.add_object(b2)
 203         self.assertEquals([], self.store.packs)
 204         self.assertEquals(2, self.store.pack_loose_objects())
 205         self.assertNotEquals([], self.store.packs)
 206         self.assertEquals(0, self.store.pack_loose_objects())
 207
 208
 209 class DiskObjectStoreTests(PackBasedObjectStoreTests, TestCase):
 210
 211     def setUp(self):
 212         TestCase.setUp(self)
 213         self.store_dir = tempfile.mkdtemp()
 214         self.store = DiskObjectStore.init(self.store_dir)
 215
 216     def tearDown(self):
 217         TestCase.tearDown(self)
 218         PackBasedObjectStoreTests.tearDown(self)
 219         shutil.rmtree(self.store_dir)
 220
 221     def test_pack_dir(self):
 222         o = DiskObjectStore(self.store_dir)
 223         self.assertEquals(os.path.join(self.store_dir, "pack"), o.pack_dir)
 224
 225     def test_add_pack(self):
 226         o = DiskObjectStore(self.store_dir)
 227         f, commit = o.add_pack()
 228         b = make_object(Blob, data="more yummy data")
 229         write_pack_objects(f, [(b, None)])
 230         commit()
 231
 232     def test_add_thin_pack(self):
 233         o = DiskObjectStore(self.store_dir)
 234         f, commit = o.add_thin_pack()
 235         b = make_object(Blob, data="more yummy data")
 236         write_pack_objects(f, [(b, None)])
 237         commit()
 238
 239
 240 class TreeLookupPathTests(TestCase):
 241
 242     def setUp(self):
 243         TestCase.setUp(self)
 244         self.store = MemoryObjectStore()
 245         blob_a = make_object(Blob, data='a')
 246         blob_b = make_object(Blob, data='b')
 247         blob_c = make_object(Blob, data='c')
 248         for blob in [blob_a, blob_b, blob_c]:
 249             self.store.add_object(blob)
 250
 251         blobs = [
 252           ('a', blob_a.id, 0100644),
 253           ('ad/b', blob_b.id, 0100644),
 254           ('ad/bd/c', blob_c.id, 0100755),
 255           ('ad/c', blob_c.id, 0100644),
 256           ('c', blob_c.id, 0100644),
 257           ]
 258         self.tree_id = commit_tree(self.store, blobs)
 259
 260     def get_object(self, sha):
 261         return self.store[sha]
 262
 263     def test_lookup_blob(self):
 264         o_id = tree_lookup_path(self.get_object, self.tree_id, 'a')[1]
 265         self.assertTrue(isinstance(self.store[o_id], Blob))
 266
 267     def test_lookup_tree(self):
 268         o_id = tree_lookup_path(self.get_object, self.tree_id, 'ad')[1]
 269         self.assertTrue(isinstance(self.store[o_id], Tree))
 270         o_id = tree_lookup_path(self.get_object, self.tree_id, 'ad/bd')[1]
 271         self.assertTrue(isinstance(self.store[o_id], Tree))
 272         o_id = tree_lookup_path(self.get_object, self.tree_id, 'ad/bd/')[1]
 273         self.assertTrue(isinstance(self.store[o_id], Tree))
 274
 275     def test_lookup_nonexistent(self):
 276         self.assertRaises(KeyError, tree_lookup_path, self.get_object, self.tree_id, 'j')
 277
 278     def test_lookup_not_tree(self):
 279         self.assertRaises(NotTreeError, tree_lookup_path, self.get_object, self.tree_id, 'ad/b/j')
 280
 281 # TODO: MissingObjectFinderTests