# or implied, of GRNET S.A.
from pithos.lib.client import Pithos_Client, Fault
-import unittest
from django.utils import simplejson as json
from xml.dom import minidom
from StringIO import StringIO
+import unittest
import time as _time
import types
import hashlib
self.containers = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
for item in self.containers:
self.client.create_container(item)
+
+ #keep track of initial account groups
+ self.initial_groups = self.client.retrieve_account_groups()
+
+ #keep track of initial account meta
+ self.initial_meta = self.client.retrieve_account_metadata(restricted=True)
+
+ meta = {'foo':'bar'}
+ self.client.update_account_metadata(**meta)
+ self.updated_meta = self.initial_meta.update(meta)
def tearDown(self):
- self.client.delete_account_metadata(['foo'])
+ #delete additionally created meta
+ l = []
+ for m in self.client.retrieve_account_metadata(restricted=True):
+ if m not in self.initial_meta:
+ l.append(m)
+ self.client.delete_account_metadata(l)
+
+ #delete additionally created groups
+ l = []
+ for g in self.client.retrieve_account_groups():
+ if g not in self.initial_groups:
+ l.append(g)
+ self.client.unset_account_groups(l)
+
+ #print '#', self.client.retrieve_account_groups()
+ #print '#', self.client.retrieve_account_metadata(restricted=True)
BaseTestCase.tearDown(self)
def test_get_account_meta(self):
past = t - datetime.timedelta(minutes=-15)
past = int(_time.mktime(past.timetuple()))
- meta = {'foo':'bar'}
+ meta = {'premium':True}
self.client.update_account_metadata(**meta)
meta = self.client.retrieve_account_metadata(restricted=True,
until=past)
- self.assertTrue('foo' not in meta)
+ self.assertTrue('premium' not in meta)
meta = self.client.retrieve_account_metadata(restricted=True)
- self.assertTrue('foo' in meta)
+ self.assertTrue('premium' in meta)
def test_get_account_meta_until_invalid_date(self):
- meta = {'foo':'bar'}
+ meta = {'premium':True}
self.client.update_account_metadata(**meta)
meta = self.client.retrieve_account_metadata(restricted=True,
until='kshfksfh')
- self.assertTrue('foo' in meta)
+ self.assertTrue('premium' in meta)
class AccountGet(BaseTestCase):
def setUp(self):
self.assertEqual(meta, self.client.retrieve_account_metadata(restricted=True))
- #def test_delete_meta(self):
- # with AssertMappingInvariant(self.client.reset_account_groups):
- # meta = {'test':'test', 'tost':'tost'}
- # self.client.update_account_metadata(**meta)
- #
- # self.client.delete_account_metadata(**meta)
+ def test_delete_meta(self):
+ with AssertMappingInvariant(self.client.retrieve_account_groups):
+ meta = {'test':'test', 'tost':'tost'}
+ self.client.update_account_metadata(**meta)
+
+ self.client.delete_account_metadata(meta.keys())
+
+ account_meta = self.client.retrieve_account_metadata(restricted=True)
+ for m in meta:
+ self.assertTrue(m not in account_meta.keys())
def test_set_account_groups(self):
with AssertMappingInvariant(self.client.retrieve_account_metadata):
#assert content-type
self.assertEqual(h['content-type'], o['meta']['content_type'])
+ def test_upload_with_name_containing_slash(self):
+ name = '/%s' % o_names[0]
+ meta = {'test':'test1'}
+ o = self.upload_random_data(self.container, name, **meta)
+
+ self.assertEqual(o['data'],
+ self.client.retrieve_object(self.container, name))
+
+ self.assertTrue(name in self.client.list_objects(self.container))
+
+ def test_create_directory_marker(self):
+ self.client.create_directory_marker(self.container, 'foo')
+ meta = self.client.retrieve_object_metadata(self.container, 'foo')
+ self.assertEqual(meta['content-length'], '0')
+ self.assertEqual(meta['content-type'], 'application/directory')
+
def test_upload_unprocessable_entity(self):
meta={'etag':'123', 'test':'test1'}
self.obj['name'],
self.containers[0],
'testcopy',
- **meta)[0]
+ meta)[0]
#assert copy success
self.assertEqual(status, 201)
self.obj['name'],
self.containers[1],
'testcopy',
- **meta)[0]
+ meta)[0]
self.assertEqual(status, 201)
# assert updated metadata
#copy from invalid object
meta = {'test':'testcopy'}
self.assert_raises_fault(404, self.client.copy_object, self.containers[0],
- 'test.py', self.containers[1], 'testcopy',
- **meta)
+ 'test.py', self.containers[1], 'testcopy', meta)
#copy from invalid container
meta = {'test':'testcopy'}
self.assert_raises_fault(404, self.client.copy_object, self.containers[1],
self.obj['name'], self.containers[1],
- 'testcopy', **meta)
+ 'testcopy', meta)
-class ObjectMove(ObjectCopy):
+class ObjectMove(BaseTestCase):
+ def setUp(self):
+ BaseTestCase.setUp(self)
+ self.account = 'test'
+ self.containers = ['c1', 'c2']
+ for c in self.containers:
+ self.client.create_container(c)
+ self.obj = self.upload_random_data(self.containers[0], o_names[0])
+
def test_move(self):
#perform move
meta = {'test':'testcopy'}
src_path = os.path.join('/', self.containers[0], self.obj['name'])
status = self.client.move_object(self.containers[0], self.obj['name'],
self.containers[0], 'testcopy',
- **meta)[0]
+ meta)[0]
#assert successful move
self.assertEqual(status, 201)
return self.post(path, data, headers=headers)
def _change_obj_location(self, src_container, src_object, dst_container,
- dst_object, remove=False, public=False, **meta):
+ dst_object, remove=False, meta={}, **headers):
path = '/%s/%s' % (dst_container, dst_object)
- headers = {}
+ headers = {} if not headers else headers
for k, v in meta.items():
headers['x-object-meta-%s' % k] = v
if remove:
headers['x-move-from'] = '/%s/%s' % (src_container, src_object)
else:
headers['x-copy-from'] = '/%s/%s' % (src_container, src_object)
- self._set_public_header(headers, public)
- self.headers = headers if headers else None
- headers['content-length'] = 0
+ headers['content_length'] = 0
return self.put(path, headers=headers)
- def copy_object(self, src_container, src_object, dst_container,
- dst_object, public=False, **meta):
+ def copy_object(self, src_container, src_object, dst_container, dst_object,
+ meta, **headers):
+ """copies an object"""
return self._change_obj_location(src_container, src_object,
- dst_container, dst_object, False,
- public, **meta)
+ dst_container, dst_object, remove=False,
+ meta=meta, **headers)
def move_object(self, src_container, src_object, dst_container,
- dst_object, public=False, **meta):
+ dst_object, meta={}, **headers):
+ """moves an object"""
return self._change_obj_location(src_container, src_object,
- dst_container, dst_object, True,
- public, **meta)
+ dst_container, dst_object, remove=True,
+ meta=meta, **headers)
def delete_object(self, container, object, params={}):
+ """deletes an object"""
return self.delete('/%s/%s' % (container, object), params=params)
def retrieve_object_metadata(self, container, object, restricted=False,
"""restores a trashed object"""
return self.delete_object_metadata(container, object, ['trash'])
- def _set_public_header(self, headers, public=False):
- """sets the public header"""
- if not headers:
- headers = {}
- if public == None:
- return
- else:
- headers['x-object-public'] = public if public else ''
-
def publish_object(self, container, object):
"""sets a previously created object publicly accessible"""
path = '/%s/%s' % (container, object)
headers = {'content_range':'bytes */*'}
- self._set_public_header(headers, public=True)
+ headers['x_object_public'] = True
return self.post(path, headers=headers)
def unpublish_object(self, container, object):
"""unpublish an object"""
path = '/%s/%s' % (container, object)
headers = {'content_range':'bytes */*'}
- self._set_public_header(headers, public=False)
+ headers['x_object_public'] = False
return self.post(path, headers=headers)
+
+ def _change_obj_location(self, src_container, src_object, dst_container,
+ dst_object, remove=False, meta={}, **headers):
+ path = '/%s/%s' % (dst_container, dst_object)
+ headers = {} if not headers else headers
+ for k, v in meta.items():
+ headers['x-object-meta-%s' % k] = v
+ if remove:
+ headers['x-move-from'] = '/%s/%s' % (src_container, src_object)
+ else:
+ headers['x-copy-from'] = '/%s/%s' % (src_container, src_object)
+ headers['content_length'] = 0
+ return self.put(path, headers=headers)
+
+ def copy_object(self, src_container, src_object, dst_container, dst_object,
+ meta={}, public=False, version=None):
+ """copies an object"""
+ headers = {}
+ headers['x_object_public'] = public
+ if version:
+ headers['x_object_version'] = version
+ return OOS_Client.copy_object(self, src_container, src_object,
+ dst_container, dst_object, meta=meta,
+ **headers)
+
+ def move_object(self, src_container, src_object, dst_container,
+ dst_object, meta={}, public=False, version=None):
+ """moves an object"""
+ headers = {}
+ headers['x_object_public'] = public
+ if version:
+ headers['x_object_version'] = version
+ return OOS_Client.move_object(self, src_container, src_object,
+ dst_container, dst_object, meta=meta,
+ **headers)
\ No newline at end of file
if getattr(self, 'until'):
t = _time.strptime(self.until, self.format)
args['until'] = int(_time.mktime(t))
-
+
if object:
meta = self.client.retrieve_object_metadata(container, object,
self.restricted,
help='file descriptor to read from (pass - for standard input)')
parser.add_option('--public', action='store_true',
dest='x_object_public', default=False,
- help='make object publicly accessible (\'True\'/\'False\')')
+ help='make object publicly accessible')
def execute(self, path, *args):
if path.find('=') != -1:
@cli_command('copy', 'cp')
class CopyObject(Command):
- syntax = '<src container>/<src object> [<dst container>/]<dst object>'
+ syntax = '<src container>/<src object> [<dst container>/]<dst object> [key=val] [...]'
description = 'copy an object to a different location'
def add_options(self, parser):
parser.add_option('--version', action='store',
dest='version', default=False,
help='copy specific version')
- parser.add_option('--public', action='store',
- dest='public', default=None,
- help='publish/unpublish object (\'True\'/\'False\')')
+ parser.add_option('--public', action='store_true',
+ dest='public', default=False,
+ help='make object publicly accessible')
- def execute(self, src, dst):
+ def execute(self, src, dst, *args):
src_container, sep, src_object = src.partition('/')
dst_container, sep, dst_object = dst.partition('/')
+
+ #prepare user defined meta
+ meta = {}
+ for arg in args:
+ key, sep, val = arg.partition('=')
+ meta[key] = val
+
if not sep:
dst_container = src_container
dst_object = dst
- version = getattr(self, 'version')
- headers = None
- if version:
- headers = {}
- headers['X_SOURCE_VERSION'] = version
- if self.public and self.nopublic:
- raise Fault('Conflicting options')
- if self.public not in ['True', 'False', None]:
- raise Fault('Not acceptable value for public')
- public = eval(self.public) if self.public else None
+
self.client.copy_object(src_container, src_object, dst_container,
- dst_object, public, headers)
+ dst_object, meta, self.public, self.version, **meta)
@cli_command('set')
class SetMeta(Command):
parser.add_option('-f', action='store',
dest='srcpath', default=None,
help='file descriptor to read from: pass - for standard input')
- parser.add_option('--public', action='store',
+ parser.add_option('--public', action='store_true',
dest='x_object_public', default=False,
- help='publish/unpublish object (\'True\'/\'False\')')
+ help='make object publicly accessible')
def execute(self, path, *args):
if path.find('=') != -1:
description = 'move an object to a different location'
def add_options(self, parser):
- parser.add_option('--public', action='store',
- dest='public', default=None,
- help='publish/unpublish object (\'True\'/\'False\')')
+ parser.add_option('--version', action='store',
+ dest='version', default=None,
+ help='move a specific object version')
+ parser.add_option('--public', action='store_true',
+ dest='public', default=False,
+ help='make object publicly accessible')
- def execute(self, src, dst):
+ def execute(self, src, dst, *args):
src_container, sep, src_object = src.partition('/')
dst_container, sep, dst_object = dst.partition('/')
if not sep:
dst_container = src_container
dst_object = dst
- if self.public not in ['True', 'False', None]:
- raise Fault('Not acceptable value for public')
- public = eval(self.public) if self.public else None
+
+ #prepare user defined meta
+ meta = {}
+ for arg in args:
+ key, sep, val = arg.partition('=')
+ meta[key] = val
+
self.client.move_object(src_container, src_object, dst_container,
- dst_object, public, headers)
+ dst_object, meta, self.public, self.version)
@cli_command('remove')
class TrashObject(Command):
cmd = cls(name, argv[2:])
- #cmd.execute(*cmd.args)
- try:
- cmd.execute(*cmd.args)
- except TypeError, e:
- cmd.parser.print_help()
- exit(1)
- except Fault, f:
- status = f.status and '%s ' % f.status or ''
- print '%s%s' % (status, f.data)
+ cmd.execute(*cmd.args)
+ #try:
+ # cmd.execute(*cmd.args)
+ #except TypeError, e:
+ # cmd.parser.print_help()
+ # exit(1)
+ #except Fault, f:
+ # status = f.status and '%s ' % f.status or ''
+ # print '%s%s' % (status, f.data)
if __name__ == '__main__':
main()
projects / pithos / commitdiff