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.objects import (
  30     Blob,
  31     )
  32 from dulwich.object_store import (
  33     DiskObjectStore,
  34     MemoryObjectStore,
  35     )
  36 from dulwich.pack import (
  37     write_pack_data,
  38     )
  39 from dulwich.tests import (
  40     TestCase,
  41     )
  42 from utils import (
  43     make_object,
  44     )
  45
  46
  47 testobject = make_object(Blob, data="yummy data")
  48
  49
  50 class ObjectStoreTests(object):
  51
  52     def test_iter(self):
  53         self.assertEquals([], list(self.store))
  54
  55     def test_get_nonexistant(self):
  56         self.assertRaises(KeyError, lambda: self.store["a" * 40])
  57
  58     def test_contains_nonexistant(self):
  59         self.assertFalse(("a" * 40) in self.store)
  60
  61     def test_add_objects_empty(self):
  62         self.store.add_objects([])
  63
  64     def test_add_commit(self):
  65         # TODO: Argh, no way to construct Git commit objects without
  66         # access to a serialized form.
  67         self.store.add_objects([])
  68
  69     def test_add_object(self):
  70         self.store.add_object(testobject)
  71         self.assertEquals(set([testobject.id]), set(self.store))
  72         self.assertTrue(testobject.id in self.store)
  73         r = self.store[testobject.id]
  74         self.assertEquals(r, testobject)
  75
  76     def test_add_objects(self):
  77         data = [(testobject, "mypath")]
  78         self.store.add_objects(data)
  79         self.assertEquals(set([testobject.id]), set(self.store))
  80         self.assertTrue(testobject.id in self.store)
  81         r = self.store[testobject.id]
  82         self.assertEquals(r, testobject)
  83
  84     def test_iter_tree_contents(self):
  85         blob_a = make_object(Blob, data='a')
  86         blob_b = make_object(Blob, data='b')
  87         blob_c = make_object(Blob, data='c')
  88         for blob in [blob_a, blob_b, blob_c]:
  89             self.store.add_object(blob)
  90
  91         blobs = [
  92           ('a', blob_a.id, 0100644),
  93           ('ad/b', blob_b.id, 0100644),
  94           ('ad/bd/c', blob_c.id, 0100755),
  95           ('ad/c', blob_c.id, 0100644),
  96           ('c', blob_c.id, 0100644),
  97           ]
  98         tree_id = commit_tree(self.store, blobs)
  99         self.assertEquals([(p, m, h) for (p, h, m) in blobs],
 100                           list(self.store.iter_tree_contents(tree_id)))
 101
 102     def test_iter_tree_contents_include_trees(self):
 103         blob_a = make_object(Blob, data='a')
 104         blob_b = make_object(Blob, data='b')
 105         blob_c = make_object(Blob, data='c')
 106         for blob in [blob_a, blob_b, blob_c]:
 107             self.store.add_object(blob)
 108
 109         blobs = [
 110           ('a', blob_a.id, 0100644),
 111           ('ad/b', blob_b.id, 0100644),
 112           ('ad/bd/c', blob_c.id, 0100755),
 113           ]
 114         tree_id = commit_tree(self.store, blobs)
 115         tree = self.store[tree_id]
 116         tree_ad = self.store[tree['ad'][1]]
 117         tree_bd = self.store[tree_ad['bd'][1]]
 118
 119         expected = [
 120           ('', 0040000, tree_id),
 121           ('a', 0100644, blob_a.id),
 122           ('ad', 0040000, tree_ad.id),
 123           ('ad/b', 0100644, blob_b.id),
 124           ('ad/bd', 0040000, tree_bd.id),
 125           ('ad/bd/c', 0100755, blob_c.id),
 126           ]
 127         actual = self.store.iter_tree_contents(tree_id, include_trees=True)
 128         self.assertEquals(expected, list(actual))
 129
 130
 131 class MemoryObjectStoreTests(ObjectStoreTests, TestCase):
 132
 133     def setUp(self):
 134         TestCase.setUp(self)
 135         self.store = MemoryObjectStore()
 136
 137
 138 class PackBasedObjectStoreTests(ObjectStoreTests):
 139
 140     def test_empty_packs(self):
 141         self.assertEquals([], self.store.packs)
 142
 143     def test_pack_loose_objects(self):
 144         b1 = make_object(Blob, data="yummy data")
 145         self.store.add_object(b1)
 146         b2 = make_object(Blob, data="more yummy data")
 147         self.store.add_object(b2)
 148         self.assertEquals([], self.store.packs)
 149         self.assertEquals(2, self.store.pack_loose_objects())
 150         self.assertNotEquals([], self.store.packs)
 151         self.assertEquals(0, self.store.pack_loose_objects())
 152
 153
 154 class DiskObjectStoreTests(PackBasedObjectStoreTests, TestCase):
 155
 156     def setUp(self):
 157         TestCase.setUp(self)
 158         self.store_dir = tempfile.mkdtemp()
 159         self.store = DiskObjectStore.init(self.store_dir)
 160
 161     def tearDown(self):
 162         TestCase.tearDown(self)
 163         shutil.rmtree(self.store_dir)
 164
 165     def test_pack_dir(self):
 166         o = DiskObjectStore(self.store_dir)
 167         self.assertEquals(os.path.join(self.store_dir, "pack"), o.pack_dir)
 168
 169     def test_add_pack(self):
 170         o = DiskObjectStore(self.store_dir)
 171         f, commit = o.add_pack()
 172         b = make_object(Blob, data="more yummy data")
 173         write_pack_data(f, [(b, None)], 1)
 174         commit()
 175
 176     def test_add_thin_pack(self):
 177         o = DiskObjectStore(self.store_dir)
 178         f, commit = o.add_thin_pack()
 179         b = make_object(Blob, data="more yummy data")
 180         write_pack_data(f, [(b, None)], 1)
 181         commit()
 182
 183
 184 # TODO: MissingObjectFinderTests