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 from io import BytesIO
  23 import os
  24 import shutil
  25 import tempfile
  26
  27 from dulwich.index import (
  28     commit_tree,
  29     )
  30 from dulwich.errors import (
  31     NotTreeError,
  32     )
  33 from dulwich.objects import (
  34     sha_to_hex,
  35     object_class,
  36     Blob,
  37     Tag,
  38     Tree,
  39     TreeEntry,
  40     )
  41 from dulwich.object_store import (
  42     DiskObjectStore,
  43     MemoryObjectStore,
  44     ObjectStoreGraphWalker,
  45     tree_lookup_path,
  46     )
  47 from dulwich.pack import (
  48     REF_DELTA,
  49     write_pack_objects,
  50     )
  51 from dulwich.tests import (
  52     TestCase,
  53     )
  54 from dulwich.tests.utils import (
  55     make_object,
  56     build_pack,
  57     )
  58
  59
  60 testobject = make_object(Blob, data="yummy data")
  61
  62
  63 class ObjectStoreTests(object):
  64
  65     def test_determine_wants_all(self):
  66         self.assertEqual(["1" * 40],
  67             self.store.determine_wants_all({"refs/heads/foo": "1" * 40}))
  68
  69     def test_determine_wants_all_zero(self):
  70         self.assertEqual([],
  71             self.store.determine_wants_all({"refs/heads/foo": "0" * 40}))
  72
  73     def test_iter(self):
  74         self.assertEqual([], list(self.store))
  75
  76     def test_get_nonexistant(self):
  77         self.assertRaises(KeyError, lambda: self.store["a" * 40])
  78
  79     def test_contains_nonexistant(self):
  80         self.assertFalse(("a" * 40) in self.store)
  81
  82     def test_add_objects_empty(self):
  83         self.store.add_objects([])
  84
  85     def test_add_commit(self):
  86         # TODO: Argh, no way to construct Git commit objects without
  87         # access to a serialized form.
  88         self.store.add_objects([])
  89
  90     def test_add_object(self):
  91         self.store.add_object(testobject)
  92         self.assertEqual(set([testobject.id]), set(self.store))
  93         self.assertTrue(testobject.id in self.store)
  94         r = self.store[testobject.id]
  95         self.assertEqual(r, testobject)
  96
  97     def test_add_objects(self):
  98         data = [(testobject, "mypath")]
  99         self.store.add_objects(data)
 100         self.assertEqual(set([testobject.id]), set(self.store))
 101         self.assertTrue(testobject.id in self.store)
 102         r = self.store[testobject.id]
 103         self.assertEqual(r, testobject)
 104
 105     def test_tree_changes(self):
 106         blob_a1 = make_object(Blob, data='a1')
 107         blob_a2 = make_object(Blob, data='a2')
 108         blob_b = make_object(Blob, data='b')
 109         for blob in [blob_a1, blob_a2, blob_b]:
 110             self.store.add_object(blob)
 111
 112         blobs_1 = [('a', blob_a1.id, 0o100644), ('b', blob_b.id, 0o100644)]
 113         tree1_id = commit_tree(self.store, blobs_1)
 114         blobs_2 = [('a', blob_a2.id, 0o100644), ('b', blob_b.id, 0o100644)]
 115         tree2_id = commit_tree(self.store, blobs_2)
 116         change_a = (('a', 'a'), (0o100644, 0o100644), (blob_a1.id, blob_a2.id))
 117         self.assertEqual([change_a],
 118                           list(self.store.tree_changes(tree1_id, tree2_id)))
 119         self.assertEqual(
 120           [change_a, (('b', 'b'), (0o100644, 0o100644), (blob_b.id, blob_b.id))],
 121           list(self.store.tree_changes(tree1_id, tree2_id,
 122                                        want_unchanged=True)))
 123
 124     def test_iter_tree_contents(self):
 125         blob_a = make_object(Blob, data='a')
 126         blob_b = make_object(Blob, data='b')
 127         blob_c = make_object(Blob, data='c')
 128         for blob in [blob_a, blob_b, blob_c]:
 129             self.store.add_object(blob)
 130
 131         blobs = [
 132           ('a', blob_a.id, 0o100644),
 133           ('ad/b', blob_b.id, 0o100644),
 134           ('ad/bd/c', blob_c.id, 0o100755),
 135           ('ad/c', blob_c.id, 0o100644),
 136           ('c', blob_c.id, 0o100644),
 137           ]
 138         tree_id = commit_tree(self.store, blobs)
 139         self.assertEqual([TreeEntry(p, m, h) for (p, h, m) in blobs],
 140                           list(self.store.iter_tree_contents(tree_id)))
 141
 142     def test_iter_tree_contents_include_trees(self):
 143         blob_a = make_object(Blob, data='a')
 144         blob_b = make_object(Blob, data='b')
 145         blob_c = make_object(Blob, data='c')
 146         for blob in [blob_a, blob_b, blob_c]:
 147             self.store.add_object(blob)
 148
 149         blobs = [
 150           ('a', blob_a.id, 0o100644),
 151           ('ad/b', blob_b.id, 0o100644),
 152           ('ad/bd/c', blob_c.id, 0o100755),
 153           ]
 154         tree_id = commit_tree(self.store, blobs)
 155         tree = self.store[tree_id]
 156         tree_ad = self.store[tree['ad'][1]]
 157         tree_bd = self.store[tree_ad['bd'][1]]
 158
 159         expected = [
 160           TreeEntry('', 0o040000, tree_id),
 161           TreeEntry('a', 0o100644, blob_a.id),
 162           TreeEntry('ad', 0o040000, tree_ad.id),
 163           TreeEntry('ad/b', 0o100644, blob_b.id),
 164           TreeEntry('ad/bd', 0o040000, tree_bd.id),
 165           TreeEntry('ad/bd/c', 0o100755, blob_c.id),
 166           ]
 167         actual = self.store.iter_tree_contents(tree_id, include_trees=True)
 168         self.assertEqual(expected, list(actual))
 169
 170     def make_tag(self, name, obj):
 171         tag = make_object(Tag, name=name, message='',
 172                           tag_time=12345, tag_timezone=0,
 173                           tagger='Test Tagger <test@example.com>',
 174                           object=(object_class(obj.type_name), obj.id))
 175         self.store.add_object(tag)
 176         return tag
 177
 178     def test_peel_sha(self):
 179         self.store.add_object(testobject)
 180         tag1 = self.make_tag('1', testobject)
 181         tag2 = self.make_tag('2', testobject)
 182         tag3 = self.make_tag('3', testobject)
 183         for obj in [testobject, tag1, tag2, tag3]:
 184             self.assertEqual(testobject, self.store.peel_sha(obj.id))
 185
 186     def test_get_raw(self):
 187         self.store.add_object(testobject)
 188         self.assertEqual((Blob.type_num, 'yummy data'),
 189                          self.store.get_raw(testobject.id))
 190
 191     def test_close(self):
 192         # For now, just check that close doesn't barf.
 193         self.store.add_object(testobject)
 194         self.store.close()
 195
 196
 197 class MemoryObjectStoreTests(ObjectStoreTests, TestCase):
 198
 199     def setUp(self):
 200         TestCase.setUp(self)
 201         self.store = MemoryObjectStore()
 202
 203     def test_add_pack(self):
 204         o = MemoryObjectStore()
 205         f, commit, abort = o.add_pack()
 206         try:
 207             b = make_object(Blob, data="more yummy data")
 208             write_pack_objects(f, [(b, None)])
 209         except:
 210             abort()
 211             raise
 212         else:
 213             commit()
 214
 215     def test_add_thin_pack(self):
 216         o = MemoryObjectStore()
 217         blob = make_object(Blob, data='yummy data')
 218         o.add_object(blob)
 219
 220         f = BytesIO()
 221         entries = build_pack(f, [
 222           (REF_DELTA, (blob.id, 'more yummy data')),
 223           ], store=o)
 224         o.add_thin_pack(f.read, None)
 225         packed_blob_sha = sha_to_hex(entries[0][3])
 226         self.assertEqual((Blob.type_num, 'more yummy data'),
 227                          o.get_raw(packed_blob_sha))
 228
 229
 230     def test_add_thin_pack_empty(self):
 231         o = MemoryObjectStore()
 232
 233         f = BytesIO()
 234         entries = build_pack(f, [], store=o)
 235         self.assertEquals([], entries)
 236         o.add_thin_pack(f.read, None)
 237
 238
 239 class PackBasedObjectStoreTests(ObjectStoreTests):
 240
 241     def tearDown(self):
 242         for pack in self.store.packs:
 243             pack.close()
 244
 245     def test_empty_packs(self):
 246         self.assertEqual([], self.store.packs)
 247
 248     def test_pack_loose_objects(self):
 249         b1 = make_object(Blob, data="yummy data")
 250         self.store.add_object(b1)
 251         b2 = make_object(Blob, data="more yummy data")
 252         self.store.add_object(b2)
 253         self.assertEqual([], self.store.packs)
 254         self.assertEqual(2, self.store.pack_loose_objects())
 255         self.assertNotEqual([], self.store.packs)
 256         self.assertEqual(0, self.store.pack_loose_objects())
 257
 258
 259 class DiskObjectStoreTests(PackBasedObjectStoreTests, TestCase):
 260
 261     def setUp(self):
 262         TestCase.setUp(self)
 263         self.store_dir = tempfile.mkdtemp()
 264         self.addCleanup(shutil.rmtree, self.store_dir)
 265         self.store = DiskObjectStore.init(self.store_dir)
 266
 267     def tearDown(self):
 268         TestCase.tearDown(self)
 269         PackBasedObjectStoreTests.tearDown(self)
 270
 271     def test_alternates(self):
 272         alternate_dir = tempfile.mkdtemp()
 273         self.addCleanup(shutil.rmtree, alternate_dir)
 274         alternate_store = DiskObjectStore(alternate_dir)
 275         b2 = make_object(Blob, data="yummy data")
 276         alternate_store.add_object(b2)
 277         store = DiskObjectStore(self.store_dir)
 278         self.assertRaises(KeyError, store.__getitem__, b2.id)
 279         store.add_alternate_path(alternate_dir)
 280         self.assertIn(b2.id, store)
 281         self.assertEqual(b2, store[b2.id])
 282
 283     def test_add_alternate_path(self):
 284         store = DiskObjectStore(self.store_dir)
 285         self.assertEqual([], store._read_alternate_paths())
 286         store.add_alternate_path("/foo/path")
 287         self.assertEqual(["/foo/path"], store._read_alternate_paths())
 288         store.add_alternate_path("/bar/path")
 289         self.assertEqual(
 290             ["/foo/path", "/bar/path"],
 291             store._read_alternate_paths())
 292
 293     def test_rel_alternative_path(self):
 294         alternate_dir = tempfile.mkdtemp()
 295         self.addCleanup(shutil.rmtree, alternate_dir)
 296         alternate_store = DiskObjectStore(alternate_dir)
 297         b2 = make_object(Blob, data="yummy data")
 298         alternate_store.add_object(b2)
 299         store = DiskObjectStore(self.store_dir)
 300         self.assertRaises(KeyError, store.__getitem__, b2.id)
 301         store.add_alternate_path(os.path.relpath(alternate_dir, self.store_dir))
 302         self.assertEqual(list(alternate_store), list(store.alternates[0]))
 303         self.assertIn(b2.id, store)
 304         self.assertEqual(b2, store[b2.id])
 305
 306     def test_pack_dir(self):
 307         o = DiskObjectStore(self.store_dir)
 308         self.assertEqual(os.path.join(self.store_dir, "pack"), o.pack_dir)
 309
 310     def test_add_pack(self):
 311         o = DiskObjectStore(self.store_dir)
 312         f, commit, abort = o.add_pack()
 313         try:
 314             b = make_object(Blob, data="more yummy data")
 315             write_pack_objects(f, [(b, None)])
 316         except:
 317             abort()
 318             raise
 319         else:
 320             commit()
 321
 322     def test_add_thin_pack(self):
 323         o = DiskObjectStore(self.store_dir)
 324         try:
 325             blob = make_object(Blob, data='yummy data')
 326             o.add_object(blob)
 327
 328             f = BytesIO()
 329             entries = build_pack(f, [
 330               (REF_DELTA, (blob.id, 'more yummy data')),
 331               ], store=o)
 332
 333             with o.add_thin_pack(f.read, None) as pack:
 334                 packed_blob_sha = sha_to_hex(entries[0][3])
 335                 pack.check_length_and_checksum()
 336                 self.assertEqual(sorted([blob.id, packed_blob_sha]), list(pack))
 337                 self.assertTrue(o.contains_packed(packed_blob_sha))
 338                 self.assertTrue(o.contains_packed(blob.id))
 339                 self.assertEqual((Blob.type_num, 'more yummy data'),
 340                                  o.get_raw(packed_blob_sha))
 341         finally:
 342             o.close()
 343
 344     def test_add_thin_pack_empty(self):
 345         o = DiskObjectStore(self.store_dir)
 346
 347         f = BytesIO()
 348         entries = build_pack(f, [], store=o)
 349         self.assertEquals([], entries)
 350         o.add_thin_pack(f.read, None)
 351
 352
 353 class TreeLookupPathTests(TestCase):
 354
 355     def setUp(self):
 356         TestCase.setUp(self)
 357         self.store = MemoryObjectStore()
 358         blob_a = make_object(Blob, data='a')
 359         blob_b = make_object(Blob, data='b')
 360         blob_c = make_object(Blob, data='c')
 361         for blob in [blob_a, blob_b, blob_c]:
 362             self.store.add_object(blob)
 363
 364         blobs = [
 365           ('a', blob_a.id, 0o100644),
 366           ('ad/b', blob_b.id, 0o100644),
 367           ('ad/bd/c', blob_c.id, 0o100755),
 368           ('ad/c', blob_c.id, 0o100644),
 369           ('c', blob_c.id, 0o100644),
 370           ]
 371         self.tree_id = commit_tree(self.store, blobs)
 372
 373     def get_object(self, sha):
 374         return self.store[sha]
 375
 376     def test_lookup_blob(self):
 377         o_id = tree_lookup_path(self.get_object, self.tree_id, 'a')[1]
 378         self.assertTrue(isinstance(self.store[o_id], Blob))
 379
 380     def test_lookup_tree(self):
 381         o_id = tree_lookup_path(self.get_object, self.tree_id, 'ad')[1]
 382         self.assertTrue(isinstance(self.store[o_id], Tree))
 383         o_id = tree_lookup_path(self.get_object, self.tree_id, 'ad/bd')[1]
 384         self.assertTrue(isinstance(self.store[o_id], Tree))
 385         o_id = tree_lookup_path(self.get_object, self.tree_id, 'ad/bd/')[1]
 386         self.assertTrue(isinstance(self.store[o_id], Tree))
 387
 388     def test_lookup_nonexistent(self):
 389         self.assertRaises(KeyError, tree_lookup_path, self.get_object, self.tree_id, 'j')
 390
 391     def test_lookup_not_tree(self):
 392         self.assertRaises(NotTreeError, tree_lookup_path, self.get_object, self.tree_id, 'ad/b/j')
 393
 394 # TODO: MissingObjectFinderTests
 395
 396 class ObjectStoreGraphWalkerTests(TestCase):
 397
 398     def get_walker(self, heads, parent_map):
 399         new_parent_map = dict([
 400             (k * 40, [(p * 40) for p in ps]) for (k, ps) in parent_map.items()])
 401         return ObjectStoreGraphWalker([x * 40 for x in heads],
 402             new_parent_map.__getitem__)
 403
 404     def test_empty(self):
 405         gw = self.get_walker([], {})
 406         self.assertIs(None, next(gw))
 407         gw.ack("a" * 40)
 408         self.assertIs(None, next(gw))
 409
 410     def test_descends(self):
 411         gw = self.get_walker(["a"], {"a": ["b"], "b": []})
 412         self.assertEqual("a" * 40, next(gw))
 413         self.assertEqual("b" * 40, next(gw))
 414
 415     def test_present(self):
 416         gw = self.get_walker(["a"], {"a": ["b"], "b": []})
 417         gw.ack("a" * 40)
 418         self.assertIs(None, next(gw))
 419
 420     def test_parent_present(self):
 421         gw = self.get_walker(["a"], {"a": ["b"], "b": []})
 422         self.assertEqual("a" * 40, next(gw))
 423         gw.ack("a" * 40)
 424         self.assertIs(None, next(gw))
 425
 426     def test_child_ack_later(self):
 427         gw = self.get_walker(["a"], {"a": ["b"], "b": ["c"], "c": []})
 428         self.assertEqual("a" * 40, next(gw))
 429         self.assertEqual("b" * 40, next(gw))
 430         gw.ack("a" * 40)
 431         self.assertIs(None, next(gw))
 432
 433     def test_only_once(self):
 434         # a  b
 435         # |  |
 436         # c  d
 437         # \ /
 438         #  e
 439         gw = self.get_walker(["a", "b"], {
 440                 "a": ["c"],
 441                 "b": ["d"],
 442                 "c": ["e"],
 443                 "d": ["e"],
 444                 "e": [],
 445                 })
 446         walk = []
 447         acked = False
 448         walk.append(next(gw))
 449         walk.append(next(gw))
 450         # A branch (a, c) or (b, d) may be done after 2 steps or 3 depending on
 451         # the order walked: 3-step walks include (a, b, c) and (b, a, d), etc.
 452         if walk == ["a" * 40, "c" * 40] or walk == ["b" * 40, "d" * 40]:
 453           gw.ack(walk[0])
 454           acked = True
 455
 456         walk.append(next(gw))
 457         if not acked and walk[2] == "c" * 40:
 458           gw.ack("a" * 40)
 459         elif not acked and walk[2] == "d" * 40:
 460           gw.ack("b" * 40)
 461         walk.append(next(gw))
 462         self.assertIs(None, next(gw))
 463
 464         self.assertEqual(["a" * 40, "b" * 40, "c" * 40, "d" * 40], sorted(walk))
 465         self.assertLess(walk.index("a" * 40), walk.index("c" * 40))
 466         self.assertLess(walk.index("b" * 40), walk.index("d" * 40))