import types\r
import json\r
\r
-class TestBackend(unittest.TestCase):\r
+class TestAccount(unittest.TestCase):\r
def setUp(self):\r
self.basepath = './test/content'\r
- log_file = './test_backend.log'\r
- self.b = BackEnd(self.basepath, log_file)\r
+ self.b = BackEnd(self.basepath)\r
+ self.account = 'account1'\r
\r
def tearDown(self):\r
- # clear fs\r
- for root, dirs, files in os.walk(self.basepath, topdown=False):\r
- for name in files: \r
- os.remove(os.path.join(root, name))\r
- for name in dirs:\r
- os.rmdir(os.path.join(root, name))\r
- os.removedirs(self.basepath)\r
+ containers = self.b.list_containers(self.account)\r
+ for container in containers:\r
+ try:\r
+ self.b.delete_container(self.account, container)\r
+ except Exception: # container not empty\r
+ for obj in self.b.list_objects(self.account, container):\r
+ self.b.delete_object(self.account, container, obj)\r
+ self.b.delete_container(self.account, container)\r
+ \r
+ def test_list_containers(self):\r
+ l1 = ['images', 'movies', 'documents', 'backups']\r
+ for item in l1:\r
+ self.b.create_container(self.account, item)\r
+ l2 = self.b.list_containers(self.account)\r
+ l1.sort()\r
+ self.assertEquals(l1, l2)\r
+ \r
+ def test_list_containers_with_limit_marker(self):\r
+ l1 = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']\r
+ for item in l1:\r
+ self.b.create_container(self.account, item)\r
+ l2 = self.b.list_containers(self.account, limit=2)\r
+ self.assertEquals(len(l2), 2)\r
+ self.assertEquals(l1[:2], l2)\r
+ \r
+ l2 = self.b.list_containers(self.account, limit=2, marker='bananas')\r
+ self.assertTrue(len(l2) <= 2)\r
+ self.assertEquals(l1[2:4], l2)\r
+\r
+ l2 = self.b.list_containers(self.account, limit=2, marker='oranges')\r
+ self.assertTrue(len(l2) <= 2)\r
+ self.assertEquals(l1[4:], l2)\r
\r
def test_get_account_meta(self):\r
- account = 'account1'\r
meta = {\r
"name": "account1",\r
"username": "aaitest@uth.gr",\r
"lastLogin": 1223372769275,\r
"quota": {"totalFiles": 7, "totalBytes": 429330,"bytesRemaining": 10736988910}\r
}\r
- self.b.update_account_meta(account, meta)\r
- d = self.b.get_account_meta(account)\r
- p = os.path.join(self.basepath, account)\r
+ self.b.update_account_meta(self.account, meta)\r
+ d = self.b.get_account_meta(self.account)\r
+ p = os.path.join(self.basepath, self.account)\r
self.assertEquals(d['count'], len(os.listdir(p)))\r
self.assertEquals(d['bytes'], os.stat(p).st_size)\r
- self.assertEquals(d['name'], account)\r
+ self.assertEquals(d['name'], self.account)\r
exp_len = len(meta)+3\r
if meta.has_key('count'):\r
exp_len = exp_len - 1\r
self.assertEquals(len(d), exp_len)\r
\r
def test_get_non_existing_account_meta(self):\r
- account = 'account1'\r
- self.assertRaises(NameError, self.b.get_account_meta, account)\r
+ self.assertRaises(NameError, self.b.get_account_meta, 'account2')\r
\r
def test_update_account_meta(self):\r
- account = 'account1'\r
meta = {\r
"name": "account1",\r
"username": "aaitest@uth.gr",\r
"lastLogin": 1223372769275,\r
"quota": {"totalFiles": 7, "totalBytes": 429330,"bytesRemaining": 10736988910}\r
}\r
- self.b.update_account_meta(account, meta)\r
- p = os.path.join(self.basepath, account)\r
+ self.b.update_account_meta(self.account, meta)\r
+ p = os.path.join(self.basepath, self.account)\r
self.assertTrue(os.path.exists(p))\r
\r
- db_meta = self.b.get_account_meta(account)\r
+ db_meta = self.b.get_account_meta(self.account)\r
for k,v in meta.iteritems():\r
self.assertTrue(k in db_meta)\r
db_value = db_meta[k]\r
if type(v) != types.StringType:\r
db_value = json.loads(db_value)\r
self.assertEquals(v, db_value)\r
+\r
+class TestContainer(unittest.TestCase):\r
+ def setUp(self):\r
+ self.basepath = './test/content'\r
+ self.b = BackEnd(self.basepath)\r
+ self.account = 'account1'\r
+ \r
+ def tearDown(self):\r
+ containers = self.b.list_containers(self.account)\r
+ for container in containers:\r
+ try:\r
+ self.b.delete_container(self.account, container)\r
+ except Exception: # container not empty\r
+ for obj in self.b.list_objects(self.account, container):\r
+ self.b.delete_object(self.account, container, obj)\r
+ self.b.delete_container(self.account, container)\r
+ \r
+ def test_list_non_existing_account_objects(self):\r
+ self.assertRaises(NameError, self.b.list_objects, 'account2', 'container1')\r
+ \r
+ def test_list_objects(self):\r
+ self.b.create_container(self.account, 'container1')\r
+ objects = self.b.list_objects(self.account, 'container1')\r
+ self.assertEquals(len([]), len(objects))\r
+ l = [\r
+ {'name':'kate_beckinsale.jpg'},\r
+ {'name':'How To Win Friends And Influence People.pdf'},\r
+ {'name':'moms_birthday.jpg'},\r
+ {'name':'poodle_strut.mov'},\r
+ {'name':'Disturbed - Down With The Sickness.mp3'},\r
+ {'name':'army_of_darkness.avi'},\r
+ {'name':'the_mad.avi'}\r
+ ]\r
+ for item in l:\r
+ self.b.update_object(self.account, 'container1', item['name'], json.dumps(item))\r
+ objects = self.b.list_objects(self.account, 'container1')\r
+ self.assertEquals(len(l), len(objects))\r
+ \r
+ def test_list_objects_with_limit_marker(self):\r
+ self.b.create_container(self.account, 'container1')\r
+ l = ['gala', 'grannysmith', 'honeycrisp', 'jonagold', 'reddelicious']\r
+ for item in l:\r
+ self.b.update_object(self.account, 'container1', item, item)\r
+ objects = self.b.list_objects(self.account, 'container1', limit=2)\r
+ self.assertEquals(l[:2], objects)\r
+ \r
+ objects = self.b.list_objects(self.account, 'container1', limit=2, marker='grannysmith')\r
+ self.assertEquals(l[2:4], objects)\r
+ \r
+ objects = self.b.list_objects(self.account, 'container1', limit=2, marker='jonagold')\r
+ self.assertEquals(l[4:], objects)\r
+ \r
+ def test_list_pseudo_hierarchical_folders(self):\r
+ self.b.create_container(self.account, 'container1')\r
+ l = ['photos/animals/dogs/poodle.jpg',\r
+ 'photos/animals/dogs/terrier.jpg',\r
+ 'photos/animals/cats/persian.jpg',\r
+ 'photos/animals/cats/siamese.jpg',\r
+ 'photos/plants/fern.jpg',\r
+ 'photos/plants/rose.jpg',\r
+ 'photos/me.jpg'\r
+ ]\r
+ for item in l:\r
+ self.b.update_object(self.account, 'container1', item, item)\r
+ \r
+ objects = self.b.list_objects(self.account, 'container1', prefix='photos/', delimiter='/')\r
+ self.assertEquals(['animals', 'me.jpg', 'plants'], objects)\r
+ \r
+ objects = self.b.list_objects(self.account, 'container1', prefix='photos/animals/', delimiter='/')\r
+ self.assertEquals(['cats', 'dogs'], objects)\r
+ \r
+ self.b.create_container(self.account, 'container2')\r
+ l = ['photos/photo1', 'photos/photo2', 'movieobject', 'videos/movieobj4']\r
+ for item in l:\r
+ self.b.update_object(self.account, 'container2', item, item)\r
+ objects = self.b.list_objects(self.account, 'container2', delimiter='/')\r
+ self.assertEquals(['movieobject', 'photos', 'videos'], objects) \r
\r
def test_create_container(self):\r
- account = 'account1'\r
cname = 'container1' \r
- self.b.create_container(account, cname)\r
- fullpath = os.path.join(self.basepath, account, cname)\r
- self.assertTrue(os.path.exists(fullpath))\r
- \r
+ self.b.create_container(self.account, cname)\r
+ self.assertTrue(cname in self.b.list_containers(self.account))\r
+ \r
def test_create_container_twice(self):\r
- account = 'account1'\r
cname = 'container1'\r
- self.b.create_container(account, cname)\r
- self.assertRaises(NameError, self.b.create_container, account, cname)\r
+ self.b.create_container(self.account, cname)\r
+ self.assertRaises(NameError, self.b.create_container, self.account, cname)\r
\r
def test_delete_container(self):\r
- account = 'account1'\r
cname = 'container1'\r
- self.b.create_container(account, cname)\r
- self.b.delete_container(account, cname)\r
- self.assertTrue(not os.path.exists(os.path.join(account, cname)))\r
- path = os.path.join(account, cname)\r
- c = self.b.con.execute('select * from metadata where object_id in (select rowid from objects where name = ''?'')', (path,))\r
- self.assertEqual(c.rowcount, -1)\r
- \r
- c = self.b.con.execute('select * from objects where name = ''?''', (path,))\r
- self.assertEqual(c.rowcount, -1)\r
+ self.b.create_container(self.account, cname)\r
+ self.b.delete_container(self.account, cname)\r
+ self.assertTrue(cname not in self.b.list_containers(self.account))\r
\r
def test_delete_non_exisitng_container(self):\r
- account = 'account1'\r
cname = 'container1'\r
- self.assertRaises(NameError, self.b.delete_container, account, cname)\r
+ self.assertRaises(NameError, self.b.delete_container, self.account, cname)\r
\r
def test_delete_non_empty_container(self):\r
- account = 'account1'\r
cname = 'container1'\r
- self.b.create_container(account, cname)\r
- self.b.update_object(account, cname, 'object1', 'alkhadlkhalkdhal')\r
- self.assertRaises(Exception, self.b.delete_container, account, cname)\r
+ self.b.create_container(self.account, cname)\r
+ self.b.update_object(self.account, cname, 'object1', 'alkhadlkhalkdhal')\r
+ self.assertRaises(Exception, self.b.delete_container, self.account, cname)\r
+ \r
+ def test_get_container_meta(self):\r
+ cname = 'container1'\r
+ self.b.create_container(self.account, cname)\r
+ meta = self.b.get_container_meta(self.account, cname)\r
+ self.assertEquals(meta['count'], 0)\r
+ \r
+ l = ['photos/photo1', 'photos/photo2', 'movieobject', 'videos/movieobj4']\r
+ for item in l:\r
+ self.b.update_object(self.account, cname, item, item)\r
+ meta = self.b.get_container_meta(self.account, cname)\r
+ self.assertEquals(meta['count'], 4)\r
\r
if __name__ == "__main__":\r
unittest.main()
\ No newline at end of file
projects / pithos / commitdiff