Synnefo

#!/usr/bin/env python

#coding=utf8

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

from pithos.api.test import (PithosAPITest, DATE_FORMATS, o_names,

                             pithos_settings, pithos_test_settings)

from pithos.api.test.util import strnextling, get_random_data, get_random_name

from synnefo.lib import join_urls

import django.utils.simplejson as json

from django.http import urlencode

from xml.dom import minidom

from urllib import quote

import time as _time

import random

import datetime

class ContainerHead(PithosAPITest):

    def test_get_meta(self):

        self.create_container('apples')

        # populate with objects

        objects = {}

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

            # upload object

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

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

            objects[name] = data

        t1 = datetime.datetime.utcnow()

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

        r = self.head(url)

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

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

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

        self.assertTrue('X-Container-Block-Size' in r)

        self.assertTrue('X-Container-Block-Hash' in r)

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

        self.assertEqual(r['X-Container-Policy-Versioning'], 'auto')

        self.assertEqual(int(r['X-Container-Policy-Quota']), 0)

        t2 = datetime.datetime.strptime(r['Last-Modified'], DATE_FORMATS[2])

        delta = (t2 - t1)

        threashold = datetime.timedelta(seconds=1)

        self.assertTrue(delta < threashold)

        self.assertTrue(r['X-Container-Object-Meta'])

        (self.assertTrue('foo%s' % i in r['X-Container-Object-Meta'])

            for i in range(len(objects)))

    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(

            sorted(container_info['X-Container-Object-Meta'].split(',')),

            sorted(metalist))

