Synnefo » Pithos

#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 django.utils import simplejson as json

from xml.dom import minidom

from StringIO import StringIO

import unittest

import time as _time

import types

import hashlib

import os

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

DEFAULT_HOST = 'pithos.dev.grnet.gr'

#DEFAULT_HOST = '127.0.0.1:8000'

DEFAULT_API = 'v1'

DEFAULT_USER = 'test'

DEFAULT_AUTH = '0000'

OTHER_ACCOUNTS = {

    '0001': 'verigak',

    '0002': 'chazapis',

    '0003': 'gtsouk',

    '0004': 'papagian',

    '0005': 'louridas',

    '0006': 'chstath',

    '0007': 'pkanavos',

    '0008': 'mvasilak'}

class BaseTestCase(unittest.TestCase):

    #TODO unauthorized request

    def setUp(self):

        self.client = Pithos_Client(DEFAULT_HOST,

                                    DEFAULT_AUTH,

                                    DEFAULT_USER,

                                    DEFAULT_API)

        self.invalid_client = Pithos_Client(DEFAULT_HOST,

                                                  DEFAULT_AUTH,

                                                  'invalid',

                                                  DEFAULT_API)

        #self.headers = {

        #    'account': ('x-account-container-count',

        #                'x-account-bytes-used',

        #                'last-modified',

        #                'content-length',

        #                'date',

        #                'content_type',

        #                'server',),

        #    'object': ('etag',

        #               'content-length',

        #               'content_type',

        #               'content-encoding',

        #               'last-modified',

        #               'date',

        #               'x-object-manifest',

        #               'content-range',

        #               'x-object-modified-by',

        #               'x-object-version',

        #               'x-object-version-timestamp',

        #               'server',),

        #    'container': ('x-container-object-count',

        #                  'x-container-bytes-used',

        #                  'content_type',

        #                  'last-modified',

        #                  'content-length',

        #                  'date',

        #                  'x-container-block-size',

        #                  'x-container-block-hash',

        #                  'x-container-policy-quota',

        #                  'x-container-policy-versioning',

        #                  'server',

        #                  'x-container-object-meta',

        #                  'x-container-policy-versioning',

        #                  'server',)}

        #

        #self.contentTypes = {'xml':'application/xml',

        #                     'json':'application/json',

        #                     '':'text/plain'}

        self.extended = {

            'container':(

                'name',

                'count',

                'bytes',

                'last_modified',

                'x_container_policy_quota',

                'x_container_policy_versioning',),

            'object':(

                'name',

                'hash',

                'bytes',

                'content_type',

                'content_encoding',

                'last_modified',)}

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

    def tearDown(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_headers(self, headers, type, **exp_meta):

    #    prefix = 'x-%s-meta-' %type

    #    system_headers = [h for h in headers if not h.startswith(prefix)]

    #    for k,v in headers.items():

    #        if k in system_headers:

    #            self.assertTrue(k in headers[type])

    #        elif exp_meta:

    #            k = k.split(prefix)[-1]

    #            self.assertEqual(v, exp_meta[k])

    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:

            self.failUnless(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_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 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, StringIO(obj['data']),

                                      meta, **args)

            return obj

        except IOError:

            return

class TopLevel(BaseTestCase):

    def test_list_shared_by_other(self):

        pass

class AccountHead(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.account = 'test'

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

        for item in self.containers:

            self.client.create_container(item)

                #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)

        meta = {'foo':'bar'}

        self.client.update_account_metadata(**meta)

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

    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)

        #print '#', self.client.retrieve_account_groups()

        #print '#', self.client.retrieve_account_metadata(restricted=True)

        BaseTestCase.tearDown(self)

    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_401(self):

        self.assert_raises_fault(401,

                                 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)

        self.account = 'test'

        #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_401(self):

        self.assert_raises_fault(401, 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.account = 'test'

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

        for item in self.containers:

            self.client.create_container(item)

        #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)

        meta = {'foo':'bar'}

        self.client.update_account_metadata(**meta)

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

    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)

        #print '#', self.client.retrieve_account_groups()

        #print '#', self.client.retrieve_account_metadata(restricted=True)

        BaseTestCase.tearDown(self)

    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(401,

                                 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(groups, self.client.retrieve_account_groups())

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

            self.client.set_account_groups(**more_groups)

            groups.update(more_groups)

            self.assertEqual(groups, self.client.retrieve_account_groups())

    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(groups, self.client.retrieve_account_groups())

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

            self.client.reset_account_groups(**groups)

            self.assertTrue(groups, self.client.retrieve_account_groups())

    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.account = 'test'

        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.account = 'test'

        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')

        #objects = self.client.list_objects('test', prefix='/')

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

        #

        #objects = self.client.list_objects('test', path='/')

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

        #

        #objects = self.client.list_objects('test', prefix='/', delimiter='n')

        #self.assertEqual(objects, ['/object'])

    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 = len(l) >= end and end or 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' % 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: