Synnefo

import django.utils.simplejson as json

        cnames = ['apples', 'bananas', 'kiwis']

        # create containers

        uploaded_bytes = 0

        for cname in cnames:

            self.create_container(cname)

            # upload object

            name, data, resp = self.upload_object(cname)

            uploaded_bytes += len(data)

        # add containers

        cnames = ['oranges', 'pears']

        for cname in cnames:

            self.create_container(cname)

            # upload object

            self.upload_object(cname)

        self.assertTrue('X-Account-Container-Count' in account_info)

        self.assertEqual(int(account_info['X-Account-Container-Count']), 3)

        self.assertTrue('X-Account-Bytes-Used' in account_info)

    def test_list_until(self):

        account_info = self.get_account_info()

        t = datetime.datetime.strptime(account_info['Last-Modified'],

                                       DATE_FORMATS[2])

        t1 = t + datetime.timedelta(seconds=1)

        until = int(_time.mktime(t1.timetuple()))

        _time.sleep(2)

        self.create_container()

        url = join_urls(self.pithos_path, self.user)

        r = self.get('%s?until=%s' % (url, until))

        self.assertEqual(r.status_code, 200)

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

        if '' in containers:

            containers.remove('')

        self.assertEqual(containers,

                         ['apples', 'bananas', 'kiwis', 'oranges', 'pears'])

        r = self.get('%s?until=%s&format=json' % (url, until))

        self.assertEqual(r.status_code, 200)

        try:

            containers = json.loads(r.content)

        except:

            self.fail('json format expected')

        self.assertEqual([c['name'] for c in containers],

                         ['apples', 'bananas', 'kiwis', 'oranges', 'pears']) 

    def test_get_container_meta_until(self):

        self.create_container('apples')

        # populate with objects

        objects = {}

        metalist = []

        for i in range(random.randint(1, 100)):

            # upload object

            metakey = 'foo%s' % i

            meta = {metakey: 'bar'}

            name, data, resp = self.upload_object('apples', **meta)

            objects[name] = data

            metalist.append(metakey) 

        self.update_container_meta('apples', {'foo': 'bar'})

        container_info = self.get_container_info('apples')

        t = datetime.datetime.strptime(container_info['Last-Modified'],

                                       DATE_FORMATS[2])

        t1 = t + datetime.timedelta(seconds=1)

        until = int(_time.mktime(t1.timetuple()))

        _time.sleep(2)

        for i in range(random.randint(1, 100)):

            # upload object

            meta = {'foo%s' % i: 'bar'}

            self.upload_object('apples', **meta)

        self.update_container_meta('apples', {'quality': 'AAA'})

        container_info = self.get_container_info('apples')

        self.assertTrue('X-Container-Meta-Quality' in container_info)

        self.assertTrue('X-Container-Meta-Foo' in container_info)

        self.assertTrue('X-Container-Object-Count' in container_info)

        self.assertTrue(int(container_info['X-Container-Object-Count']) > len(objects))

        self.assertTrue('X-Container-Bytes-Used' in container_info)

        t = datetime.datetime.strptime(container_info['Last-Modified'],

                                       DATE_FORMATS[-1])

        last_modified = int(_time.mktime(t.timetuple()))

        assert until < last_modified

        container_info = self.get_container_info('apples', until=until)

        self.assertTrue('X-Container-Meta-Quality' not in container_info)

        self.assertTrue('X-Container-Meta-Foo' in container_info)

        self.assertTrue('X-Container-Until-Timestamp' in container_info)

        t = datetime.datetime.strptime(

            container_info['X-Container-Until-Timestamp'], DATE_FORMATS[2])

        self.assertTrue(int(_time.mktime(t1.timetuple())) <= until)

        self.assertTrue('X-Container-Object-Count' in container_info)

        self.assertEqual(int(container_info['X-Container-Object-Count']), len(objects))

        self.assertTrue('X-Container-Bytes-Used' in container_info)

        self.assertEqual(int(container_info['X-Container-Bytes-Used']),

                         sum([len(data) for data in objects.values()]))

        self.assertTrue('X-Container-Object-Meta' in container_info)

        self.assertEqual(container_info['X-Container-Object-Meta'],

                         metalist) 

    def test_list_until(self):

        account_info = self.get_account_info()

        t = datetime.datetime.strptime(account_info['Last-Modified'],

                                       DATE_FORMATS[2])

        t1 = t + datetime.timedelta(seconds=1)

        until = int(_time.mktime(t1.timetuple()))

        _time.sleep(2)

        cname = self.cnames[0]

        self.upload_object(cname)

        url = join_urls(self.pithos_path, self.user, cname)

        r = self.get('%s?until=%s' % (url, until))

        self.assertTrue(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertEqual(objects,

                         sorted(self.objects[cname].keys()))

        r = self.get('%s?until=%s&format=json' % (url, until))

        self.assertTrue(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

        self.assertEqual([o['name'] for o in objects],

                         sorted(self.objects[cname].keys()))