commit f53483d80521d296d456883de4cf99cc7d0e76d8
Author: Sofia Papagiannaki <spapagian@grnet.gr>
Date:   Fri Sep 6 17:46:11 2013 +0300

    pithos: Add tests

diff --git a/snf-pithos-app/pithos/api/test/__init__.py b/snf-pithos-app/pithos/api/test/__init__.py
index 2f8d992..c487c3f 100644
--- a/snf-pithos-app/pithos/api/test/__init__.py
+++ b/snf-pithos-app/pithos/api/test/__init__.py
@@ -34,7 +34,7 @@
 # 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
 
@@ -48,6 +48,7 @@ from synnefo.lib.services import get_service_path
 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
@@ -102,7 +103,7 @@ django_sqlalchemy_engines = {
     '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']
@@ -141,7 +142,7 @@ class PithosTestSuiteRunner(DjangoTestSuiteRunner):
     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):
@@ -151,8 +152,52 @@ class PithosTestSuiteRunner(DjangoTestSuiteRunner):
                                                               **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
@@ -177,45 +222,73 @@ class PithosAPITest(TestCase):
             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(
@@ -223,29 +296,41 @@ class PithosAPITest(TestCase):
         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())
@@ -438,6 +523,7 @@ class PithosAPITest(TestCase):
     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())
diff --git a/snf-pithos-app/pithos/api/test/listing.py b/snf-pithos-app/pithos/api/test/listing.py
index 782a0fe..5d54ea4 100644
--- a/snf-pithos-app/pithos/api/test/listing.py
+++ b/snf-pithos-app/pithos/api/test/listing.py
@@ -37,6 +37,7 @@ from synnefo.lib import join_urls
 
 import django.utils.simplejson as json
 
+
 class ListSharing(PithosAPITest):
     def _build_structure(self, user=None):
         user = user or self.user
diff --git a/snf-pithos-app/pithos/api/test/objects.py b/snf-pithos-app/pithos/api/test/objects.py
index 0d02b74..1b899cf 100644
--- a/snf-pithos-app/pithos/api/test/objects.py
+++ b/snf-pithos-app/pithos/api/test/objects.py
@@ -35,7 +35,7 @@
 # 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,
@@ -60,25 +60,57 @@ merkle = partial(merkle,
 
 
 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):
@@ -702,7 +734,7 @@ class ObjectPut(PithosAPITest):
         self.assertEqual(r.content, data)
 
 
-class ObjectCopy(PithosAPITest):
+class ObjectPutCopy(PithosAPITest):
     def setUp(self):
         PithosAPITest.setUp(self)
         self.container = 'c1'
@@ -761,6 +793,99 @@ class ObjectCopy(PithosAPITest):
             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()
@@ -822,7 +947,7 @@ class ObjectCopy(PithosAPITest):
         self.assertEqual(r.status_code, 404)
 
 
-class ObjectMove(PithosAPITest):
+class ObjectPutMove(PithosAPITest):
     def setUp(self):
         PithosAPITest.setUp(self)
         self.container = 'c1'
@@ -901,6 +1026,402 @@ class ObjectMove(PithosAPITest):
         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):
@@ -1277,6 +1798,66 @@ class ObjectPost(PithosAPITest):
         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):
diff --git a/snf-pithos-app/pithos/api/test/permissions.py b/snf-pithos-app/pithos/api/test/permissions.py
index 5349cbc..1b150b4 100644
--- a/snf-pithos-app/pithos/api/test/permissions.py
+++ b/snf-pithos-app/pithos/api/test/permissions.py
@@ -307,3 +307,39 @@ class TestPermissions(PithosAPITest):
                 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)
diff --git a/snf-pithos-app/pithos/api/test/public.py b/snf-pithos-app/pithos/api/test/public.py
index 61061e6..1294525 100644
--- a/snf-pithos-app/pithos/api/test/public.py
+++ b/snf-pithos-app/pithos/api/test/public.py
@@ -58,7 +58,7 @@ class TestPublic(PithosAPITest):
         (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)
 
@@ -212,7 +212,7 @@ class TestPublic(PithosAPITest):
         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()))
@@ -222,3 +222,30 @@ class TestPublic(PithosAPITest):
         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)
diff --git a/snf-pithos-app/pithos/api/tests.py b/snf-pithos-app/pithos/api/tests.py
index 392c42c..8e2b2fd 100644
--- a/snf-pithos-app/pithos/api/tests.py
+++ b/snf-pithos-app/pithos/api/tests.py
@@ -40,3 +40,4 @@ from pithos.api.test.public import *
 from pithos.api.test.views import *
 from pithos.api.test.unicode import *
 from pithos.api.test.listing import *
+from pithos.api.test.top_level import *
