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

import os

import tempfile

from kamaki.clients import tests, ClientError

from kamaki.clients.pithos import PithosClient

class Pithos(tests.Generic):

    fnames = []

    def setUp(self):

        self.client = PithosClient(

            self['store', 'url'],

            self['store', 'token'],

            self['store', 'account'])

        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': 'someUser'})

        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 fname in self.fnames:

            try:

                os.remove(fname)

            except OSError:

                pass

        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

            r1.release()

            r2 = self.client.account_head(if_unmodified_since=now_formated,

                success=(204, 304, 412))

            sc2 = r2.status_code

            r2.release()

            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)

        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

            r1.release()

            r2 = self.client.account_get(if_unmodified_since=now_formated,

                success=(200, 304, 412))

            sc2 = r2.status_code

            r2.release()

            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

        """

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

        r = self.client.get_account_group()

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

        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

            r1.release()

            r2 = self.client.container_head(if_unmodified_since=now_formated,

                success=(204, 304, 412))

            sc2 = r2.status_code

            r2.release()

            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

            r1.release()

            r2 = self.client.container_get(if_unmodified_since=now_formated,

                success=(200, 304, 412))

            sc2 = r2.status_code

            r2.release()

            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.container_put()

        self.assertEqual(r.status_code, 202)

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

        cquota = r.values()[0]

        newquota = 2 * int(cquota)

        r = self.client.container_put(quota=newquota)

        self.assertEqual(r.status_code, 202)

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

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

        self.assertEqual(newquota, xquota)

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

        self.assertEqual(r.status_code, 202)

        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.container_put(metadata={'m1': 'v1', 'm2': 'v2'})

        self.assertEqual(r.status_code, 202)

        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_quota(self.client.container)

        cquota = r.values()[0]

        newquota = 2 * int(cquota)

        r = self.client.set_container_quota(newquota)

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

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

        self.assertEqual(newquota, xquota)

        r = self.client.set_container_quota(cquota)

        r = self.client.get_container_quota(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"""

        fname = 'l100M.' + unicode(self.now)

        self.create_large_file(1024 * 1024 * 100, fname)

        self.fnames.append(fname)

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

        self.client.create_directory('dir')

        newf = open(fname, 'rb')

        self.client.upload_object('/dir/sample.file', newf)

        newf.close()

        """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']

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

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

        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=40,

            if_etag_match=etag,

            success=412)

        self.assertEqual(r.status_code, 412)

        """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

            r1.release()

            r2 = self.client.object_head(obj, if_unmodified_since=now_formated,

                success=(200, 304, 412))

            sc2 = r2.status_code

            r2.release()

            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)

        self.assertTrue('hashes' in r.json\

            and 'block_hash' in r.json\

            and 'block_size' in r.json\

            and 'bytes' 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

            r1.release()

            r2 = self.client.object_get(obj,

                if_unmodified_since=now_formated,

                success=(200, 304, 412))

            sc2 = r2.status_code

            r2.release()

            self.assertNotEqual(sc1, sc2)

        """Upload an object to download"""

        src_file = tempfile.NamedTemporaryFile(delete=False)

        dnl_file = tempfile.NamedTemporaryFile(delete=False)

        src_fname = src_file.name

        dnl_fname = dnl_file.name

        src_file.close()

        dnl_file.close()

        trg_fname = 'remotefile_%s' % self.now

        f_size = 59247824

        self.create_large_file(f_size, src_fname)

        src_f = open(src_fname, 'rb+')

        print('\tUploading...')

        self.client.upload_object(trg_fname, src_f)

        src_f.close()

        print('\tDownloading...')

        dnl_f = open(dnl_fname, 'wb+')

        self.client.download_object(trg_fname, dnl_f)

        dnl_f.close()

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

        src_f = open(src_fname)

        dnl_f = open(dnl_fname)

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

            src_f.seek(pos)

            dnl_f.seek(pos)

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

        src_f.close()

        dnl_f.close()

        os.remove(src_fname)

        os.remove(dnl_fname)

    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={

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

                'write': ['u2', 'u3']},

            metadata={'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)

        r = self.client.object_get(obj)

        self.assertTrue('x-object-public' in r.headers)

        vers2 = int(r.headers['x-object-version'])

        etag = r.headers['etag']

        self.assertEqual(r.text, 'b')

        """Check if_etag_not_match"""

        r = self.client.object_put(obj, if_etag_not_match=etag, data='c',

            content_type='application/octet-stream', success=(201, 412))

        self.assertEqual(r.status_code, 412)

        """Check content_type and content_length"""

        tmpdir = 'dir' + unicode(self.now)

        r = self.client.object_put(tmpdir,

            content_type='application/directory',

            content_length=0)

        r = self.client.get_object_info(tmpdir)

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

        """Check copy_from, content_encoding"""

        r = self.client.object_put('%s/%s' % (tmpdir, obj),

            format=None,

            copy_from='/%s/%s' % (self.client.container, obj),

            content_encoding='application/octet-stream',

            source_account=self.client.account,

            content_length=0, success=201)

        self.assertEqual(r.status_code, 201)

        """Test copy_object for cross-conctainer copy"""

        self.client.copy_object(src_container=self.c2,

            src_object='%s/%s' % (tmpdir, obj),

            dst_container=self.c1,

            dst_object=obj)

        self.client.container = self.c1

        r1 = self.client.get_object_info(obj)

        self.client.container = self.c2

        r2 = self.client.get_object_info('%s/%s' % (tmpdir, obj))

        self.assertEqual(r1['x-object-hash'], r2['x-object-hash'])

        """Check cross-container copy_from, content_encoding"""

        self.client.container = self.c1