code.grnet.gr Git - pithos/blob - pithos/backends/tests.py

   1 import unittest
   2 import os
   3 import types
   4 import json
   5
   6 from simple import SimpleBackend
   7
   8
   9 class TestAccount(unittest.TestCase):
  10     def setUp(self):
  11         self.basepath = './test/content'
  12         self.b = SimpleBackend(self.basepath)
  13         self.account = 'account1'
  14
  15     def tearDown(self):
  16         containers = self.b.list_containers(self.account)
  17         for container in containers:
  18             try:
  19                 self.b.delete_container(self.account, container)
  20             except IndexError:
  21                 # container not empty
  22                 for obj in self.b.list_objects(self.account, container):
  23                     self.b.delete_object(self.account, container, obj)
  24                 self.b.delete_container(self.account, container)
  25
  26     def test_list_containers(self):
  27         l1 = ['images', 'movies', 'documents', 'backups']
  28         for item in l1:
  29             self.b.put_container(self.account, item)
  30         l2 = self.b.list_containers(self.account)
  31         l1.sort()
  32         self.assertEquals(l1, l2)
  33
  34     def test_list_containers_with_limit_marker(self):
  35         l1 = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
  36         for item in l1:
  37             self.b.put_container(self.account, item)
  38         l2 = self.b.list_containers(self.account, limit=2)
  39         self.assertEquals(len(l2), 2)
  40         self.assertEquals(l1[:2], l2)
  41
  42         l2 = self.b.list_containers(self.account, limit=2, marker='bananas')
  43         self.assertTrue(len(l2) <= 2)
  44         self.assertEquals(l1[2:4], l2)
  45
  46         l2 = self.b.list_containers(self.account, limit=2, marker='oranges')
  47         self.assertTrue(len(l2) <= 2)
  48         self.assertEquals(l1[4:], l2)
  49
  50     def test_get_account_meta(self):
  51         meta = {
  52             "name": "account1",
  53             "username": "aaitest@uth.gr",
  54             "email": "aaitest@uth.gr",
  55             "fileroot": "http://hostname/gss/rest/aaitest@uth.gr/files",
  56             "trash": "http://hostname/gss/rest/aaitest@uth.gr/trash",
  57             "shared": "http://hostname/gss/rest/aaitest@uth.gr/shared",
  58             "others": "http://hostname/gss/rest/aaitest@uth.gr/others",
  59             "tags": "http://hostname/gss/rest/aaitest@uth.gr/tags",
  60             "groups": "http://hostname/gss/rest/aaitest@uth.gr/groups",
  61             "creationDate": 1223372769275,
  62             "modificationDate": 1223372769275,
  63             "lastLogin": 1223372769275}
  64         self.b.update_account_meta(self.account, meta)
  65         d = self.b.get_account_meta(self.account)
  66         for k,v in meta.iteritems():
  67             self.assertEquals(unicode(v), d[k])
  68
  69     def test_get_non_existing_account_meta(self):
  70         meta = self.b.get_account_meta('account2')
  71         self.assertEquals(meta, {'name': 'account2', 'count': 0, 'bytes': 0})
  72
  73     def test_update_account_meta(self):
  74         meta = {
  75             "name": "account1",
  76             "username": "aaitest@uth.gr",
  77             "email": "aaitest@uth.gr",
  78             "fileroot": "http://hostname/gss/rest/aaitest@uth.gr/files",
  79             "trash": "http://hostname/gss/rest/aaitest@uth.gr/trash",
  80             "shared": "http://hostname/gss/rest/aaitest@uth.gr/shared",
  81             "others": "http://hostname/gss/rest/aaitest@uth.gr/others",
  82             "tags": "http://hostname/gss/rest/aaitest@uth.gr/tags",
  83             "groups": "http://hostname/gss/rest/aaitest@uth.gr/groups",
  84             "creationDate": 1223372769275,
  85             "modificationDate": 1223372769275,
  86             "lastLogin": 1223372769275}
  87         self.b.update_account_meta(self.account, meta)
  88         p = os.path.join(self.basepath, self.account)
  89
  90         db_meta = self.b.get_account_meta(self.account)
  91         for k,v in meta.iteritems():
  92             self.assertTrue(k in db_meta)
  93             db_value = db_meta[k]
  94             self.assertEquals(unicode(v), db_value)
  95
  96 class TestContainer(unittest.TestCase):
  97     def setUp(self):
  98         self.basepath = './test/content'
  99         self.b = SimpleBackend(self.basepath)
 100         self.account = 'account1'
 101
 102     def tearDown(self):
 103         containers = self.b.list_containers(self.account)
 104         for container in containers:
 105             try:
 106                 self.b.delete_container(self.account, container)
 107             except IndexError: # container not empty
 108                 for obj in self.b.list_objects(self.account, container):
 109                     self.b.delete_object(self.account, container, obj)
 110                 self.b.delete_container(self.account, container)
 111
 112     def test_list_non_existing_account_objects(self):
 113         self.assertRaises(NameError, self.b.list_objects, 'account2', 'container1')
 114
 115     def test_list_objects(self):
 116         self.b.put_container(self.account, 'container1')
 117         objects = self.b.list_objects(self.account, 'container1')
 118         self.assertEquals(len([]), len(objects))
 119         l = [
 120             {'name':'kate_beckinsale.jpg'},
 121             {'name':'How To Win Friends And Influence People.pdf'},
 122             {'name':'moms_birthday.jpg'},
 123             {'name':'poodle_strut.mov'},
 124             {'name':'Disturbed - Down With The Sickness.mp3'},
 125             {'name':'army_of_darkness.avi'},
 126             {'name':'the_mad.avi'}
 127         ]
 128         for item in l:
 129             self.b.update_object_hashmap(self.account, 'container1', item['name'], 0, [])
 130         objects = self.b.list_objects(self.account, 'container1')
 131         self.assertEquals(len(l), len(objects))
 132
 133     def test_list_objects_with_limit_marker(self):
 134         self.b.put_container(self.account, 'container1')
 135         l = ['gala', 'grannysmith', 'honeycrisp', 'jonagold', 'reddelicious']
 136         for item in l:
 137             self.b.update_object_hashmap(self.account, 'container1', item, 0, [])
 138         objects = self.b.list_objects(self.account, 'container1', limit=2)
 139         self.assertEquals(l[:2], objects)
 140
 141         objects = self.b.list_objects(self.account, 'container1', limit=2, marker='grannysmith')
 142         self.assertEquals(l[2:4], objects)
 143
 144         objects = self.b.list_objects(self.account, 'container1', limit=2, marker='jonagold')
 145         self.assertEquals(l[4:], objects)
 146
 147     def test_list_pseudo_hierarchical_folders(self):
 148         self.b.put_container(self.account, 'container1')
 149         l = ['photos/animals/dogs/poodle.jpg',
 150              'photos/animals/dogs/terrier.jpg',
 151              'photos/animals/cats/persian.jpg',
 152              'photos/animals/cats/siamese.jpg',
 153              'photos/plants/fern.jpg',
 154              'photos/plants/rose.jpg',
 155              'photos/me.jpg'
 156              ]
 157         for item in l:
 158             self.b.update_object_hashmap(self.account, 'container1', item, 0, [])
 159
 160         objects = self.b.list_objects(self.account, 'container1', prefix='photos/', delimiter='/')
 161         self.assertEquals(['photos/animals/', 'photos/me.jpg', 'photos/plants/'], objects)
 162
 163         objects = self.b.list_objects(self.account, 'container1', prefix='photos/animals/', delimiter='/')
 164         self.assertEquals(['photos/animals/cats/', 'photos/animals/dogs/'], objects)
 165
 166         self.b.put_container(self.account, 'container2')
 167         l = ['photos/photo1', 'photos/photo2', 'movieobject', 'videos', 'videos/movieobj4']
 168         for item in l:
 169             self.b.update_object_hashmap(self.account, 'container2', item, 0, [])
 170         objects = self.b.list_objects(self.account, 'container2', delimiter='/')
 171         self.assertEquals(['movieobject', 'photos/', 'videos', 'videos/'], objects)
 172
 173     def test_put_container(self):
 174         cname = 'container1'
 175         self.b.put_container(self.account, cname)
 176         self.assertTrue(cname in self.b.list_containers(self.account))
 177
 178     def test_put_container_twice(self):
 179         cname = 'container1'
 180         self.b.put_container(self.account, cname)
 181         self.assertRaises(NameError, self.b.put_container, self.account, cname)
 182
 183     def test_delete_container(self):
 184         cname = 'container1'
 185         self.b.put_container(self.account, cname)
 186         self.b.delete_container(self.account, cname)
 187         self.assertTrue(cname not in self.b.list_containers(self.account))
 188
 189     def test_delete_non_exisitng_container(self):
 190         cname = 'container1'
 191         self.assertRaises(NameError, self.b.delete_container, self.account, cname)
 192
 193     def test_delete_non_empty_container(self):
 194         cname = 'container1'
 195         self.b.put_container(self.account, cname)
 196         self.b.update_object_hashmap(self.account, cname, 'object1', 0, [])
 197         self.assertRaises(IndexError, self.b.delete_container, self.account, cname)
 198
 199     def test_get_container_meta(self):
 200         cname = 'container1'
 201         self.b.put_container(self.account, cname)
 202         meta = self.b.get_container_meta(self.account, cname)
 203         self.assertEquals(meta['count'], 0)
 204
 205         l = ['photos/photo1', 'photos/photo2', 'movieobject', 'videos/movieobj4']
 206         for item in l:
 207             self.b.update_object_hashmap(self.account, cname, item, 0, [])
 208         meta = self.b.get_container_meta(self.account, cname)
 209         self.assertEquals(meta['count'], 4)
 210
 211 class TestObject(unittest.TestCase):
 212     def setUp(self):
 213         self.basepath = './test/content'
 214         self.b = SimpleBackend(self.basepath)
 215         self.account = 'account1'
 216
 217     def tearDown(self):
 218         containers = self.b.list_containers(self.account)
 219         for container in containers:
 220             try:
 221                 self.b.delete_container(self.account, container)
 222             except IndexError: # container not empty
 223                 for obj in self.b.list_objects(self.account, container):
 224                     self.b.delete_object(self.account, container, obj)
 225                 self.b.delete_container(self.account, container)
 226
 227     def test_get_non_existing_object(self):
 228         cname = 'container1'
 229         self.b.put_container(self.account, cname)
 230         self.assertRaises(NameError, self.b.get_object_hashmap, self.account, 'cname', 'testobj')
 231         self.assertRaises(NameError, self.b.get_object_hashmap, self.account, cname, 'testobj')
 232
 233     def test_get_object(self):
 234         cname = 'container1'
 235         self.b.put_container(self.account, cname)
 236         input = {'name':'kate_beckinsale.jpg'}
 237         data = json.dumps(input)
 238         hash = self.b.put_block(data)
 239         self.b.update_object_hashmap(self.account, cname, input['name'], len(data), [hash])
 240         size, hashmap = self.b.get_object_hashmap(self.account, cname, 'kate_beckinsale.jpg')
 241         self.assertEquals(len(data), size)
 242         self.assertEquals(hash, hashmap[0])
 243         self.assertEquals(input, json.loads(self.b.get_block(hash)))
 244
 245 #     def test_update_object(self):
 246 #         cname = 'container1'
 247 #         self.b.put_container(self.account, cname)
 248 #         input = {'name':'kate_beckinsale.jpg'}
 249 #         self.b.update_object(self.account, cname, input['name'], json.dumps(input))
 250 #         meta = self.b.get_object_meta(self.account, cname, input['name'])
 251
 252     def test_copy_object(self):
 253         src_cname = 'container1'
 254         src_obj = 'photos/me.jpg'
 255         dest_cname = 'container2'
 256         dest_obj = 'photos/personal/myself.jpg'
 257
 258         # non existing source account
 259         self.assertRaises(NameError,
 260                           self.b.copy_object,
 261                           'account',
 262                           src_cname,
 263                           src_obj,
 264                           dest_cname,
 265                           dest_obj)
 266         # non existing source container
 267         self.assertRaises(NameError,
 268                           self.b.copy_object,
 269                           self.account,
 270                           src_cname,
 271                           src_obj,
 272                           dest_cname,
 273                           dest_obj)
 274
 275         self.b.put_container(self.account, src_cname)
 276         # non existing source object
 277         self.assertRaises(NameError,
 278                           self.b.copy_object,
 279                           self.account,
 280                           src_cname,
 281                           src_obj,
 282                           dest_cname,
 283                           dest_obj)
 284
 285         self.b.update_object_hashmap(self.account, src_cname, src_obj, 0, [])
 286         # non existing destination container
 287         self.assertRaises(NameError,
 288                           self.b.copy_object,
 289                           self.account,
 290                           src_cname,
 291                           src_obj,
 292                           dest_cname,
 293                           dest_obj)
 294
 295         self.b.put_container(self.account, dest_cname)
 296         self.b.update_object_meta(self.account, src_cname, src_obj, {'tag':'sfsfssf'})
 297         self.b.copy_object(self.account, src_cname, src_obj, dest_cname, dest_obj)
 298         self.assertTrue(dest_obj in self.b.list_objects(self.account,
 299                                                         dest_cname,
 300                                                         prefix='photos/personal/',
 301                                                         delimiter='/'))
 302         # TODO: test metadata changes
 303         meta_tag = self.b.get_object_meta(self.account, dest_cname, dest_obj)['tag']
 304         self.assertEquals(meta_tag, unicode('sfsfssf'))
 305
 306     def test_delete_non_existing_object(self):
 307         cname = 'container1'
 308         self.b.put_container(self.account, cname)
 309         name = 'kate_beckinsale.jpg'
 310         self.assertRaises(NameError, self.b.delete_object, self.account, cname, name)
 311
 312     def test_delete_object(self):
 313         cname = 'container1'
 314         self.b.put_container(self.account, cname)
 315         name = 'kate_beckinsale.jpg'
 316         self.b.update_object_hashmap(self.account, cname, name, 0, [])
 317         self.assertTrue(name in self.b.list_objects(self.account, cname))
 318
 319         self.b.delete_object(self.account, cname, name)
 320         self.assertTrue(name not in self.b.list_objects(self.account, cname))
 321         self.assertRaises(NameError, self.b.delete_object, self.account, cname, name)
 322
 323     def test_get_non_existing_object_meta(self):
 324         cname = 'container1'
 325         self.b.put_container(self.account, cname)
 326         name = 'kate_beckinsale.jpg'
 327         self.assertRaises(NameError, self.b.get_object_meta, self.account, cname, name)
 328
 329     def test_get_update_object_meta(self):
 330         cname = 'container1'
 331         self.b.put_container(self.account, cname)
 332         name = 'kate_beckinsale.jpg'
 333         self.b.update_object_hashmap(self.account, cname, name, 0, [])
 334
 335         m1 = {'X-Object-Meta-Meat': 'Bacon',
 336              'X-Object-Meta-Fruit': 'Bacon',
 337              'X-Object-Meta-Dairy': 'Bacon'}
 338         self.b.update_object_meta(self.account, cname, name, m1)
 339         meta = self.b.get_object_meta(self.account, cname, name)
 340         for k,v in m1.iteritems():
 341             self.assertTrue(k in meta)
 342             self.assertEquals(unicode(v), meta[k])
 343
 344         m2 = {'X-Object-Meta-Meat': 'Bacon',
 345              'X-Object-Meta-Fruit': 'Bacon',
 346              'X-Object-Meta-Veggie': 'Bacon',
 347              'X-Object-Meta-Dairy': 'Chicken'}
 348         self.b.update_object_meta(self.account, cname, name, m2)
 349         meta = self.b.get_object_meta(self.account, cname, name)
 350         m1.update(m2)
 351         for k,v in m1.iteritems():
 352             self.assertTrue(k in meta)
 353             self.assertEquals(unicode(v), meta[k])
 354
 355     def test_update_non_existing_object_meta(self):
 356         cname = 'container1'
 357         self.b.put_container(self.account, cname)
 358         name = 'kate_beckinsale.jpg'
 359         self.assertRaises(NameError, self.b.update_object_meta, self.account, cname, name, {})
 360
 361 if __name__ == "__main__":
 362     unittest.main()