Synnefo » ./kamaki

# Copyright 2012-2013 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

import time

import datetime

from os import urandom

from tempfile import NamedTemporaryFile

from string import ascii_letters

from kamaki.clients import livetest, ClientError

from kamaki.clients.pithos import PithosClient

from kamaki.clients.astakos import AstakosClient

def chargen():

    """10 + 2 * 26 + 26 = 88"""

    while True:

        for CH in xrange(10):

            yield '%s' % CH

        for CH in ascii_letters:

            yield CH

        for CH in '~!@#$%^&*()_+`-=:";|<>?,./':

            yield CH

def sample_block(f, block):

    block_size = 4 * 1024 * 1024

    f.seek(block * block_size)

    ch = [f.read(1)]

    f.seek(block_size / 2, 1)

    ch.append(f.read(1))

    f.seek((block + 1) * block_size - 1)

    ch.append(f.read(1))

    return ch

class Pithos(livetest.Generic):

    files = []

    def setUp(self):

        self.cloud = 'cloud.%s' % self['testcloud']

        aurl, self.token = self[self.cloud, 'url'], self[self.cloud, 'token']

        self.auth_base = AstakosClient(aurl, self.token)

        purl = self.auth_base.get_service_endpoints(

            'object-store')['publicURL']

        self.uuid = self.auth_base.user_term('id')

        self.client = PithosClient(purl, self.token, self.uuid)

        self.now = time.mktime(time.gmtime())

        self.now_unformated = datetime.datetime.utcnow()

        self._init_data()

        """Prepare an object to be shared - also its container"""

        self.client.container = self.c1

        self.client.object_post(

            'test',

            update=True,

            permissions={'read': [self.client.account]})

        self.create_remote_object(self.c1, 'another.test')

    def _init_data(self):

        self.c1 = 'c1_' + unicode(self.now)

        self.c2 = 'c2_' + unicode(self.now)

        self.c3 = 'c3_' + unicode(self.now)

        try:

            self.client.create_container(self.c2)

        except ClientError:

            pass

        try:

            self.client.create_container(self.c1)

        except ClientError:

            pass

        try:

            self.client.create_container(self.c3)

        except ClientError:

            pass

        self.create_remote_object(self.c1, 'test')

        self.create_remote_object(self.c2, 'test')

        self.create_remote_object(self.c1, 'test1')

        self.create_remote_object(self.c2, 'test1')

    def create_remote_object(self, container, obj):

        self.client.container = container

        self.client.object_put(

            obj,

            content_type='application/octet-stream',

            data='file %s that lives in %s' % (obj, container),

            metadata={'incontainer': container})

    def forceDeleteContainer(self, container):

        self.client.container = container

        try:

            r = self.client.list_objects()

        except ClientError:

            return

        for obj in r:

            name = obj['name']

            self.client.del_object(name)

        r = self.client.container_delete()

        self.container = ''

    def tearDown(self):

        """Destroy test cases"""

        for f in self.files:

            f.close()

        self.forceDeleteContainer(self.c1)

        self.forceDeleteContainer(self.c2)

        try:

            self.forceDeleteContainer(self.c3)

        except ClientError:

            pass

        self.client.container = ''

    def test_000(self):

        """Prepare a full Pithos+ test"""

        print('')

        super(self.__class__, self).test_000()

    def test_account_head(self):

        """Test account_HEAD"""

        self._test_0010_account_head()

    def _test_0010_account_head(self):

        r = self.client.account_head()

        self.assertEqual(r.status_code, 204)

        r = self.client.account_head(until='1000000000')

        self.assertEqual(r.status_code, 204)

        r = self.client.get_account_info(until='1000000000')

        datestring = unicode(r['x-account-until-timestamp'])

        self.assertEqual(u'Sun, 09 Sep 2001 01:46:40 GMT', datestring)

        r = self.client.get_account_quota()

        self.assertTrue('x-account-policy-quota' in r)

        #r = self.client.get_account_versioning()

        #self.assertTrue('x-account-policy-versioning' in r)

        """Check if(un)modified_since"""

        for format in self.client.DATE_FORMATS:

            now_formated = self.now_unformated.strftime(format)

            r1 = self.client.account_head(

                if_modified_since=now_formated, success=(204, 304, 412))

            sc1 = r1.status_code

            r2 = self.client.account_head(

                if_unmodified_since=now_formated, success=(204, 304, 412))

            sc2 = r2.status_code

            self.assertNotEqual(sc1, sc2)

    def test_account_get(self):

        """Test account_GET"""

        self._test_0020_account_get()

    def _test_0020_account_get(self):

        #r = self.client.account_get()

        #self.assertEqual(r.status_code, 200)

        r = self.client.list_containers()

        fullLen = len(r)

        self.assertTrue(fullLen > 2)

        r = self.client.account_get(limit=1)

        self.assertEqual(len(r.json), 1)

        r = self.client.account_get(marker='c2_')

        temp_c0 = r.json[0]['name']

        temp_c2 = r.json[2]['name']

        r = self.client.account_get(limit=2, marker='c2_')

        conames = [container['name'] for container in r.json if (

            container['name'].lower().startswith('c2_'))]

        self.assertTrue(temp_c0 in conames)

        self.assertFalse(temp_c2 in conames)

        r = self.client.account_get(show_only_shared=True)

        self.assertTrue(self.c1 in [c['name'] for c in r.json])

        r = self.client.account_get(until=1342609206.0)

        self.assertTrue(len(r.json) <= fullLen)

        """Check if(un)modified_since"""

        for format in self.client.DATE_FORMATS:

            now_formated = self.now_unformated.strftime(format)

            r1 = self.client.account_get(

                if_modified_since=now_formated, success=(200, 304, 412))

            sc1 = r1.status_code

            r2 = self.client.account_get(

                if_unmodified_since=now_formated, success=(200, 304, 412))

            sc2 = r2.status_code

            self.assertNotEqual(sc1, sc2)

        """Check sharing_accounts"""

        r = self.client.get_sharing_accounts()

        self.assertTrue(len(r) > 0)

    def test_account_post(self):

        """Test account_POST"""

        self._test_0030_account_post()

    def _test_0030_account_post(self):

        r = self.client.account_post()

        self.assertEqual(r.status_code, 202)

        grpName = 'grp' + unicode(self.now)

        """Method set/del_account_meta and set_account_groupcall use

            account_post internally

        """

        u1 = self.client.account

        #  Invalid display name

        u2 = '1nc0r3c7-d15p14y-n4m3'

        #  valid display name

        u3 = '6488c1b2-cb06-40a8-a02a-d474b8d29c59'

        self.assertRaises(

            ClientError,

            self.client.set_account_group,

            grpName, [u1, u2])

        self.client.set_account_group(grpName, [u1])

        r = self.client.get_account_group()

        self.assertEqual(r['x-account-group-' + grpName], '%s' % u1)

        try:

            self.client.set_account_group(grpName, [u1, u3])

            r = self.client.get_account_group()

            self.assertEqual(

                r['x-account-group-' + grpName], '%s,%s' % (u1, u3))

        except:

            print('\tInvalid user id %s (it is ok, though)' % u3)

        self.client.del_account_group(grpName)

        r = self.client.get_account_group()

        self.assertTrue('x-account-group-' + grpName not in r)

        mprefix = 'meta' + unicode(self.now)

        self.client.set_account_meta({

            mprefix + '1': 'v1', mprefix + '2': 'v2'})

        r = self.client.get_account_meta()

        self.assertEqual(r['x-account-meta-' + mprefix + '1'], 'v1')

        self.assertEqual(r['x-account-meta-' + mprefix + '2'], 'v2')

        self.client.del_account_meta(mprefix + '1')

        r = self.client.get_account_meta()

        self.assertTrue('x-account-meta-' + mprefix + '1' not in r)

        self.client.del_account_meta(mprefix + '2')

        r = self.client.get_account_meta()

        self.assertTrue('x-account-meta-' + mprefix + '2' not in r)

        """Missing testing for quota, versioning, because normally

        you don't have permissions to modify those at account level

        """

        #newquota = 1000000

        #self.client.set_account_quota(newquota)

        #r = self.client.get_account_info()

        #print(unicode(r))

        #r = self.client.get_account_quota()

        #self.assertEqual(r['x-account-policy-quota'], newquota)

        #self.client.set_account_versioning('auto')

    def test_container_head(self):

        """Test container_HEAD"""

        self._test_0040_container_head()

    def _test_0040_container_head(self):

        self.client.container = self.c1

        r = self.client.container_head()

        self.assertEqual(r.status_code, 204)

        """Check until"""

        r = self.client.container_head(until=1000000, success=(204, 404))

        self.assertEqual(r.status_code, 404)

        """Check and if(un)modified_since"""

        for format in self.client.DATE_FORMATS:

            now_formated = self.now_unformated.strftime(format)

            r1 = self.client.container_head(

                if_modified_since=now_formated, success=(204, 304, 412))

            sc1 = r1.status_code

            r2 = self.client.container_head(

                if_unmodified_since=now_formated, success=(204, 304, 412))

            sc2 = r2.status_code

            self.assertNotEqual(sc1, sc2)

        """Check container object meta"""

        r = self.client.get_container_object_meta()

        self.assertEqual(r['x-container-object-meta'], 'Incontainer')

    def test_container_get(self):

        """Test container_GET"""

        self._test_0050_container_get()

    def _test_0050_container_get(self):

        self.client.container = self.c1

        r = self.client.container_get()

        self.assertEqual(r.status_code, 200)

        fullLen = len(r.json)

        r = self.client.container_get(prefix='test')

        lalobjects = [obj for obj in r.json if obj['name'].startswith('test')]

        self.assertTrue(len(r.json) > 1)

        self.assertEqual(len(r.json), len(lalobjects))

        r = self.client.container_get(limit=1)

        self.assertEqual(len(r.json), 1)

        r = self.client.container_get(marker='another')

        self.assertTrue(len(r.json) > 1)

        neobjects = [obj for obj in r.json if obj['name'] > 'another']

        self.assertEqual(len(r.json), len(neobjects))

        r = self.client.container_get(prefix='another.test', delimiter='.')

        self.assertTrue(fullLen > len(r.json))

        r = self.client.container_get(path='/')

        self.assertEqual(fullLen, len(r.json))

        r = self.client.container_get(format='xml')

        self.assertEqual(r.text.split()[4], 'name="' + self.c1 + '">')

        r = self.client.container_get(meta=['incontainer'])

        self.assertTrue(len(r.json) > 0)

        r = self.client.container_get(show_only_shared=True)

        self.assertTrue(len(r.json) < fullLen)

        try:

            r = self.client.container_get(until=1000000000)

            datestring = unicode(r.headers['x-account-until-timestamp'])

            self.assertEqual(u'Sun, 09 Sep 2001 01:46:40 GMT', datestring)

        except ClientError:

            pass

        """Check and if un/modified_since"""

        for format in self.client.DATE_FORMATS:

            now_formated = self.now_unformated.strftime(format)

            r1 = self.client.container_get(

                if_modified_since=now_formated, success=(200, 304, 412))

            sc1 = r1.status_code

            r2 = self.client.container_get(

                if_unmodified_since=now_formated, success=(200, 304, 412))

            sc2 = r2.status_code

            self.assertNotEqual(sc1, sc2)

    def test_container_put(self):

        """Test container_PUT"""

        self._test_0050_container_put()

    def _test_0050_container_put(self):

        self.client.container = self.c2

        r = self.client.create_container()

        self.assertTrue(isinstance(r, dict))

        r = self.client.get_container_limit(self.client.container)

        cquota = r.values()[0]

        newquota = 2 * int(cquota)

        r = self.client.create_container(sizelimit=newquota)

        self.assertTrue(isinstance(r, dict))

        r = self.client.get_container_limit(self.client.container)

        xquota = int(r.values()[0])

        self.assertEqual(newquota, xquota)

        r = self.client.create_container(versioning='auto')

        self.assertTrue(isinstance(r, dict))

        r = self.client.get_container_versioning(self.client.container)

        nvers = r.values()[0]

        self.assertEqual('auto', nvers)

        r = self.client.container_put(versioning='none')

        self.assertEqual(r.status_code, 202)

        r = self.client.get_container_versioning(self.client.container)

        nvers = r.values()[0]

        self.assertEqual('none', nvers)

        r = self.client.create_container(metadata={'m1': 'v1', 'm2': 'v2'})

        self.assertTrue(isinstance(r, dict))

        r = self.client.get_container_meta(self.client.container)

        self.assertTrue('x-container-meta-m1' in r)

        self.assertEqual(r['x-container-meta-m1'], 'v1')

        self.assertTrue('x-container-meta-m2' in r)

        self.assertEqual(r['x-container-meta-m2'], 'v2')

        r = self.client.container_put(metadata={'m1': '', 'm2': 'v2a'})

        self.assertEqual(r.status_code, 202)

        r = self.client.get_container_meta(self.client.container)

        self.assertTrue('x-container-meta-m1' not in r)

        self.assertTrue('x-container-meta-m2' in r)

        self.assertEqual(r['x-container-meta-m2'], 'v2a')

        self.client.del_container_meta(self.client.container)

    def test_container_post(self):

        """Test container_POST"""

        self._test_0060_container_post()

    def _test_0060_container_post(self):

        self.client.container = self.c2

        """Simple post"""

        r = self.client.container_post()

        self.assertEqual(r.status_code, 202)

        """post meta"""

        self.client.set_container_meta({'m1': 'v1', 'm2': 'v2'})

        r = self.client.get_container_meta(self.client.container)

        self.assertTrue('x-container-meta-m1' in r)

        self.assertEqual(r['x-container-meta-m1'], 'v1')

        self.assertTrue('x-container-meta-m2' in r)

        self.assertEqual(r['x-container-meta-m2'], 'v2')

        """post/2del meta"""

        r = self.client.del_container_meta('m1')

        r = self.client.set_container_meta({'m2': 'v2a'})

        r = self.client.get_container_meta(self.client.container)

        self.assertTrue('x-container-meta-m1' not in r)

        self.assertTrue('x-container-meta-m2' in r)

        self.assertEqual(r['x-container-meta-m2'], 'v2a')

        """check quota"""

        r = self.client.get_container_limit(self.client.container)

        cquota = r.values()[0]

        newquota = 2 * int(cquota)

        r = self.client.set_container_limit(newquota)

        r = self.client.get_container_limit(self.client.container)

        xquota = int(r.values()[0])

        self.assertEqual(newquota, xquota)

        r = self.client.set_container_limit(cquota)

        r = self.client.get_container_limit(self.client.container)

        xquota = r.values()[0]

        self.assertEqual(cquota, xquota)

        """Check versioning"""

        self.client.set_container_versioning('auto')

        r = self.client.get_container_versioning(self.client.container)

        nvers = r.values()[0]

        self.assertEqual('auto', nvers)

        self.client.set_container_versioning('none')

        r = self.client.get_container_versioning(self.client.container)

        nvers = r.values()[0]

        self.assertEqual('none', nvers)

        """put_block uses content_type and content_length to

        post blocks of data 2 container. All that in upload_object"""

        """Change a file at fs"""

        f = self.create_large_file(1024 * 1024 * 100)

        """Upload it at a directory in container"""

        self.client.create_directory('dir')

        r = self.client.upload_object('/dir/sample.file', f)

        for term in ('content-length', 'content-type', 'x-object-version'):

            self.assertTrue(term in r)

        """Check if file has been uploaded"""

        r = self.client.get_object_info('/dir/sample.file')

        self.assertTrue(int(r['content-length']) > 100000000)

        """What is tranfer_encoding? What should I check about it? """

        #TODO

        """Check update=False"""

        r = self.client.object_post(

            'test',

            update=False,

            metadata={'newmeta': 'newval'})

        r = self.client.get_object_info('test')

        self.assertTrue('x-object-meta-newmeta' in r)

        self.assertFalse('x-object-meta-incontainer' in r)

        r = self.client.del_container_meta('m2')

    def test_container_delete(self):

        """Test container_DELETE"""

        self._test_0070_container_delete()

    def _test_0070_container_delete(self):

        """Fail to delete a non-empty container"""

        self.client.container = self.c2

        r = self.client.container_delete(success=409)

        self.assertEqual(r.status_code, 409)

        """Fail to delete c3 (empty) container"""

        self.client.container = self.c3

        r = self.client.container_delete(until='1000000000')

        self.assertEqual(r.status_code, 204)

        """Delete c3 (empty) container"""

        r = self.client.container_delete()

        self.assertEqual(r.status_code, 204)

        """Purge container(empty a container), check versionlist"""

        self.client.container = self.c1

        r = self.client.object_head('test', success=(200, 404))

        self.assertEqual(r.status_code, 200)

        self.client.del_container(delimiter='/')

        r = self.client.object_head('test', success=(200, 404))

        self.assertEqual(r.status_code, 404)

        r = self.client.get_object_versionlist('test')

        self.assertTrue(len(r) > 0)

        self.assertTrue(len(r[0]) > 1)

        self.client.purge_container()

        self.assertRaises(

            ClientError, self.client.get_object_versionlist, 'test')

    def _test_0080_recreate_deleted_data(self):

        self._init_data()

    def test_object_head(self):

        """Test object_HEAD"""

        self._test_0090_object_head()

    def _test_0090_object_head(self):

        self.client.container = self.c2

        obj = 'test'

        r = self.client.object_head(obj)

        self.assertEqual(r.status_code, 200)

        etag = r.headers['etag']

        real_version = r.headers['x-object-version']

        self.assertRaises(

            ClientError, self.client.object_head, obj, version=-10)

        r = self.client.object_head(obj, version=real_version)

        self.assertEqual(r.headers['x-object-version'], real_version)

        r = self.client.object_head(obj, if_etag_match=etag)

        self.assertEqual(r.status_code, 200)

        r = self.client.object_head(

            obj, if_etag_not_match=etag, success=(200, 412, 304))

        self.assertNotEqual(r.status_code, 200)

        r = self.client.object_head(

            obj, version=real_version, if_etag_match=etag, success=200)

        self.assertEqual(r.status_code, 200)

        """Check and if(un)modified_since"""

        for format in self.client.DATE_FORMATS:

            now_formated = self.now_unformated.strftime(format)

            r1 = self.client.object_head(

                obj, if_modified_since=now_formated, success=(200, 304, 412))

            sc1 = r1.status_code

            r2 = self.client.object_head(

                obj, if_unmodified_since=now_formated, success=(200, 304, 412))

            sc2 = r2.status_code

            self.assertNotEqual(sc1, sc2)

    def test_object_get(self):

        """Test object_GET"""

        self._test_0100_object_get()

    def _test_0100_object_get(self):

        self.client.container = self.c1

        obj = 'test'

        r = self.client.object_get(obj)

        self.assertEqual(r.status_code, 200)

        osize = int(r.headers['content-length'])

        etag = r.headers['etag']

        r = self.client.object_get(obj, hashmap=True)

        for term in ('hashes', 'block_hash', 'block_hash', 'bytes'):

            self.assertTrue(term in r.json)

        r = self.client.object_get(obj, format='xml', hashmap=True)

        self.assertEqual(len(r.text.split('hash>')), 3)

        rangestr = 'bytes=%s-%s' % (osize / 3, osize / 2)

        r = self.client.object_get(

            obj, data_range=rangestr, success=(200, 206))

        partsize = int(r.headers['content-length'])

        self.assertTrue(0 < partsize and partsize <= 1 + osize / 3)

        rangestr = 'bytes=%s-%s' % (osize / 3, osize / 2)

        r = self.client.object_get(

            obj, data_range=rangestr, if_range=True, success=(200, 206))

        partsize = int(r.headers['content-length'])

        self.assertTrue(0 < partsize and partsize <= 1 + osize / 3)

        r = self.client.object_get(obj, if_etag_match=etag)

        self.assertEqual(r.status_code, 200)

        r = self.client.object_get(obj, if_etag_not_match=etag + 'LALALA')

        self.assertEqual(r.status_code, 200)

        """Check and if(un)modified_since"""

        for format in self.client.DATE_FORMATS:

            now_formated = self.now_unformated.strftime(format)

            r1 = self.client.object_get(

                obj, if_modified_since=now_formated, success=(200, 304, 412))

            sc1 = r1.status_code

            r2 = self.client.object_get(

                obj, if_unmodified_since=now_formated, success=(200, 304, 412))

            sc2 = r2.status_code

            self.assertNotEqual(sc1, sc2)

        """Upload an object to download"""

        container_info_cache = dict()

        trg_fname = 'remotefile_%s' % self.now

        f_size = 59247824

        src_f = self.create_large_file(f_size)

        print('\tUploading...')

        r = self.client.upload_object(

            trg_fname, src_f, container_info_cache=container_info_cache)

        print('\tDownloading...')

        self.files.append(NamedTemporaryFile())

        dnl_f = self.files[-1]

        self.client.download_object(trg_fname, dnl_f)

        print('\tCheck if files match...')

        for pos in (0, f_size / 2, f_size - 128):

            src_f.seek(pos)

            dnl_f.seek(pos)

            self.assertEqual(src_f.read(64), dnl_f.read(64))

        print('\tDownload KiBs to string and check again...')

        for pos in (0, f_size / 2, f_size - 256):

            src_f.seek(pos)

            tmp_s = self.client.download_to_string(

                trg_fname, range_str='%s-%s' % (pos, (pos + 128)))

            self.assertEqual(tmp_s, src_f.read(len(tmp_s)))

        print('\tUploading KiBs as strings...')

        trg_fname = 'fromString_%s' % self.now

        src_size = 2 * 1024

        src_f.seek(0)

        src_str = src_f.read(src_size)

        self.client.upload_from_string(trg_fname, src_str)

        print('\tDownload as string and check...')

        tmp_s = self.client.download_to_string(trg_fname)

        self.assertEqual(tmp_s, src_str)

        """Upload a boring file"""

        trg_fname = 'boringfile_%s' % self.now

        src_f = self.create_boring_file(42)

        print('\tUploading boring file...')

        self.client.upload_object(

            trg_fname, src_f, container_info_cache=container_info_cache)

        print('\tDownloading boring file...')

        self.files.append(NamedTemporaryFile())

        dnl_f = self.files[-1]

        self.client.download_object(trg_fname, dnl_f)

        print('\tCheck if files match...')

        for i in range(42):

            self.assertEqual(sample_block(src_f, i), sample_block(dnl_f, i))

    def test_object_put(self):

        """Test object_PUT"""

        self._test_object_put()

    def _test_object_put(self):

        self.client.container = self.c2

        obj = 'another.test'

        self.client.create_object(obj + '.FAKE')

        r = self.client.get_object_info(obj + '.FAKE')

        self.assertEqual(r['content-type'], 'application/octet-stream')

        """create the object"""

        r = self.client.object_put(

            obj,

            data='a',

            content_type='application/octer-stream',

            permissions=dict(

                read=['accX:groupA', 'u1', 'u2'],

                write=['u2', 'u3']),

            metadata=dict(key1='val1', key2='val2'),

            content_encoding='UTF-8',

            content_disposition='attachment; filename="fname.ext"')

        self.assertEqual(r.status_code, 201)

        etag = r.headers['etag']

        """Check content-disposition"""

        r = self.client.get_object_info(obj)

        self.assertTrue('content-disposition' in r)

        """Check permissions"""

        r = self.client.get_object_sharing(obj)

        self.assertTrue('accx:groupa' in r['read'])

        self.assertTrue('u1' in r['read'])

        self.assertTrue('u2' in r['write'])

        self.assertTrue('u3' in r['write'])

        """Check metadata"""

        r = self.client.get_object_meta(obj)

        self.assertEqual(r['x-object-meta-key1'], 'val1')

        self.assertEqual(r['x-object-meta-key2'], 'val2')

        """Check public and if_etag_match"""

        r = self.client.object_put(

            obj,

            if_etag_match=etag,

            data='b',

            content_type='application/octet-stream',

            public=True)