Synnefo » Pithos

#!/usr/bin/env python

#coding=utf8

# Copyright 2011 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.

from pithos.lib.client import Pithos_Client, Fault

from pithos.lib.util import get_user, get_auth, get_server, get_api

from xml.dom import minidom

from StringIO import StringIO

from hashlib import new as newhasher

from binascii import hexlify

import json

import unittest

import time as _time

import types

import hashlib

import mimetypes

import random

import datetime

import string

import re

DATE_FORMATS = ["%a %b %d %H:%M:%S %Y",

                "%A, %d-%b-%y %H:%M:%S GMT",

                "%a, %d %b %Y %H:%M:%S GMT"]

OTHER_ACCOUNTS = {

    '0001': 'verigak',

    '0002': 'chazapis',

    '0003': 'gtsouk',

    '0004': 'papagian',

    '0005': 'louridas',

    '0006': 'chstath',

    '0007': 'pkanavos',

    '0008': 'mvasilak',

    '0009': 'διογένης'}

class BaseTestCase(unittest.TestCase):

    #TODO unauthorized request

    def setUp(self):

        self.client = Pithos_Client(get_server(), get_auth(), get_user(),

                                    get_api())

        self._clean_account()

        self.invalid_client = Pithos_Client(get_server(), get_auth(), 'invalid',

                                            get_api())

        #keep track of initial account groups

        self.initial_groups = self.client.retrieve_account_groups()

        #keep track of initial account meta

        self.initial_meta = self.client.retrieve_account_metadata(restricted=True)

        self.extended = {

            'container':(

                'name',

                'count',

                'bytes',

                'last_modified',

                'x_container_policy'),

            'object':(

                'name',

                'hash',

                'bytes',

                'content_type',

                'content_encoding',

                'last_modified',)}

        self.return_codes = (400, 401, 403, 404, 503,)

    def tearDown(self):

        #delete additionally created meta

        l = []

        for m in self.client.retrieve_account_metadata(restricted=True):

            if m not in self.initial_meta:

                l.append(m)

        self.client.delete_account_metadata(l)

        #delete additionally created groups

        l = []

        for g in self.client.retrieve_account_groups():

            if g not in self.initial_groups:

                l.append(g)

        self.client.unset_account_groups(l)

        self._clean_account()

    def _clean_account(self):

        for c in self.client.list_containers():

            while True:

                #list objects returns at most 10000 objects

                #so repeat until there are no more objects

                objects = self.client.list_objects(c)

                if not objects:

                    break

                for o in objects:

                    self.client.delete_object(c, o)

            self.client.delete_container(c)

    def assert_status(self, status, codes):

        l = [elem for elem in self.return_codes]

        if type(codes) == types.ListType:

            l.extend(codes)

        else:

            l.append(codes)

        self.assertTrue(status in l)

    def assert_extended(self, data, format, type, size=10000):

        if format == 'xml':

            self._assert_xml(data, type, size)

        elif format == 'json':

            self._assert_json(data, type, size)

    def _assert_json(self, data, type, size):

        convert = lambda s: s.lower()

        info = [convert(elem) for elem in self.extended[type]]

        self.assertTrue(len(data) <= size)

        for item in info:

            for i in data:

                if 'subdir' in i.keys():

                    continue

                self.assertTrue(item in i.keys())

    def _assert_xml(self, data, type, size):

        convert = lambda s: s.lower()

        info = [convert(elem) for elem in self.extended[type]]

        try:

            info.remove('content_encoding')

        except ValueError:

            pass

        xml = data

        entities = xml.getElementsByTagName(type)

        self.assertTrue(len(entities) <= size)

        for e in entities:

            for item in info:

                self.assertTrue(e.getElementsByTagName(item))

    def assert_raises_fault(self, status, callableObj, *args, **kwargs):

        """

        asserts that a Fault with a specific status is raised

        when callableObj is called with the specific arguments

        """

        try:

            r = callableObj(*args, **kwargs)

            self.fail('Should never reach here')

        except Fault, f:

            if type(status) == types.ListType:

                self.failUnless(f.status in status)

            else:

                self.failUnless(f.status == status)

    def assert_not_raises_fault(self, status, callableObj, *args, **kwargs):

        """

        asserts that a Fault with a specific status is not raised

        when callableObj is called with the specific arguments

        """

        try:

            r = callableObj(*args, **kwargs)

        except Fault, f:

            self.failIfEqual(f.status, status)

    def assert_container_exists(self, container):

        """

        asserts the existence of a container

        """

        try:

            self.client.retrieve_container_metadata(container)

        except Fault, f:

            self.failIf(f.status == 404)

    def assert_container_not_exists(self, container):

        """

        asserts there is no such a container

        """

        self.assert_raises_fault(404, self.client.retrieve_container_metadata,

                                 container)

    def assert_object_exists(self, container, object):

        """

        asserts the existence of an object

        """

        try:

            self.client.retrieve_object_metadata(container, object)

        except Fault, f:

            self.failIf(f.status == 404)

    def assert_object_not_exists(self, container, object):

        """

        asserts there is no such an object

        """

        self.assert_raises_fault(404, self.client.retrieve_object_metadata,

                                 container, object)

    def assert_versionlist_structure(self, versionlist):

        self.assertTrue(type(versionlist) == types.ListType)

        for elem in versionlist:

            self.assertTrue(type(elem) == types.ListType)

            self.assertEqual(len(elem), 2)

    def upload_random_data(self, container, name, length=1024, type=None,

                           enc=None, **meta):

        data = get_random_data(length)

        return self.upload_data(container, name, data, type, enc, **meta)

    def upload_data(self, container, name, data, type=None, enc=None, etag=None,

                    **meta):

        obj = {}

        obj['name'] = name

        try:

            obj['data'] = data

            obj['hash'] = compute_md5_hash(obj['data'])

            args = {}

            args['etag'] = etag if etag else obj['hash']

            try:

                guess = mimetypes.guess_type(name)

                type = type if type else guess[0]

                enc = enc if enc else guess[1]

            except:

                pass

            args['content_type'] = type if type else 'plain/text'

            args['content_encoding'] = enc if enc else None

            obj['meta'] = args

            path = '/%s/%s' % (container, name)

            self.client.create_object(container, name, f=StringIO(obj['data']),

                                      meta=meta, **args)

            return obj

        except IOError:

            return