class ContainerGet(PithosAPITest):

    def setUp(self):

        PithosAPITest.setUp(self)

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

        self.objects = {}

        for c in self.cnames:

            self.create_container(c)

        self.objects['pears'] = {}

        for o in o_names[:8]:

            name, data, resp = self.upload_object('pears', o)

            self.objects['pears'][name] = data

        self.objects['apples'] = {}

        for o in o_names[8:]:

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

            self.objects['apples'][name] = data

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

    def test_list_shared(self):

        # share an object

        cname = self.cnames[0]

        onames = self.objects[cname].keys()

        shared1 = onames.pop()

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=*')

        self.assertEqual(r.status_code, 202)

        # share another object

        shared2 = onames.pop()

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=*')

        self.assertEqual(r.status_code, 202)

        # publish another object

        public1 = onames.pop()

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        # publish another object

        public2 = onames.pop()

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        # list shared and assert only the shared object is returned

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

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

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertEqual(sorted([shared1, shared2]), objects)

        # list detailed shared and assert only the shared object is returned

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

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

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

        l = sorted([shared1, shared2])

        i = 0

        for name in l:

            self.assertEqual(objects[i]['name'], name)

            self.assertEqual(objects[i]['bytes'],

                             len(self.objects[cname][name]))

            self.assertTrue('x_object_sharing' in objects[i])

            self.assertTrue('x_object_public' not in objects[i])

            i += 1

        # publish the shared object and assert it is still listed in the

        # shared objects

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

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

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

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

        i = 0

        for name in l:

            self.assertEqual(objects[i]['name'], name)

            self.assertEqual(objects[i]['bytes'],

                             len(self.objects[cname][name]))

            self.assertTrue('x_object_sharing' in objects[i])

            if name == shared1:

                self.assertTrue('x_object_public' in objects[i])

            else:

                self.assertTrue('x_object_public' not in objects[i])

            i += 1

        # create child object

        descendant = strnextling(shared1)

        self.upload_object(cname, descendant)

        # request shared and assert child obejct is not listed

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

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

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertTrue(shared1 in objects)

        self.assertTrue(descendant not in objects)

        # check folder inheritance

        folder, _ = self.create_folder(cname, HTTP_X_OBJECT_SHARING='read=*')

        # create child object

        descendant = '%s/%s' % (folder, get_random_name())

        self.upload_object(cname, descendant)

        # request shared

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

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

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertTrue(folder in objects)

        self.assertTrue(descendant in objects)

    def test_list_public(self):

        cname = self.cnames[0]

        onames = self.objects[cname].keys()

        # publish an object

        public1 = onames.pop()

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        # publish another

        public2 = onames.pop()

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        # share an object

        shared1 = onames.pop()

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')

        self.assertEqual(r.status_code, 202)

        # share another

        shared2 = onames.pop()

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')

        self.assertEqual(r.status_code, 202)

        # list public and assert only the public object is returned

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

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

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertEqual(sorted([public1, public2]), objects)

        # list detailed public and assert only the public object is returned

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

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

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

        l = sorted([public1, public2])

        i = 0

        for name in l:

            self.assertEqual(objects[i]['name'], name)

            self.assertEqual(objects[i]['bytes'],

                             len(self.objects[cname][name]))

            self.assertTrue('x_object_sharing' not in objects[i])

            self.assertTrue('x_object_public' in objects[i])

            i += 1

        # share the public object and assert it is still listed in the

        # public objects

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

        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')

        self.assertEqual(r.status_code, 202)

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

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

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

        i = 0

        for name in l:

            self.assertEqual(objects[i]['name'], name)

            self.assertEqual(objects[i]['bytes'],

                             len(self.objects[cname][name]))

            if name == public1:

                self.assertTrue('x_object_sharing' in objects[i])

            else:

                self.assertTrue('x_object_sharing' not in objects[i])

            i += 1

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

        # Assert listing the container public contents is forbidden to not

        # shared users

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

        self.assertEqual(r.status_code, 403)

        # Assert forbidden public object listing to shared users

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

        self.assertEqual(r.status_code, 403)

        # create child object

        descendant = strnextling(public1)

        self.upload_object(cname, descendant)

        # request public and assert child obejct is not listed

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

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

        if '' in objects:

            objects.remove('')

        self.assertEqual(r.status_code, 200)

        self.assertTrue(public1 in objects)

        self.assertTrue(descendant not in objects)

        # test folder inheritance

        folder, _ = self.create_folder(cname, HTTP_X_OBJECT_PUBLIC='true')

        # create child object

        descendant = '%s/%s' % (folder, get_random_name())

        self.upload_object(cname, descendant)

        # request public

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

        self.assertEqual(r.status_code, 200)

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

        self.assertTrue(folder in objects)

        self.assertTrue(descendant not in objects)

    def test_list_shared_public(self):

        cname = self.cnames[0]

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

        onames = self.objects[cname].keys()

        r = self.get('%s?shared=&public=&format=json' % container_url)

        self.assertEqual(r.status_code, 200)

        objects = json.loads(r.content)

        self.assertEqual(len(objects), 0)

        # publish an object

        public1 = onames.pop()

        url = join_urls(container_url, public1)

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        r = self.get('%s?shared=&public=&format=json' % container_url)

        self.assertEqual(r.status_code, 200)

        objects = json.loads(r.content)

        self.assertEqual(len(objects), 1)

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

        self.assertEqual(objects[0]['bytes'],

                         len(self.objects[cname][public1]))

        self.assertTrue('x_object_public' in objects[0])

        # publish another

        public2 = onames.pop()

        url = join_urls(container_url, public2)

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        # share an object

        shared1 = onames.pop()

        url = join_urls(container_url, shared1)

        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')

        self.assertEqual(r.status_code, 202)

        # share another

        shared2 = onames.pop()

        url = join_urls(container_url, shared2)

        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')

        self.assertEqual(r.status_code, 202)

        # list shared and public objects and assert object is listed

        r = self.get('%s?shared=&public=&format=json' % container_url)

        self.assertEqual(r.status_code, 200)

        objects = json.loads(r.content)

        l = sorted([public1, public2, shared1, shared2])

        i = 0

        for name in l:

            self.assertEqual(objects[i]['name'], name)

            self.assertEqual(objects[i]['bytes'],

                             len(self.objects[cname][name]))

            self.assertTrue('x_object_sharing' in objects[i] or

                            'x_object_public' in objects[i])

            i += 1

        # assert not listing shared and public to a not shared user

        r = self.get('%s?shared=&public=&format=json' % container_url,

                     user='bob')

        self.assertEqual(r.status_code, 403)

        # assert not listing public to a shared user

        r = self.get('%s?shared=&public=&format=json' % container_url,

                     user='alice')

        self.assertEqual(r.status_code, 403)

        # create child object

        descendant = strnextling(public1)

        self.upload_object(cname, descendant)

        # request public and assert child object is not listed

        r = self.get('%s?shared=&public=' % container_url)

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

        if '' in objects:

            objects.remove('')

        self.assertEqual(r.status_code, 200)

        self.assertEqual(objects, l)

        # test folder inheritance

        folder, _ = self.create_folder(cname, HTTP_X_OBJECT_PUBLIC='true')

        # create child object

        descendant = '%s/%s' % (folder, get_random_name())

        self.upload_object(cname, descendant)

        # request public

        r = self.get('%s?shared=&public=' % container_url)

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertTrue(folder in objects)

        self.assertTrue(descendant not in objects)

        # unpublish public1

        url = join_urls(container_url, public1)

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='false')

        self.assertEqual(r.status_code, 202)

        # unpublish public2

        url = join_urls(container_url, public2)

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='false')

        self.assertEqual(r.status_code, 202)

        # unpublish folder

        url = join_urls(container_url, folder)

        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='false')

        self.assertEqual(r.status_code, 202)

        r = self.get('%s?shared=&public=' % container_url)

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        l = sorted([shared1, shared2])

        self.assertEqual(objects, l)

    def test_list_objects(self):

        cname = self.cnames[0]

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

        r = self.get(url)

        self.assertTrue(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

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

    def test_list_objects_containing_slash(self):

        self.create_container('test')

        self.upload_object('test', quote('/objectname', ''))

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

        r = self.get(url)

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

        if '' in objects:

            objects.remove('')

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

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

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

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

        r = self.get('%s?format=xml' % url)

        try:

            objects = minidom.parseString(r.content)

        except:

            self.fail('xml format expected')

        self.assertEqual(

            [n.firstChild.data for n in objects.getElementsByTagName('name')],

            ['/objectname'])

    def test_list_objects_with_limit_marker(self):

        cname = self.cnames[0]

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

        r = self.get('%s?limit=qwert' % url)

        self.assertTrue(r.status_code != 500)

        r = self.get('%s?limit=2' % url)

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

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

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

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

                   'moms_birthday.jpg']

        limit = 4

        for m in markers:

            r = self.get('%s?limit=%s&marker=%s' % (url, limit, m))

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

            if '' in objects:

                objects.remove('')

            start = onames.index(m) + 1

            end = start + limit

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

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

    @pithos_test_settings(API_LIST_LIMIT=10)

    def test_list_limit_exceeds(self):

        self.create_container('container')

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

        for _ in range(pithos_settings.API_LIST_LIMIT + 1):

            self.upload_object('container')

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

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

        self.assertEqual(pithos_settings.API_LIST_LIMIT,

                         len(objects))

    def test_list_pseudo_hierarchical_folders(self):

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

        r = self.get('%s?prefix=photos&delimiter=/' % url)

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertEquals(

            ['photos/animals/', 'photos/me.jpg', 'photos/plants/'],

            objects)

        r = self.get('%s?prefix=photos/animals&delimiter=/' % url)

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

        if '' in objects:

            objects.remove('')

        self.assertEquals(

            ['photos/animals/cats/', 'photos/animals/dogs/'], objects)

        r = self.get('%s?path=photos' % url)

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

        if '' in objects:

            objects.remove('')

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

    def test_extended_list_json(self):

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

        params = {'format': 'json', 'limit': 2, 'prefix': 'photos/animals',

                  'delimiter': '/'}

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

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

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

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

    def test_extended_list_xml(self):

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

        params = {'format': 'xml', 'limit': 4, 'prefix': 'photos',

                  'delimiter': '/'}

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

        self.assertEqual(r.status_code, 200)

        try:

            xml = minidom.parseString(r.content)

        except:

            self.fail('xml format expected')

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

        # update object meta

        cname = 'apples'

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

        oname = self.objects[cname].keys().pop()

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

        headers = dict(('HTTP_X_OBJECT_META_%s' % k.upper(), v)

                       for k, v in meta.iteritems())

        object_url = join_urls(container_url, oname)

        self.post(object_url, content_type='', **headers)

        # list objects that satisfy the criteria

        r = self.get('%s?meta=Quality,Stock' % container_url)

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertEqual(objects, [oname])

    def test_list_using_meta(self):

        # update object meta

        cname = 'apples'

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

        onames = self.objects[cname].keys()

        url = join_urls(container_url, onames[0])

        r = self.post(url, content_type='', HTTP_X_OBJECT_META_QUALITY='aaa')

        self.assertEqual(r.status_code, 202)

        url = join_urls(container_url, onames[1])

        r = self.post(url, content_type='', HTTP_X_OBJECT_META_QUALITY='ab')

        self.assertEqual(r.status_code, 202)

        url = join_urls(container_url, onames[2])

        r = self.post(url, content_type='', HTTP_X_OBJECT_META_STOCK='100')

        self.assertEqual(r.status_code, 202)

        url = join_urls(container_url, onames[3])

        r = self.post(url, content_type='', HTTP_X_OBJECT_META_STOCK='200')

        self.assertEqual(r.status_code, 202)

        # test multiple existence criteria matches

        r = self.get('%s?meta=Quality,Stock' % container_url)

        self.assertEqual(r.status_code, 200)

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

        if '' in objects: