/ - Diff - Synnefo - Greek Research and Technology Network's projects

Revision f53483d8

     # interpreted as representing official policies, either expressed
     # or implied, of GRNET S.A.
     from urlparse import urlunsplit, urlsplit
     from urlparse import urlunsplit, urlsplit, urlparse
     from xml.dom import minidom
     from urllib import quote, unquote
-...
     from synnefo.lib import join_urls
     from django.test import TestCase
     from django.test.client import Client, MULTIPART_CONTENT, FakePayload
     from django.test.simple import DjangoTestSuiteRunner
     from django.conf import settings
     from django.utils.http import urlencode
-...
         'django.db.backends.oracle': 'oracle'}
     def prepate_db_connection():
     def prepare_db_connection():
         """Build pithos backend connection string from django default database"""
         db = settings.DATABASES['default']
-...
         def setup_databases(self, **kwargs):
             old_names, mirrors = super(PithosTestSuiteRunner,
                                        self).setup_databases(**kwargs)
             prepate_db_connection()
             prepare_db_connection()
             return old_names, mirrors
         def teardown_databases(self, old_config, **kwargs):
-...
                                                                   **kwargs)
     class PithosTestClient(Client):
         def copy(self, path, data={}, content_type=MULTIPART_CONTENT,
                  follow=False, **extra):
             """
             Send a resource to the server using COPY.
             """
             parsed = urlparse(path)
             r = {
                 'CONTENT_TYPE':    'text/html; charset=utf-8',
                 'PATH_INFO':       self._get_path(parsed),
                 'QUERY_STRING':    urlencode(data, doseq=True) or parsed[4],
                 'REQUEST_METHOD': 'COPY',
                 'wsgi.input':      FakePayload('')
+            }
             r.update(extra)
             response = self.request(**r)
             if follow:
                 response = self._handle_redirects(response, **extra)
             return response
         def move(self, path, data={}, content_type=MULTIPART_CONTENT,
                  follow=False, **extra):
             """
             Send a resource to the server using MOVE.
             """
             parsed = urlparse(path)
             r = {
                 'CONTENT_TYPE':    'text/html; charset=utf-8',
                 'PATH_INFO':       self._get_path(parsed),
                 'QUERY_STRING':    urlencode(data, doseq=True) or parsed[4],
                 'REQUEST_METHOD': 'MOVE',
                 'wsgi.input':      FakePayload('')
+            }
             r.update(extra)
             response = self.request(**r)
             if follow:
                 response = self._handle_redirects(response, **extra)
             return response
     class PithosAPITest(TestCase):
         def setUp(self):
             self.client = PithosTestClient()
             # Override default block size to spead up tests
             pithos_settings.BACKEND_BLOCK_SIZE = TEST_BLOCK_SIZE
             pithos_settings.BACKEND_HASH_ALGORITHM = TEST_HASH_ALGORITHM
-...
                 self.delete_container_content(c['name'])
                 self.delete_container(c['name'])
         def head(self, url, user='user', data={}, follow=False, **extra):
         def head(self, url, user='user', token='DummyToken', data={}, follow=False,
                  **extra):
             with astakos_user(user):
                 extra = dict((quote(k), quote(v)) for k, v in extra.items())
                 extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
                 if token:
                     extra['HTTP_X_AUTH_TOKEN'] = token
                 response = self.client.head(url, data, follow, **extra)
             return response
         def get(self, url, user='user', data={}, follow=False, **extra):
         def get(self, url, user='user', token='DummyToken', data={}, follow=False,
                 **extra):
             with astakos_user(user):
                 extra = dict((quote(k), quote(v)) for k, v in extra.items())
                 extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
                 if token:
                     extra['HTTP_X_AUTH_TOKEN'] = token
                 response = self.client.get(url, data, follow, **extra)
             return response
         def delete(self, url, user='user', data={}, follow=False, **extra):
         def delete(self, url, user='user', token='DummyToken', data={},
                    follow=False, **extra):
             with astakos_user(user):
                 extra = dict((quote(k), quote(v)) for k, v in extra.items())
                 extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
                 if token:
                     extra['HTTP_X_AUTH_TOKEN'] = token
                 response = self.client.delete(url, data, follow, **extra)
             return response
         def post(self, url, user='user', data={},
         def post(self, url, user='user', token='DummyToken', data={},
                  content_type='application/octet-stream', follow=False, **extra):
             with astakos_user(user):
                 extra = dict((quote(k), quote(v)) for k, v in extra.items())
                 extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
                 if token:
                     extra['HTTP_X_AUTH_TOKEN'] = token
                 response = self.client.post(url, data, content_type, follow,
                                             **extra)
             return response
         def put(self, url, user='user', data={},
         def put(self, url, user='user', token='DummyToken', data={},
                 content_type='application/octet-stream', follow=False, **extra):
             with astakos_user(user):
                 extra = dict((quote(k), quote(v)) for k, v in extra.items())
                 extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
                 if token:
                     extra['HTTP_X_AUTH_TOKEN'] = token
                 response = self.client.put(url, data, content_type, follow,
                                            **extra)
             return response
         def copy(self, url, user='user', token='DummyToken', data={},
                  content_type='application/octet-stream', follow=False, **extra):
             with astakos_user(user):
                 extra = dict((quote(k), quote(v)) for k, v in extra.items())
                 if token:
                     extra['HTTP_X_AUTH_TOKEN'] = token
                 response = self.client.copy(url, data, content_type, follow,
                                             **extra)
             return response
         def move(self, url, user='user', token='DummyToken', data={},
                  content_type='application/octet-stream', follow=False, **extra):
             with astakos_user(user):
                 extra = dict((quote(k), quote(v)) for k, v in extra.items())
                 if token:
                     extra['HTTP_X_AUTH_TOKEN'] = token
                 response = self.client.move(url, data, content_type, follow,
                                             **extra)
             return response
         def update_account_meta(self, meta, user=None):
             user = user or self.user
             kwargs = dict(
-...
             url = join_urls(self.pithos_path, user)
             r = self.post('%s?update=' % url, user=user, **kwargs)
             self.assertEqual(r.status_code, 202)
             account_meta = self.get_account_meta()
             (self.assertTrue(k in account_meta) for k in meta.keys())
             (self.assertEqual(account_meta[k], v) for k, v in meta.items())
             account_meta = self.get_account_meta(user=user)
             (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
                 k in meta.keys())
             (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for
                 k, v in meta.items())
         def reset_account_meta(self, meta, user=None):
             user = user or self.user
             kwargs = dict(
                 ('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())
             url = join_urls(self.pithos_path, user)
             r = self.post(url, user=user, **kwargs)
             self.assertEqual(r.status_code, 202)
             account_meta = self.get_account_meta(user=user)
             (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
                 k in meta.keys())
             (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for
                 k, v in meta.items())
         def delete_account_meta(self, meta, user=None):
             user = user or self.user
             transform = (lambda k: 'HTTP_X_ACCOUNT_META_%s' %
                          k.replace('-', '_').upper())
             transform = lambda k: 'HTTP_%s' % k.replace('-', '_').upper()
             kwargs = dict((transform(k), '') for k, v in meta.items())
             url = join_urls(self.pithos_path, user)
             r = self.post('%s?update=' % url, user=user, **kwargs)
             self.assertEqual(r.status_code, 202)
             account_meta = self.get_account_meta()
             (self.assertTrue(k not in account_meta) for k in meta.keys())
             account_meta = self.get_account_meta(user=user)
             (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for
                 k in meta.keys())
             return r
         def delete_account_groups(self, groups, user=None):
             user = user or self.user
             url = join_urls(self.pithos_path, user)
             transform = (lambda k: 'HTTP_X_ACCOUNT_GROUP_%s' %
                          k.replace('-', '_').upper())
             kwargs = dict((transform(k), '') for k, v in groups.items())
             r = self.post('%s?update=' % url, user=user, **kwargs)
             r = self.post('%s?update=' % url, user=user, **groups)
             self.assertEqual(r.status_code, 202)
             account_groups = self.get_account_groups()
             (self.assertTrue(k not in account_groups) for k in groups.keys())
-...
         def get_object_meta(self, container, object, version=None, until=None,
                             user=None):
             prefix = 'X-Object-Meta-'
             user = user or self.user
             r = self.get_object_info(container, object, version, until=until,
                                      user=user)
             headers = dict(r._headers.values())

b/snf-pithos-app/pithos/api/test/listing.py
37	37
38	38	import django.utils.simplejson as json
39	39
	40
40	41	class ListSharing(PithosAPITest):
41	42	def _build_structure(self, user=None):
42	43	user = user or self.user

     # or implied, of GRNET S.A.
     from collections import defaultdict
     from urllib import quote
     from urllib import quote, unquote
     from functools import partial
     from pithos.api.test import (PithosAPITest, pithos_settings,
-...
     class ObjectHead(PithosAPITest):
         def setUp(self):
         def test_get_object_meta(self):
             cname = self.create_container()[0]
             oname, odata = self.upload_object(cname)[:-1]
             url = join_urls(self.pithos_path, cname, oname)
             url = join_urls(self.pithos_path, self.user, cname, oname)
             r = self.head(url)
             mandatory = ['Etag',
                          'Content-Length',
                          'Content-Type',
                          'Last-Modified',
                          'X-Object-Hash',
                          'X-Object-UUID',
                          'X-Object-Version',
                          'X-Object-Version-Timestamp',
                          'X-Object-Modified-By']
             for i in mandatory:
                 self.assertTrue(i in r)
             r = self.post(url, content_type='',
                           HTTP_CONTENT_ENCODING='gzip',
                           HTTP_CONTENT_DISPOSITION=(
                               'attachment; filename="%s"' % oname))
             self.assertEqual(r.status_code, 202)
             r = self.head(url)
             map(lambda i: self.assertTrue(i in r),
                 ['Etag',
                  'Content-Length',
                  'Content-Type',
                  'Last-Modified',
                  'Content-Encoding',
                  'Content-Disposition',
                  'X-Object-Hash',
                  'X-Object-UUID',
                  'X-Object-Version',
                  'X-Object-Version-Timestamp',
                  'X-Object-Modified-By',
                  'X-Object-Manifest'])
             for i in mandatory:
                 self.assertTrue(i in r)
             self.assertTrue('Content-Encoding' in r)
             self.assertEqual(r['Content-Encoding'], 'gzip')
             self.assertTrue('Content-Disposition' in r)
             self.assertEqual(unquote(r['Content-Disposition']),
                              'attachment; filename="%s"' % oname)
             prefix = 'myobject/'
             data = ''
             for i in range(random.randint(2, 10)):
                 part = '%s%d' % (prefix, i)
                 data += self.upload_object(cname, oname=part)[1]
             manifest = '%s/%s' % (cname, prefix)
             oname = get_random_name()
             url = join_urls(self.pithos_path, self.user, cname, oname)
             r = self.put(url, data='', HTTP_X_OBJECT_MANIFEST=manifest)
             self.assertEqual(r.status_code, 201)
             r = self.head(url)
             for i in mandatory:
                 self.assertTrue(i in r)
             self.assertTrue('X-Object-Manifest' in r)
             self.assertEqual(r['X-Object-Manifest'], manifest)
     class ObjectGet(PithosAPITest):
-...
             self.assertEqual(r.content, data)
     class ObjectCopy(PithosAPITest):
     class ObjectPutCopy(PithosAPITest):
         def setUp(self):
             PithosAPITest.setUp(self)
             self.container = 'c1'
-...
                 self.assertTrue('X-Object-Hash' in r)
                 self.assertEqual(r['X-Object-Hash'], self.etag)
         def test_copy_from_other_account(self):
             cname = 'c2'
             self.create_container(cname, user='chuck')
             self.create_container(cname, user='alice')
             # share object for read with alice
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.post(url, content_type='', HTTP_CONTENT_RANGE='bytes */*',
                           HTTP_X_OBJECT_SHARING='read=alice')
             self.assertEqual(r.status_code, 202)
             # assert not allowed for chuck
             oname = get_random_name()
             url = join_urls(self.pithos_path, 'chuck', cname, oname)
             r = self.put(url, data='', user='chuck',
                          HTTP_X_OBJECT_META_TEST='testcopy',
                          HTTP_X_COPY_FROM='/%s/%s' % (
                              self.container, self.object),
                          HTTP_X_SOURCE_ACCOUNT='user')
             self.assertEqual(r.status_code, 403)
             # assert copy success for alice
             url = join_urls(self.pithos_path, 'alice', cname, oname)
             r = self.put(url, data='', user='alice',
                          HTTP_X_OBJECT_META_TEST='testcopy',
                          HTTP_X_COPY_FROM='/%s/%s' % (
                              self.container, self.object),
                          HTTP_X_SOURCE_ACCOUNT='user')
             self.assertEqual(r.status_code, 201)
             # assert access the new object
             r = self.head(url, user='alice')
             self.assertEqual(r.status_code, 200)
             self.assertTrue('X-Object-Meta-Test' in r)
             self.assertEqual(r['X-Object-Meta-Test'], 'testcopy')
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
             # share object for write
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.post(url, content_type='', HTTP_CONTENT_RANGE='bytes */*',
                           HTTP_X_OBJECT_SHARING='write=dan')
             self.assertEqual(r.status_code, 202)
             # assert not allowed copy for alice
             url = join_urls(self.pithos_path, 'alice', cname, oname)
             r = self.put(url, data='', user='alice',
                          HTTP_X_OBJECT_META_TEST='testcopy',
                          HTTP_X_COPY_FROM='/%s/%s' % (
                              self.container, self.object),
                          HTTP_X_SOURCE_ACCOUNT='user')
             self.assertEqual(r.status_code, 403)
             # assert allowed copy for dan
             self.create_container(cname, user='dan')
             url = join_urls(self.pithos_path, 'dan', cname, oname)
             r = self.put(url, data='', user='dan',
                          HTTP_X_OBJECT_META_TEST='testcopy',
                          HTTP_X_COPY_FROM='/%s/%s' % (
                              self.container, self.object),
                          HTTP_X_SOURCE_ACCOUNT='user')
             self.assertEqual(r.status_code, 201)
             # assert access the new object
             r = self.head(url, user='dan')
             self.assertEqual(r.status_code, 200)
             self.assertTrue('X-Object-Meta-Test' in r)
             self.assertEqual(r['X-Object-Meta-Test'], 'testcopy')
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
             # assert source object still exists
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.head(url)
             self.assertEqual(r.status_code, 200)
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
             r = self.get(url)
             self.assertEqual(r.status_code, 200)
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
         def test_copy_invalid(self):
             # copy from non-existent object
             oname = get_random_name()
-...
             self.assertEqual(r.status_code, 404)
     class ObjectMove(PithosAPITest):
     class ObjectPutMove(PithosAPITest):
         def setUp(self):
             PithosAPITest.setUp(self)
             self.container = 'c1'
-...
             r = self.head(url)
             self.assertEqual(r.status_code, 404)
         def test_move_from_other_account(self):
             cname = 'c2'
             self.create_container(cname, user='chuck')
             self.create_container(cname, user='alice')
             # share object for read with alice
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.post(url, content_type='', HTTP_CONTENT_RANGE='bytes */*',
                           HTTP_X_OBJECT_SHARING='read=alice')
             self.assertEqual(r.status_code, 202)
             # assert not allowed move for chuck
             oname = get_random_name()
             url = join_urls(self.pithos_path, 'chuck', cname, oname)
             r = self.put(url, data='', user='chuck',
                          HTTP_X_OBJECT_META_TEST='testcopy',
                          HTTP_X_MOVE_FROM='/%s/%s' % (
                              self.container, self.object),
                          HTTP_X_SOURCE_ACCOUNT='user')
             self.assertEqual(r.status_code, 403)
             # assert no new object was created
             r = self.head(url, user='chuck')
             self.assertEqual(r.status_code, 404)
             # assert not allowed move for alice
             url = join_urls(self.pithos_path, 'alice', cname, oname)
             r = self.put(url, data='', user='alice',
                          HTTP_X_OBJECT_META_TEST='testcopy',
                          HTTP_X_MOVE_FROM='/%s/%s' % (
                              self.container, self.object),
                          HTTP_X_SOURCE_ACCOUNT='user')
             self.assertEqual(r.status_code, 403)
             # assert no new object was created
             r = self.head(url, user='alice')
             self.assertEqual(r.status_code, 404)
             # share object for write with dan
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.post(url, content_type='', HTTP_CONTENT_RANGE='bytes */*',
                           HTTP_X_OBJECT_SHARING='write=dan')
             self.assertEqual(r.status_code, 202)
             # assert not allowed move for alice
             url = join_urls(self.pithos_path, 'alice', cname, oname)
             r = self.put(url, data='', user='alice',
                          HTTP_X_OBJECT_META_TEST='testcopy',
                          HTTP_X_MOVE_FROM='/%s/%s' % (
                              self.container, self.object),
                          HTTP_X_SOURCE_ACCOUNT='user')
             self.assertEqual(r.status_code, 403)
             # assert no new object was created
             r = self.head(url, user='alice')
             self.assertEqual(r.status_code, 404)
             # assert not allowed move for dan
             self.create_container(cname, user='dan')
             url = join_urls(self.pithos_path, 'dan', cname, oname)
             r = self.put(url, data='', user='dan',
                          HTTP_X_OBJECT_META_TEST='testcopy',
                          HTTP_X_MOVE_FROM='/%s/%s' % (
                              self.container, self.object),
                          HTTP_X_SOURCE_ACCOUNT='user')
             self.assertEqual(r.status_code, 403)
             # assert no new object was created
             r = self.head(url, user='dan')
             self.assertEqual(r.status_code, 404)
     class ObjectCopy(PithosAPITest):
         def setUp(self):
             PithosAPITest.setUp(self)
             self.container = 'c1'
             self.create_container(self.container)
             self.object, self.data = self.upload_object(self.container)[:-1]
             url = join_urls(
                 self.pithos_path, self.user, self.container, self.object)
             r = self.head(url)
             self.etag = r['X-Object-Hash']
         def test_copy(self):
             with AssertMappingInvariant(self.get_object_info, self.container,
                                         self.object):
                 oname = get_random_name()
                 # copy object
                 url = join_urls(self.pithos_path, self.user, self.container,
                                 self.object)
                 r = self.copy(url, HTTP_X_OBJECT_META_TEST='testcopy',
                               HTTP_DESTINATION='/%s/%s' % (self.container,
                                                            oname))
                 # assert copy success
                 url = join_urls(self.pithos_path, self.user, self.container,
                                 oname)
                 self.assertEqual(r.status_code, 201)
                 # assert access the new object
                 r = self.head(url)
                 self.assertEqual(r.status_code, 200)
                 self.assertTrue('X-Object-Meta-Test' in r)
                 self.assertEqual(r['X-Object-Meta-Test'], 'testcopy')
                 # assert etag is the same
                 self.assertTrue('X-Object-Hash' in r)
                 self.assertEqual(r['X-Object-Hash'], self.etag)
                 # assert source object still exists
                 url = join_urls(self.pithos_path, self.user, self.container,
                                 self.object)
                 r = self.head(url)
                 self.assertEqual(r.status_code, 200)
                 # assert etag is the same
                 self.assertTrue('X-Object-Hash' in r)
                 self.assertEqual(r['X-Object-Hash'], self.etag)
                 r = self.get(url)
                 self.assertEqual(r.status_code, 200)
                 # assert etag is the same
                 self.assertTrue('X-Object-Hash' in r)
                 self.assertEqual(r['X-Object-Hash'], self.etag)
                 # copy object to other container (not existing)
                 cname = get_random_name()
                 url = join_urls(self.pithos_path, self.user, self.container,
                                 self.object)
                 r = self.copy(url, HTTP_X_OBJECT_META_TEST='testcopy',
                               HTTP_DESTINATION='/%s/%s' % (cname, self.object))
                 # assert destination container does not exist
                 url = join_urls(self.pithos_path, self.user, cname,
                                 self.object)
                 self.assertEqual(r.status_code, 404)
                 # create container
                 self.create_container(cname)
                 # copy object to other container (existing)
                 url = join_urls(self.pithos_path, self.user, self.container,
                                 self.object)
                 r = self.copy(url, HTTP_X_OBJECT_META_TEST='testcopy',
                               HTTP_DESTINATION='/%s/%s' % (cname, self.object))
                 # assert copy success
                 url = join_urls(self.pithos_path, self.user, cname,
                                 self.object)
                 self.assertEqual(r.status_code, 201)
                 # assert access the new object
                 r = self.head(url)
                 self.assertEqual(r.status_code, 200)
                 self.assertTrue('X-Object-Meta-Test' in r)
                 self.assertEqual(r['X-Object-Meta-Test'], 'testcopy')
                 # assert etag is the same
                 self.assertTrue('X-Object-Hash' in r)
                 self.assertEqual(r['X-Object-Hash'], self.etag)
                             # assert source object still exists
                 url = join_urls(self.pithos_path, self.user, self.container,
                                 self.object)
                 r = self.head(url)
                 self.assertEqual(r.status_code, 200)
                 # assert etag is the same
                 self.assertTrue('X-Object-Hash' in r)
                 self.assertEqual(r['X-Object-Hash'], self.etag)
                 r = self.get(url)
                 self.assertEqual(r.status_code, 200)
                 # assert etag is the same
                 self.assertTrue('X-Object-Hash' in r)
                 self.assertEqual(r['X-Object-Hash'], self.etag)
         def test_copy_to_other_account(self):
             # create a container under alice account
             cname = self.create_container(user='alice')[0]
             # create a folder under this container
             folder = self.create_folder(cname, user='alice')[0]
             oname = get_random_name()
             # copy object to other account container
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.copy(url, HTTP_X_OBJECT_META_TEST='testcopy',
                           HTTP_DESTINATION='/%s/%s/%s' % (cname, folder, oname),
                           HTTP_DESTINATION_ACCOUNT='alice')
             self.assertEqual(r.status_code, 403)
             # share object for read with user
             url = join_urls(self.pithos_path, 'alice', cname, folder)
             r = self.post(url, user='alice', content_type='',
                           HTTP_CONTENT_RANGE='bytes */*',
                           HTTP_X_OBJECT_SHARING='read=%s' % self.user)
             self.assertEqual(r.status_code, 202)
             # assert copy object still is not allowed
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.copy(url, HTTP_X_OBJECT_META_TEST='testcopy',
                           HTTP_DESTINATION='/%s/%s/%s' % (cname, folder, oname),
                           HTTP_DESTINATION_ACCOUNT='alice')
             self.assertEqual(r.status_code, 403)
             # share object for write with user
             url = join_urls(self.pithos_path, 'alice', cname, folder)
             r = self.post(url, user='alice',  content_type='',
                           HTTP_CONTENT_RANGE='bytes */*',
                           HTTP_X_OBJECT_SHARING='write=%s' % self.user)
             self.assertEqual(r.status_code, 202)
             # assert copy object now is allowed
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.copy(url, HTTP_X_OBJECT_META_TEST='testcopy',
                           HTTP_DESTINATION='/%s/%s/%s' % (cname, folder, oname),
                           HTTP_DESTINATION_ACCOUNT='alice')
             self.assertEqual(r.status_code, 201)
             # assert access the new object
             url = join_urls(self.pithos_path, 'alice', cname, folder, oname)
             r = self.head(url, user='alice')
             self.assertEqual(r.status_code, 200)
             self.assertTrue('X-Object-Meta-Test' in r)
             self.assertEqual(r['X-Object-Meta-Test'], 'testcopy')
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
             # assert source object still exists
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.head(url)
             self.assertEqual(r.status_code, 200)
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
             r = self.get(url)
             self.assertEqual(r.status_code, 200)
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
     class ObjectMove(PithosAPITest):
         def setUp(self):
             PithosAPITest.setUp(self)
             self.container = 'c1'
             self.create_container(self.container)
             self.object, self.data = self.upload_object(self.container)[:-1]
             url = join_urls(
                 self.pithos_path, self.user, self.container, self.object)
             r = self.head(url)
             self.etag = r['X-Object-Hash']
         def test_move(self):
             oname = get_random_name()
             # move object
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.move(url, HTTP_X_OBJECT_META_TEST='testmove',
                           HTTP_DESTINATION='/%s/%s' % (self.container,
                                                        oname))
             # assert move success
             url = join_urls(self.pithos_path, self.user, self.container,
                             oname)
             self.assertEqual(r.status_code, 201)
             # assert access the new object
             r = self.head(url)
             self.assertEqual(r.status_code, 200)
             self.assertTrue('X-Object-Meta-Test' in r)
             self.assertEqual(r['X-Object-Meta-Test'], 'testmove')
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
             # assert source object does not exist
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.head(url)
             self.assertEqual(r.status_code, 404)
         def test_move_to_other_container(self):
             # move object to other container (not existing)
             cname = get_random_name()
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.move(url, HTTP_X_OBJECT_META_TEST='testmove',
                           HTTP_DESTINATION='/%s/%s' % (cname, self.object))
             # assert destination container does not exist
             url = join_urls(self.pithos_path, self.user, cname,
                             self.object)
             self.assertEqual(r.status_code, 404)
             # create container
             self.create_container(cname)
             # move object to other container (existing)
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.move(url, HTTP_X_OBJECT_META_TEST='testmove',
                           HTTP_DESTINATION='/%s/%s' % (cname, self.object))
             # assert move success
             url = join_urls(self.pithos_path, self.user, cname,
                             self.object)
             self.assertEqual(r.status_code, 201)
             # assert access the new object
             r = self.head(url)
             self.assertEqual(r.status_code, 200)
             self.assertTrue('X-Object-Meta-Test' in r)
             self.assertEqual(r['X-Object-Meta-Test'], 'testmove')
             # assert etag is the same
             self.assertTrue('X-Object-Hash' in r)
             self.assertEqual(r['X-Object-Hash'], self.etag)
             # assert source object does not exist
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.head(url)
             self.assertEqual(r.status_code, 404)
         def test_move_to_other_account(self):
             # create a container under alice account
             cname = self.create_container(user='alice')[0]
             # create a folder under this container
             folder = self.create_folder(cname, user='alice')[0]
             oname = get_random_name()
             # move object to other account container
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.move(url, HTTP_X_OBJECT_META_TEST='testmove',
                           HTTP_DESTINATION='/%s/%s/%s' % (cname, folder, oname),
                           HTTP_DESTINATION_ACCOUNT='alice')
             self.assertEqual(r.status_code, 403)
             # share object for read with user
             url = join_urls(self.pithos_path, 'alice', cname, folder)
             r = self.post(url, user='alice', content_type='',
                           HTTP_CONTENT_RANGE='bytes */*',
                           HTTP_X_OBJECT_SHARING='read=%s' % self.user)
             self.assertEqual(r.status_code, 202)
             # assert move object still is not allowed
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.move(url, HTTP_X_OBJECT_META_TEST='testmove',
                           HTTP_DESTINATION='/%s/%s/%s' % (cname, folder, oname),
                           HTTP_DESTINATION_ACCOUNT='alice')
             self.assertEqual(r.status_code, 403)
             # share object for write with user
             url = join_urls(self.pithos_path, 'alice', cname, folder)
             r = self.post(url, user='alice',  content_type='',
                           HTTP_CONTENT_RANGE='bytes */*',
                           HTTP_X_OBJECT_SHARING='write=%s' % self.user)
             self.assertEqual(r.status_code, 202)
             # assert move object now is allowed
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.move(url, HTTP_X_OBJECT_META_TEST='testmove',
                           HTTP_DESTINATION='/%s/%s/%s' % (cname, folder, oname),
                           HTTP_DESTINATION_ACCOUNT='alice')
             self.assertEqual(r.status_code, 201)
             # assert source object does not exist
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.head(url)
             self.assertEqual(r.status_code, 404)
     class ObjectPost(PithosAPITest):
         def setUp(self):
-...
             self.assertEqual(data, v[2][2])
             append((r['X-Object-Version'], len(data), data))
         def test_update_from_other_version(self):
             versions = []
             info = self.get_object_info(self.container, self.object)
             versions.append(info['X-Object-Version'])
             pre_length = int(info['Content-Length'])
             # update object
             d1, r = self.upload_object(self.container, self.object,
                                        length=pre_length - 1)[1:]
             self.assertTrue('X-Object-Version' in r)
             versions.append(r['X-Object-Version'])
             # update object
             d2, r = self.upload_object(self.container, self.object,
                                        length=pre_length - 2)[1:]
             self.assertTrue('X-Object-Version' in r)
             versions.append(r['X-Object-Version'])
             # get previous version
             url = join_urls(self.pithos_path, self.user, self.container,
                             self.object)
             r = self.get('%s?version=list&format=json' % url)
             self.assertEqual(r.status_code, 200)
             l = json.loads(r.content)['versions']
             self.assertEqual(len(l), 3)
             self.assertEqual([str(v[0]) for v in l], versions)
             # update with the previous version
             r = self.post(url,
                           HTTP_CONTENT_RANGE='bytes 0-/*',
                           HTTP_X_SOURCE_OBJECT='/%s/%s' % (self.container,
                                                            self.object),
                           HTTP_X_SOURCE_VERSION=versions[0])
             self.assertEqual(r.status_code, 204)
             # check content
             r = self.get(url)
             content = r.content
             self.assertEqual(len(content), pre_length)
             self.assertEqual(content, self.object_data)
             # update object
             d3, r = self.upload_object(self.container, self.object,
                                        length=len(d2) + 1)[1:]
             self.assertTrue('X-Object-Version' in r)
             versions.append(r['X-Object-Version'])
             # update with the previous version
             r = self.post(url,
                           HTTP_CONTENT_RANGE='bytes 0-/*',
                           HTTP_X_SOURCE_OBJECT='/%s/%s' % (self.container,
                                                            self.object),
                           HTTP_X_SOURCE_VERSION=versions[-2])
             self.assertEqual(r.status_code, 204)
             # check content
             r = self.get(url)
             content = r.content
             self.assertEqual(content, d2 + d3[-1])
     class ObjectDelete(PithosAPITest):
         def setUp(self):

                     url, content_type='', HTTP_CONTENT_RANGE='bytes */*',
                     HTTP_X_OBJECT_SHARING='write=%s:%s' % (self.user, group))
                 self._assert_write(self.object, members)
         def test_not_allowed(self):
             cname = self.create_container()[0]
             oname, odata = self.upload_object(cname)[:-1]
             url = join_urls(self.pithos_path, self.user)
             r = self.head(url, user='chuck')
             self.assertEqual(r.status_code, 403)
             r = self.get(url, user='chuck')
             self.assertEqual(r.status_code, 403)
             r = self.post(url, user='chuck', data=get_random_data())
             self.assertEqual(r.status_code, 403)
             url = join_urls(self.pithos_path, self.user, cname)
             r = self.head(url, user='chuck')
             self.assertEqual(r.status_code, 403)
             r = self.get(url, user='chuck')
             self.assertEqual(r.status_code, 403)
             r = self.put(url, user='chuck', data=get_random_data())
             self.assertEqual(r.status_code, 403)
             r = self.post(url, user='chuck', data=get_random_data())
             self.assertEqual(r.status_code, 403)
             r = self.delete(url, user='chuck')
             self.assertEqual(r.status_code, 403)
             url = join_urls(self.pithos_path, self.user, cname, oname)
             r = self.head(url, user='chuck')
             self.assertEqual(r.status_code, 403)
             r = self.get(url, user='chuck')
             self.assertEqual(r.status_code, 403)
             r = self.put(url, user='chuck', data=get_random_data())
             self.assertEqual(r.status_code, 403)
             r = self.post(url, user='chuck', data=get_random_data())
             self.assertEqual(r.status_code, 403)
             r = self.delete(url, user='chuck')
             self.assertEqual(r.status_code, 403)

             (self.assertTrue(l in pithos_settings.PUBLIC_URL_ALPHABET) for
              l in public)
             r = self.get(public, user='user2')
             r = self.get(public, user='user2', token=None)
             self.assertEqual(r.status_code, 200)
             self.assertTrue('X-Object-Public' not in r)
-...
             public2 = self._assert_public_object(cname, oname, odata)
             self.assertEqual(public, public2)
             # delete object histoy until now
             # delete object history until now
             _time.sleep(1)
             t = datetime.datetime.utcnow()
             now = int(_time.mktime(t.timetuple()))
-...
             self.assertEqual(r.status_code, 404)
             r = self.get(public)
             self.assertEqual(r.status_code, 404)
         def test_public_manifest(self):
             cname = self.create_container()[0]
             prefix = 'myobject/'
             data = ''
             for i in range(random.randint(2, 10)):
                 part = '%s%d' % (prefix, i)
                 data += self.upload_object(cname, oname=part)[1]
             manifest = '%s/%s' % (cname, prefix)
             oname = get_random_name()
             url = join_urls(self.pithos_path, self.user, cname, oname)
             r = self.put(url, data='', HTTP_X_OBJECT_MANIFEST=manifest,
                          HTTP_X_OBJECT_PUBLIC='true')
             self.assertEqual(r.status_code, 201)
             r = self.head(url)
             self.assertTrue('X-Object-Manifest' in r)
             self.assertEqual(r['X-Object-Manifest'], manifest)
             self.assertTrue('X-Object-Public' in r)
             public = r['X-Object-Public']
             r = self.get(public)
             self.assertTrue(r.content, data)
             #self.assertTrue('X-Object-Manifest' in r)
             #self.assertEqual(r['X-Object-Manifest'], manifest)

b/snf-pithos-app/pithos/api/tests.py
40	40	from pithos.api.test.views import *
41	41	from pithos.api.test.unicode import *
42	42	from pithos.api.test.listing import *
	43	from pithos.api.test.top_level import *

Also available in: Unified diff