class AccountHead(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.containers = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']

        for item in self.containers:

            self.client.create_container(item)

        meta = {'foo':'bar'}

        self.client.update_account_metadata(**meta)

        #self.updated_meta = self.initial_meta.update(meta)

    def test_get_account_meta(self):

        meta = self.client.retrieve_account_metadata()

        containers = self.client.list_containers()

        l = str(len(containers))

        self.assertEqual(meta['x-account-container-count'], l)

        size = 0

        for c in containers:

            m = self.client.retrieve_container_metadata(c)

            size = size + int(m['x-container-bytes-used'])

        self.assertEqual(meta['x-account-bytes-used'], str(size))

    def test_get_account_403(self):

        self.assert_raises_fault(403,

                                 self.invalid_client.retrieve_account_metadata)

    def test_get_account_meta_until(self):

        t = datetime.datetime.utcnow()

        past = t - datetime.timedelta(minutes=-15)

        past = int(_time.mktime(past.timetuple()))

        meta = {'premium':True}

        self.client.update_account_metadata(**meta)

        meta = self.client.retrieve_account_metadata(restricted=True,

                                                     until=past)

        self.assertTrue('premium' not in meta)

        meta = self.client.retrieve_account_metadata(restricted=True)

        self.assertTrue('premium' in meta)

    def test_get_account_meta_until_invalid_date(self):

        meta = {'premium':True}

        self.client.update_account_metadata(**meta)

        meta = self.client.retrieve_account_metadata(restricted=True,

                                                     until='kshfksfh')

        self.assertTrue('premium' in meta)

class AccountGet(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        #create some containers

        self.containers = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']

        for item in self.containers:

            self.client.create_container(item)

    def test_list(self):

        #list containers

        containers = self.client.list_containers()

        self.assertEquals(self.containers, containers)

    def test_list_403(self):

        self.assert_raises_fault(403, self.invalid_client.list_containers)

    def test_list_with_limit(self):

        limit = 2

        containers = self.client.list_containers(limit=limit)

        self.assertEquals(len(containers), limit)

        self.assertEquals(self.containers[:2], containers)

    def test_list_with_marker(self):

        l = 2

        m = 'bananas'

        containers = self.client.list_containers(limit=l, marker=m)

        i = self.containers.index(m) + 1

        self.assertEquals(self.containers[i:(i+l)], containers)

        m = 'oranges'

        containers = self.client.list_containers(limit=l, marker=m)

        i = self.containers.index(m) + 1

        self.assertEquals(self.containers[i:(i+l)], containers)

    def test_list_json_with_marker(self):

        l = 2

        m = 'bananas'

        containers = self.client.list_containers(limit=l, marker=m, format='json')

        self.assert_extended(containers, 'json', 'container', l)

        self.assertEqual(containers[0]['name'], 'kiwis')

        self.assertEqual(containers[1]['name'], 'oranges')

    def test_list_xml_with_marker(self):

        l = 2

        m = 'oranges'

        xml = self.client.list_containers(limit=l, marker=m, format='xml')

        self.assert_extended(xml, 'xml', 'container', l)

        nodes = xml.getElementsByTagName('name')

        self.assertEqual(len(nodes), 1)

        self.assertEqual(nodes[0].childNodes[0].data, 'pears')

    def test_if_modified_since(self):

        t = datetime.datetime.utcnow()

        t2 = t - datetime.timedelta(minutes=10)

        #add a new container

        self.client.create_container('dummy')

        for f in DATE_FORMATS:

            past = t2.strftime(f)

            try:

                c = self.client.list_containers(if_modified_since=past)

                self.assertEqual(len(c), len(self.containers) + 1)

            except Fault, f:

                self.failIf(f.status == 304) #fail if not modified

    def test_if_modified_since_invalid_date(self):

        c = self.client.list_containers(if_modified_since='')

        self.assertEqual(len(c), len(self.containers))

    def test_if_not_modified_since(self):

        now = datetime.datetime.utcnow()

        since = now + datetime.timedelta(1)

        for f in DATE_FORMATS:

            args = {'if_modified_since':'%s' %since.strftime(f)}

            #assert not modified

            self.assert_raises_fault(304, self.client.list_containers, **args)

    def test_if_unmodified_since(self):

        now = datetime.datetime.utcnow()

        since = now + datetime.timedelta(1)

        for f in DATE_FORMATS:

            c = self.client.list_containers(if_unmodified_since=since.strftime(f))

            #assert success

            self.assertEqual(self.containers, c)

    def test_if_unmodified_since_precondition_failed(self):

        t = datetime.datetime.utcnow()

        t2 = t - datetime.timedelta(minutes=10)

        #add a new container

        self.client.create_container('dummy')

        for f in DATE_FORMATS:

            past = t2.strftime(f)

            args = {'if_unmodified_since':'%s' %past}

            #assert precondition failed

            self.assert_raises_fault(412, self.client.list_containers, **args)

class AccountPost(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.containers = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']

        for item in self.containers:

            self.client.create_container(item)

        meta = {'foo':'bar'}

        self.client.update_account_metadata(**meta)

        self.updated_meta = self.initial_meta.update(meta)

    def test_update_meta(self):

        with AssertMappingInvariant(self.client.retrieve_account_groups):

            meta = {'test':'test', 'tost':'tost'}

            self.client.update_account_metadata(**meta)

            meta.update(self.initial_meta)

            self.assertEqual(meta,

                             self.client.retrieve_account_metadata(

                                restricted=True))

    def test_invalid_account_update_meta(self):

        meta = {'test':'test', 'tost':'tost'}

        self.assert_raises_fault(403,

                                 self.invalid_client.update_account_metadata,

                                 **meta)

    def test_reset_meta(self):

        with AssertMappingInvariant(self.client.retrieve_account_groups):

            meta = {'test':'test', 'tost':'tost'}

            self.client.update_account_metadata(**meta)

            meta = {'test':'test33'}

            self.client.reset_account_metadata(**meta)

            self.assertEqual(meta, self.client.retrieve_account_metadata(restricted=True))

    def test_delete_meta(self):

        with AssertMappingInvariant(self.client.retrieve_account_groups):

            meta = {'test':'test', 'tost':'tost'}

            self.client.update_account_metadata(**meta)

            self.client.delete_account_metadata(meta.keys())

            account_meta = self.client.retrieve_account_metadata(restricted=True)

            for m in meta:

                self.assertTrue(m not in account_meta.keys())

    def test_set_account_groups(self):

        with AssertMappingInvariant(self.client.retrieve_account_metadata):

            groups = {'pithosdev':'verigak,gtsouk,chazapis'}

            self.client.set_account_groups(**groups)

            self.assertEqual(set(groups['pithosdev']),

                             set(self.client.retrieve_account_groups()['pithosdev']))

            more_groups = {'clientsdev':'pkanavos,mvasilak'}

            self.client.set_account_groups(**more_groups)

            groups.update(more_groups)

            self.assertEqual(set(groups['clientsdev']),

                             set(self.client.retrieve_account_groups()['clientsdev']))

    def test_reset_account_groups(self):

        with AssertMappingInvariant(self.client.retrieve_account_metadata):

            groups = {'pithosdev':'verigak,gtsouk,chazapis',

                      'clientsdev':'pkanavos,mvasilak'}

            self.client.set_account_groups(**groups)

            self.assertEqual(set(groups['pithosdev'].split(',')),

                             set(self.client.retrieve_account_groups()['pithosdev'].split(',')))

            self.assertEqual(set(groups['clientsdev'].split(',')),

                             set(self.client.retrieve_account_groups()['clientsdev'].split(',')))

            groups = {'pithosdev':'verigak,gtsouk,chazapis,papagian'}

            self.client.reset_account_groups(**groups)

            self.assertEqual(set(groups['pithosdev'].split(',')),

                             set(self.client.retrieve_account_groups()['pithosdev'].split(',')))

    def test_delete_account_groups(self):

        with AssertMappingInvariant(self.client.retrieve_account_metadata):

            groups = {'pithosdev':'verigak,gtsouk,chazapis',

                      'clientsdev':'pkanavos,mvasilak'}

            self.client.set_account_groups(**groups)

            self.client.unset_account_groups(groups.keys())

            self.assertEqual({}, self.client.retrieve_account_groups())

class ContainerHead(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.container = 'apples'

        self.client.create_container(self.container)

    def test_get_meta(self):

        meta = {'trash':'true'}

        t1 = datetime.datetime.utcnow()

        o = self.upload_random_data(self.container, o_names[0], **meta)

        if o:

            headers = self.client.retrieve_container_metadata(self.container)

            self.assertEqual(headers['x-container-object-count'], '1')

            self.assertEqual(headers['x-container-bytes-used'], str(len(o['data'])))

            t2 = datetime.datetime.strptime(headers['last-modified'], DATE_FORMATS[2])

            delta = (t2 - t1)

            threashold = datetime.timedelta(seconds=1) 

            self.assertTrue(delta < threashold)

            self.assertTrue(headers['x-container-object-meta'])

            self.assertTrue('Trash' in headers['x-container-object-meta'])

class ContainerGet(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.container = ['pears', 'apples']

        for c in self.container:

            self.client.create_container(c)

        self.obj = []

        for o in o_names[:8]:

            self.obj.append(self.upload_random_data(self.container[0], o))

        for o in o_names[8:]:

            self.obj.append(self.upload_random_data(self.container[1], o))

    def test_list_objects(self):

        objects = self.client.list_objects(self.container[0])

        l = [elem['name'] for elem in self.obj[:8]]

        l.sort()

        self.assertEqual(objects, l)

    def test_list_objects_containing_slash(self):

        self.client.create_container('test')

        self.upload_random_data('test', '/objectname')

        objects = self.client.list_objects('test')

        self.assertEqual(objects, ['/objectname'])

        objects = self.client.list_objects('test', format='json')

        self.assertEqual(objects[0]['name'], '/objectname')

        objects = self.client.list_objects('test', format='xml')

        self.assert_extended(objects, 'xml', 'object')

        node_name = objects.getElementsByTagName('name')[0]

        self.assertEqual(node_name.firstChild.data, '/objectname')

    def test_list_objects_with_limit_marker(self):

        objects = self.client.list_objects(self.container[0], limit=2)

        l = [elem['name'] for elem in self.obj[:8]]

        l.sort()

        self.assertEqual(objects, l[:2])

        markers = ['How To Win Friends And Influence People.pdf',

                   'moms_birthday.jpg']

        limit = 4

        for m in markers:

            objects = self.client.list_objects(self.container[0], limit=limit,

                                               marker=m)

            l = [elem['name'] for elem in self.obj[:8]]

            l.sort()

            start = l.index(m) + 1

            end = start + limit

            end = end if len(l) >= end else len(l)

            self.assertEqual(objects, l[start:end])

    #takes too long

    def _test_list_limit_exceeds(self):

        self.client.create_container('pithos')

        for i in range(10001):

            self.client.create_zero_length_object('pithos', i)

        self.assertEqual(10000, len(self.client.list_objects('pithos')))

    def test_list_empty_params(self):

        objects = self.client.get('/%s/%s' % (get_user(), self.container[0]))[2]

        if objects:

            objects = objects.strip().split('\n')

        self.assertEqual(objects,

                         self.client.list_objects(self.container[0]))

    def test_list_pseudo_hierarchical_folders(self):

        objects = self.client.list_objects(self.container[1], prefix='photos',

                                           delimiter='/')

        self.assertEquals(['photos/animals/', 'photos/me.jpg',

                           'photos/plants/'], objects)

        objects = self.client.list_objects(self.container[1],

                                           prefix='photos/animals',

                                           delimiter='/')

        l = ['photos/animals/cats/', 'photos/animals/dogs/']

        self.assertEquals(l, objects)

        objects = self.client.list_objects(self.container[1], path='photos')

        self.assertEquals(['photos/me.jpg'], objects)

    def test_extended_list_json(self):

        objects = self.client.list_objects(self.container[1], format='json',

                                           limit=2, prefix='photos/animals',

                                           delimiter='/')

        self.assertEqual(objects[0]['subdir'], 'photos/animals/cats/')

        self.assertEqual(objects[1]['subdir'], 'photos/animals/dogs/')

    def test_extended_list_xml(self):

        xml = self.client.list_objects(self.container[1], format='xml', limit=4,

                                       prefix='photos', delimiter='/')

        self.assert_extended(xml, 'xml', 'object', size=4)

        dirs = xml.getElementsByTagName('subdir')

        self.assertEqual(len(dirs), 2)

        self.assertEqual(dirs[0].attributes['name'].value, 'photos/animals/')

        self.assertEqual(dirs[1].attributes['name'].value, 'photos/plants/')

        objects = xml.getElementsByTagName('name')

        self.assertEqual(len(objects), 1)

        self.assertEqual(objects[0].childNodes[0].data, 'photos/me.jpg')

    def test_list_meta_double_matching(self):

        meta = {'quality':'aaa', 'stock':'true'}

        self.client.update_object_metadata(self.container[0],

                                           self.obj[0]['name'], **meta)

        obj = self.client.list_objects(self.container[0], meta='Quality,Stock')

        self.assertEqual(len(obj), 1)

        self.assertTrue(obj, self.obj[0]['name'])

    def test_list_using_meta(self):

        meta = {'quality':'aaa'}

        for o in self.obj[:2]:

            self.client.update_object_metadata(self.container[0], o['name'],

                                               **meta)

        meta = {'stock':'true'}

        for o in self.obj[3:5]:

            self.client.update_object_metadata(self.container[0], o['name'],

                                               **meta)

        obj = self.client.list_objects(self.container[0], meta='Quality')

        self.assertEqual(len(obj), 2)

        self.assertTrue(obj, [o['name'] for o in self.obj[:2]])

        # test case insensitive

        obj = self.client.list_objects(self.container[0], meta='quality')

        self.assertEqual(len(obj), 2)

        self.assertTrue(obj, [o['name'] for o in self.obj[:2]])

        # test multiple matches

        obj = self.client.list_objects(self.container[0], meta='Quality,Stock')

        self.assertEqual(len(obj), 4)

        self.assertTrue(obj, [o['name'] for o in self.obj[:4]])

        # test non 1-1 multiple match

        obj = self.client.list_objects(self.container[0], meta='Quality,aaaa')

        self.assertEqual(len(obj), 2)

        self.assertTrue(obj, [o['name'] for o in self.obj[:2]])

    def test_if_modified_since(self):

        t = datetime.datetime.utcnow()

        t2 = t - datetime.timedelta(minutes=10)

        #add a new object

        self.upload_random_data(self.container[0], o_names[0])

        for f in DATE_FORMATS:

            past = t2.strftime(f)

            try:

                o = self.client.list_objects(self.container[0],

                                            if_modified_since=past)

                self.assertEqual(o,

                                 self.client.list_objects(self.container[0]))

            except Fault, f:

                self.failIf(f.status == 304) #fail if not modified

    def test_if_modified_since_invalid_date(self):

        headers = {'if-modified-since':''}

        o = self.client.list_objects(self.container[0], if_modified_since='')

        self.assertEqual(o, self.client.list_objects(self.container[0]))

    def test_if_not_modified_since(self):

        now = datetime.datetime.utcnow()

        since = now + datetime.timedelta(1)

        for f in DATE_FORMATS:

            args = {'if_modified_since':'%s' %since.strftime(f)}

            #assert not modified

            self.assert_raises_fault(304, self.client.list_objects,

                                     self.container[0], **args)

    def test_if_unmodified_since(self):

        now = datetime.datetime.utcnow()

        since = now + datetime.timedelta(1)

        for f in DATE_FORMATS:

            obj = self.client.list_objects(self.container[0],

                                           if_unmodified_since=since.strftime(f))

            #assert unmodified

            self.assertEqual(obj, self.client.list_objects(self.container[0]))

    def test_if_unmodified_since_precondition_failed(self):

        t = datetime.datetime.utcnow()

        t2 = t - datetime.timedelta(minutes=10)

        #add a new container

        self.client.create_container('dummy')

        for f in DATE_FORMATS:

            past = t2.strftime(f)

            args = {'if_unmodified_since':'%s' %past}

            #assert precondition failed

            self.assert_raises_fault(412, self.client.list_objects,

                                     self.container[0], **args)

class ContainerPut(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.containers = ['c1', 'c2']

    def test_create(self):

        self.client.create_container(self.containers[0])

        containers = self.client.list_containers()

        self.assertTrue(self.containers[0] in containers)

        self.assert_container_exists(self.containers[0])

    def test_create_twice(self):

        self.client.create_container(self.containers[0])

        self.assertTrue(not self.client.create_container(self.containers[0]))

    def test_quota(self):

        self.client.create_container(self.containers[0])

        policy = {'quota':100}

        self.client.set_container_policies('c1', **policy)

        meta = self.client.retrieve_container_metadata('c1')

        self.assertTrue('x-container-policy-quota' in meta)

        self.assertEqual(meta['x-container-policy-quota'], '100')

        args = ['c1', 'o1']

        kwargs = {'length':101}

        self.assert_raises_fault(413, self.upload_random_data, *args, **kwargs)

        #reset quota

        policy = {'quota':0}

        self.client.set_container_policies('c1', **policy)

class ContainerPost(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.container = 'apples'

        self.client.create_container(self.container)

    def test_update_meta(self):

        meta = {'test':'test33',

                'tost':'tost22'}

        self.client.update_container_metadata(self.container, **meta)

        headers = self.client.retrieve_container_metadata(self.container)

        for k,v in meta.items():

            k = 'x-container-meta-%s' % k

            self.assertTrue(headers[k])

            self.assertEqual(headers[k], v)

class ContainerDelete(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.containers = ['c1', 'c2']

        for c in self.containers:

            self.client.create_container(c)

    def test_delete(self):

        status = self.client.delete_container(self.containers[0])[0]

        self.assertEqual(status, 204)

    def test_delete_non_empty(self):

        self.upload_random_data(self.containers[1], o_names[0])

        self.assert_raises_fault(409, self.client.delete_container,

                                 self.containers[1])

    def test_delete_invalid(self):

        self.assert_raises_fault(404, self.client.delete_container, 'c3')

class ObjectGet(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.containers = ['c1', 'c2']

        #create some containers

        for c in self.containers:

            self.client.create_container(c)

        #upload a file

        names = ('obj1', 'obj2')

        self.objects = []

        for n in names:

            self.objects.append(self.upload_random_data(self.containers[1], n))

    def test_versions(self):

        c = self.containers[1]

        o = self.objects[0]

        b = self.client.retrieve_object_versionlist(c, o['name'])['versions']

        self.assert_versionlist_structure(b)

        #update meta

        meta = {'quality':'AAA', 'stock':True}

        self.client.update_object_metadata(c, o['name'], **meta)

        a = self.client.retrieve_object_versionlist(c, o['name'])['versions']

        self.assert_versionlist_structure(a)

        self.assertEqual(len(b)+1, len(a))

        self.assertEqual(b, a[:-1])

        #get exact previous version metadata

        v = a[-2][0]

        v_meta = self.client.retrieve_object_metadata(c, o['name'],

                                                      restricted=True,

                                                      version=v)

        for k in meta.keys():

            self.assertTrue(k not in v_meta)

        #update obejct

        data = get_random_data()

        self.client.update_object(c, o['name'], StringIO(data))

        aa = self.client.retrieve_object_versionlist(c, o['name'])['versions']

        self.assert_versionlist_structure(aa)

        self.assertEqual(len(a)+1, len(aa))

        self.assertEqual(a, aa[:-1])

        #get exact previous version

        v = aa[-3][0]

        v_data = self.client.retrieve_object_version(c, o['name'], version=v)

        self.assertEqual(o['data'], v_data)

        self.assertEqual(self.client.retrieve_object(c, o['name']),

                         '%s%s' %(v_data, data))

    def test_get(self):

        #perform get

        o = self.client.retrieve_object(self.containers[1],

                                        self.objects[0]['name'],

                                        self.objects[0]['meta'])

        self.assertEqual(o, self.objects[0]['data'])

    def test_objects_with_trailing_spaces(self):

        self.client.create_container('test')

        #create 'a' object

        self.upload_random_data('test', 'a')

        #look for 'a ' object

        self.assert_raises_fault(404, self.client.retrieve_object,

                                 'test', 'a ')

        #delete 'a' object

        self.client.delete_object('test', 'a')

        self.assert_raises_fault(404, self.client.retrieve_object,

                                 'test', 'a')

        #create 'a ' object

        self.upload_random_data('test', 'a ')

        #look for 'a' object

        self.assert_raises_fault(404, self.client.retrieve_object,

                                 'test', 'a')

    def test_get_invalid(self):

        self.assert_raises_fault(404, self.client.retrieve_object,

                                 self.containers[0], self.objects[0]['name'])

    def test_get_partial(self):

        #perform get with range

        status, headers, data = self.client.request_object(self.containers[1],

                                                            self.objects[0]['name'],

                                                            range='bytes=0-499')

        #assert successful partial content

        self.assertEqual(status, 206)

        #assert content-type

        self.assertEqual(headers['content-type'],

                         self.objects[0]['meta']['content_type'])

        #assert content length

        self.assertEqual(int(headers['content-length']), 500)

        #assert content

        self.assertEqual(self.objects[0]['data'][:500], data)

    def test_get_final_500(self):

        #perform get with range

        headers = {'range':'bytes=-500'}

        status, headers, data = self.client.request_object(self.containers[1],

                                                            self.objects[0]['name'],

                                                            range='bytes=-500')

        #assert successful partial content

        self.assertEqual(status, 206)

        #assert content-type

        self.assertEqual(headers['content-type'],

                         self.objects[0]['meta']['content_type'])

        #assert content length

        self.assertEqual(int(headers['content-length']), 500)

        #assert content

        self.assertTrue(self.objects[0]['data'][-500:], data)

    def test_get_rest(self):

        #perform get with range

        offset = len(self.objects[0]['data']) - 500

        status, headers, data = self.client.request_object(self.containers[1],

                                                self.objects[0]['name'],

                                                range='bytes=%s-' %offset)

        #assert successful partial content

        self.assertEqual(status, 206)

        #assert content-type

        self.assertEqual(headers['content-type'],

                         self.objects[0]['meta']['content_type'])

        #assert content length

        self.assertEqual(int(headers['content-length']), 500)

        #assert content

        self.assertTrue(self.objects[0]['data'][-500:], data)

    def test_get_range_not_satisfiable(self):

        #perform get with range

        offset = len(self.objects[0]['data']) + 1

        #assert range not satisfiable

        self.assert_raises_fault(416, self.client.retrieve_object,

                                 self.containers[1], self.objects[0]['name'],

                                 range='bytes=0-%s' %offset)

    def test_multiple_range(self):

        #perform get with multiple range

        ranges = ['0-499', '-500', '1000-']

        bytes = 'bytes=%s' % ','.join(ranges)

        status, headers, data = self.client.request_object(self.containers[1],

                                                           self.objects[0]['name'],

                                                           range=bytes)

        # assert partial content

        self.assertEqual(status, 206)

        # assert Content-Type of the reply will be multipart/byteranges

        self.assertTrue(headers['content-type'])

        content_type_parts = headers['content-type'].split()

        self.assertEqual(content_type_parts[0], ('multipart/byteranges;'))

        boundary = '--%s' %content_type_parts[1].split('=')[-1:][0]

        cparts = data.split(boundary)[1:-1]

        # assert content parts are exactly 2

        self.assertEqual(len(cparts), len(ranges))

        # for each content part assert headers

        i = 0

        for cpart in cparts:

            content = cpart.split('\r\n')

            headers = content[1:3]

            content_range = headers[0].split(': ')

            self.assertEqual(content_range[0], 'Content-Range')

            r = ranges[i].split('-')

            if not r[0] and not r[1]:

                pass

            elif not r[0]:

                start = len(self.objects[0]['data']) - int(r[1])

                end = len(self.objects[0]['data'])