Synnefo

        #'ENGINE': 'django.db.backends.sqlite3',

        #'NAME': '/tmp/synnefo_test_db.sqlite',

        'ENGINE': 'django.db.backends.postgresql_psycopg2',

        'NAME': 'pithos',

        'USER': 'postgres',

        'PORT': '5432',

serial = 0

            with patch('astakosclient.AstakosClient.get_quotas') as m2:

                m2.return_value = {

                    "system": {

                        "pithos.diskspace": {

                            "usage": 0,

                            "limit": 1073741824,

                            "pending": 0

                        }

                    }

                }

                with patch('astakosclient.AstakosClient.issue_one_commission') as m3:

                    serials = []

                    append = serials.append

                    def get_serial(*args, **kwargs):

                        global serial

                        serial += 1

                        append(serial)

                        return serial

                    m3.side_effect = get_serial

                    with patch('astakosclient.AstakosClient.resolve_commissions') as m4:

                        m4.return_value = {'accepted': serials,

                                           'rejected': [],

                                           'failed': []}

                        with patch('astakosclient.AstakosClient.get_usernames') as m5:

                            def get_usernames(*args, **kwargs):

                                uuids = args[-1]

                                return dict((uuid, uuid) for uuid in uuids)

                            m5.side_effect = get_usernames

                            yield

#!/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 urlparse import urlunsplit, urlsplit

from xml.dom import minidom

from snf_django.utils.testing import with_settings, astakos_user

from pithos.backends.random_word import get_random_word

from pithos.api import settings as pithos_settings

from django.test import TestCase

from django.utils.http import urlencode

from django.conf import settings

import django.utils.simplejson as json

import re

import random

import threading

import functools

pithos_test_settings = functools.partial(with_settings, pithos_settings)

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

o_names = ['kate.jpg',

           'kate_beckinsale.jpg',

           'How To Win Friends And Influence People.pdf',

           'moms_birthday.jpg',

           'poodle_strut.mov',

           'Disturbed - Down With The Sickness.mp3',

           'army_of_darkness.avi',

           'the_mad.avi',

           'photos/animals/dogs/poodle.jpg',

           'photos/animals/dogs/terrier.jpg',

           'photos/animals/cats/persian.jpg',

           'photos/animals/cats/siamese.jpg',

           'photos/plants/fern.jpg',

           'photos/plants/rose.jpg',

           'photos/me.jpg']

details = {'container': ('name', 'count', 'bytes', 'last_modified',

                         'x_container_policy'),

           'object': ('name', 'hash', 'bytes', 'content_type',

                      'content_encoding', 'last_modified',)}

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

class PithosAPITest(TestCase):

    #TODO unauthorized request

    def setUp(self):

        pithos_settings.BACKEND_DB_MODULE = 'pithos.backends.lib.sqlalchemy'

        pithos_settings.BACKEND_DB_CONNECTION = construct_db_connection()

        pithos_settings.BACKEND_POOL_SIZE = 1

        self.user = 'user'

    def tearDown(self):

        #delete additionally created metadata

        meta = self.get_account_meta()

        self.delete_account_meta(meta)

        #delete additionally created groups

        groups = self.get_account_groups()

        self.delete_account_groups(groups)

        self._clean_account()

    def head(self, url, user='user', *args, **kwargs):

        with astakos_user(user):

            response = self.client.head(url, *args, **kwargs)

        return response

    def get(self, url, user='user', *args, **kwargs):

        with astakos_user(user):

            response = self.client.get(url, *args, **kwargs)

        return response

    def delete(self, url, user='user', *args, **kwargs):

        with astakos_user(user):

            response = self.client.delete(url, *args, **kwargs)

        return response

    def post(self, url, user='user', *args, **kwargs):

        with astakos_user(user):

            kwargs.setdefault('content_type', 'application/octet-stream')

            response = self.client.post(url, *args, **kwargs)

        return response

    def put(self, url, user='user', *args, **kwargs):

        with astakos_user(user):

            kwargs.setdefault('content_type', 'application/octet-stream')

            response = self.client.put(url, *args, **kwargs)

        return response

    def _clean_account(self):

        for c in self.list_containers():

            self.delete_container_content(c['name'])

            self.delete_container(c['name'])

    def update_account_meta(self, meta):

        kwargs = dict(

            ('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())

        r = self.post('/v1/%s?update=' % self.user, **kwargs)

        self.assertEqual(r.status_code, 202)

        account_meta = self.get_account_meta()

        (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for

            k in meta.keys())

        (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for

            k, v in meta.items())

    def reset_account_meta(self, meta):

        kwargs = dict(

            ('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())

        r = self.post('/v1/%s' % self.user, **kwargs)

        self.assertEqual(r.status_code, 202)

        account_meta = self.get_account_meta()

        (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for

            k in meta.keys())

        (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for

            k, v in meta.items())

    def delete_account_meta(self, meta):

        transform = lambda k: 'HTTP_%s' % k.replace('-', '_').upper()

        kwargs = dict((transform(k), '') for k, v in meta.items())

        r = self.post('/v1/%s?update=' % self.user, **kwargs)

        self.assertEqual(r.status_code, 202)

        account_meta = self.get_account_meta()

        (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for

            k in meta.keys())

        return r

    def delete_account_groups(self, groups):

        r = self.post('/v1/%s?update=' % self.user, **groups)

        self.assertEqual(r.status_code, 202)

        return r

    def get_account_info(self, until=None):

        url = '/v1/%s' % self.user

        if until is not None:

            parts = list(urlsplit(url))

            parts[3] = urlencode({

                'until': until

            })

            url = urlunsplit(parts)

        r = self.head(url)

        self.assertEqual(r.status_code, 204)

        return r

    def get_account_meta(self, until=None):

        r = self.get_account_info(until=until)

        headers = dict(r._headers.values())

        map(headers.pop,

            [k for k in headers.keys()

                if not k.startswith('X-Account-Meta-')])

        return headers

    def get_account_groups(self, until=None):

        r = self.get_account_info(until=until)

        headers = dict(r._headers.values())

        map(headers.pop,

            [k for k in headers.keys()

                if not k.startswith('X-Account-Group-')])

        return headers

    def list_containers(self, format='json', headers={}, **params):

        url = '/v1/%s' % self.user

        parts = list(urlsplit(url))

        params['format'] = format

        parts[3] = urlencode(params)

        url = urlunsplit(parts)

        _headers = dict(('HTTP_%s' % k.upper(), str(v))

                        for k, v in headers.items())

        r = self.get(url, **_headers)

        if format is None:

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

            if '' in containers:

                containers.remove('')

            return containers

        elif format == 'json':

            try:

                containers = json.loads(r.content)

            except:

                self.fail('json format expected')

            return containers

        elif format == 'xml':

            return minidom.parseString(r.content)

    def delete_container_content(self, cname):

        r = self.delete('/v1/%s/%s?delimiter=/' % (self.user, cname))

        self.assertEqual(r.status_code, 204)

        return r

    def delete_container(self, cname):

        r = self.delete('/v1/%s/%s' % (self.user, cname))

        self.assertEqual(r.status_code, 204)

        return r

    def create_container(self, cname):

        r = self.put('/v1/%s/%s' % (self.user, cname), data='')

        self.assertTrue(r.status_code in (202, 201))

        return r

    def upload_object(self, cname, oname=None, **meta):

        oname = oname or get_random_word(8)

        data = get_random_word(length=random.randint(1, 1024))

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

                       for k, v in meta.iteritems())

        r = self.put('/v1/%s/%s/%s' % (

            self.user, cname, oname), data=data, **headers)

        self.assertEqual(r.status_code, 201)

        return oname, data, r

    def create_folder(self, cname, oname=get_random_word(8), **headers):

        r = self.put('/v1/%s/%s/%s' % (

            self.user, cname, oname), data='',

            content_type='application/directory',

            **headers)

        self.assertEqual(r.status_code, 201)

        return oname, r

    def list_objects(self, cname):

        r = self.get('/v1/%s/%s?format=json' % (self.user, cname))

        self.assertTrue(r.status_code in (200, 204))

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

        return objects

    def assert_status(self, status, codes):

        l = [elem for elem in return_codes]

        if isinstance(codes, list):

            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 details[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 details[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))

class AssertMappingInvariant(object):

    def __init__(self, callable, *args, **kwargs):

        self.callable = callable

        self.args = args

        self.kwargs = kwargs

    def __enter__(self):

        self.map = self.callable(*self.args, **self.kwargs)

        return self.map

    def __exit__(self, type, value, tb):

        map = self.callable(*self.args, **self.kwargs)

        for k, v in self.map.items():

            if is_date(v):

                continue

            assert(k in map), '%s not in map' % k

            assert v == map[k]

django_sqlalchemy_engines = {

    'django.db.backends.postgresql_psycopg2': 'postgresql+psycopg2',

    'django.db.backends.postgresql': 'postgresql',

    'django.db.backends.mysql': '',

    'django.db.backends.sqlite3': 'mssql',

    'django.db.backends.oracle': 'oracle'}

def construct_db_connection():

    """Convert the django default database to an sqlalchemy connection

       string"""

    db = settings.DATABASES['default']

    if db['ENGINE'] == 'django.db.backends.sqlite3':

        return 'sqlite://'

    else:

        d = dict(scheme=django_sqlalchemy_engines.get(db['ENGINE']),

                 user=db['USER'],

                 pwd=db['PASSWORD'],

                 host=db['HOST'].lower(),

                 port=int(db['PORT']) if db['PORT'] != '' else '',

                 name=db['NAME'])

        return '%(scheme)s://%(user)s:%(pwd)s@%(host)s:%(port)s/%(name)s' % d

def is_date(date):

    __D = r'(?P<day>\d{2})'

    __D2 = r'(?P<day>[ \d]\d)'

    __M = r'(?P<mon>\w{3})'

    __Y = r'(?P<year>\d{4})'

    __Y2 = r'(?P<year>\d{2})'

    __T = r'(?P<hour>\d{2}):(?P<min>\d{2}):(?P<sec>\d{2})'

    RFC1123_DATE = re.compile(r'^\w{3}, %s %s %s %s GMT$' % (

        __D, __M, __Y, __T))

    RFC850_DATE = re.compile(r'^\w{6,9}, %s-%s-%s %s GMT$' % (

        __D, __M, __Y2, __T))

    ASCTIME_DATE = re.compile(r'^\w{3} %s %s %s %s$' % (

        __M, __D2, __T, __Y))

    for regex in RFC1123_DATE, RFC850_DATE, ASCTIME_DATE:

        m = regex.match(date)

        if m is not None:

            return True

    return False

def strnextling(prefix):

    """Return the first unicode string

       greater than but not starting with given prefix.

       strnextling('hello') -> 'hellp'

    """

    if not prefix:

        ## all strings start with the null string,

        ## therefore we have to approximate strnextling('')

        ## with the last unicode character supported by python

        ## 0x10ffff for wide (32-bit unicode) python builds

        ## 0x00ffff for narrow (16-bit unicode) python builds

        ## We will not autodetect. 0xffff is safe enough.

        return unichr(0xffff)

    s = prefix[:-1]

    c = ord(prefix[-1])

    if c >= 0xffff:

        raise RuntimeError

    s += unichr(c + 1)

    return s

def test_concurrently(times=2):

    """

    Add this decorator to small pieces of code that you want to test

    concurrently to make sure they don't raise exceptions when run at the

    same time.  E.g., some Django views that do a SELECT and then a subsequent

    INSERT might fail when the INSERT assumes that the data has not changed

    since the SELECT.

    """

    def test_concurrently_decorator(test_func):

        def wrapper(*args, **kwargs):

            exceptions = []

            def call_test_func():

                try:

                    test_func(*args, **kwargs)

                except Exception, e:

                    exceptions.append(e)

                    raise

            threads = []

            for i in range(times):

                threads.append(threading.Thread())

            for t in threads:

                t.start()

            for t in threads:

                t.join()

            if exceptions:

                raise Exception(

                    ('test_concurrently intercepted %s',

                     'exceptions: %s') % (len(exceptions), exceptions))

        return wrapper

    return test_concurrently_decorator

#!/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, AssertMappingInvariant,\

    DATE_FORMATS

import time as _time

import datetime

class AccountHead(PithosAPITest):

    def test_get_account_meta(self):

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

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

        # set account meta

        self.update_account_meta({'foo': 'bar'})

        account_info = self.get_account_info()

        self.assertTrue('X-Account-Meta-Foo' in account_info)

        self.assertEqual(account_info['X-Account-Meta-Foo'], 'bar')

        # list containers

        containers = self.list_containers()

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

                         len(containers))

        usage = 0

        for c in containers:

            # list objects

            objects = self.list_objects(c['name'])

            self.assertEqual(c['count'], len(objects))

            csum = sum([o['bytes'] for o in objects])

            self.assertEqual(int(c['bytes']), csum)

            usage += int(c['bytes'])

        self.assertEqual(

            int(account_info['x-account-bytes-used']) + uploaded_bytes,

            usage)

    def test_get_account_meta_until(self):

        self.update_account_meta({'foo': 'bar'})

        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.update_account_meta({'quality': 'AAA'})

        account_info = self.get_account_info()

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

                                       DATE_FORMATS[-1])

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

        assert until < last_modified

        self.assertTrue('X-Account-Meta-Quality' in account_info)

        self.assertTrue('X-Account-Meta-Foo' in account_info)

        account_info = self.get_account_info(until=until)

        self.assertTrue('X-Account-Meta-Quality' not in account_info)

        self.assertTrue('X-Account-Meta-Foo' in account_info)

        self.assertTrue('X-Account-Until-Timestamp' in account_info)

        t = datetime.datetime.strptime(

            account_info['X-Account-Until-Timestamp'], DATE_FORMATS[2])

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

    def test_get_account_meta_until_invalid_date(self):

        self.update_account_meta({'quality': 'AAA'})

        meta = self.get_account_meta(until='-1')

        self.assertTrue('X-Account-Meta-Quality' in meta)

class AccountGet(PithosAPITest):

    def setUp(self):

        PithosAPITest.setUp(self)

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

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

    def test_list(self):

        #list containers: row format

        containers = self.list_containers(format=None)

        self.assertEquals(containers,

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

    def test_list_with_limit(self):

        containers = self.list_containers(format=None, limit=2)

        self.assertEquals(len(containers), 2)

        self.assertEquals(containers, ['apples', 'bananas'])

    def test_list_with_marker(self):

        containers = self.list_containers(format=None, limit=2,

                                          marker='bananas')

        self.assertEquals(containers, ['kiwis', 'oranges'])

        containers = self.list_containers(format=None, limit=2,

                                          marker='oranges')

        self.assertEquals(containers, ['pears'])

    def test_list_json_with_marker(self):

        containers = self.list_containers(format='json', limit=2,

                                          marker='bananas')

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

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

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

        containers = self.list_containers(format='json', limit=2,

                                          marker='oranges')

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

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

    def test_list_xml_with_marker(self):

        xml = self.list_containers(format='xml', limit=2, marker='bananas')

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

        nodes = xml.getElementsByTagName('name')

        self.assertTrue(len(nodes) <= 2)

        names = [n.childNodes[0].data for n in nodes]

        self.assertEqual(names, ['kiwis', 'oranges'])

        xml = self.list_containers(format='xml', limit=2, marker='oranges')

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

        nodes = xml.getElementsByTagName('name')

        self.assertTrue(len(nodes) <= 2)

        names = [n.childNodes[0].data for n in nodes]

        self.assertEqual(names, ['pears'])

    def test_if_modified_since(self):

        account_info = self.get_account_info()

        last_modified = account_info['Last-Modified']

        t1 = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])

        t1_formats = map(t1.strftime, DATE_FORMATS)

        # Check not modified

        for t in t1_formats:

            r = self.get('/v1/%s' % self.user, HTTP_IF_MODIFIED_SINCE=t)

            self.assertEqual(r.status_code, 304)

        # modify account: add container

        _time.sleep(1)

        self.create_container('c1')

        # Check modified

        for t in t1_formats:

            r = self.get('/v1/%s' % self.user, HTTP_IF_MODIFIED_SINCE=t)

            self.assertEqual(r.status_code, 200)

            self.assertEqual(

                r.content.split('\n')[:-1],

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

        account_info = self.get_account_info()

        last_modified = account_info['Last-Modified']

        t2 = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])

        t2_formats = map(t2.strftime, DATE_FORMATS)

        # modify account: update account meta

        _time.sleep(1)

        self.update_account_meta({'foo': 'bar'})

        # Check modified

        for t in t2_formats:

            r = self.get('/v1/%s' % self.user, HTTP_IF_MODIFIED_SINCE=t)

            self.assertEqual(r.status_code, 200)

            self.assertEqual(

                r.content.split('\n')[:-1],

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

    def test_if_modified_since_invalid_date(self):

        r = self.get('/v1/%s' % self.user, HTTP_IF_MODIFIED_SINCE='Monday')

        self.assertEqual(r.status_code, 200)

        self.assertEqual(

            r.content.split('\n')[:-1],

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

    def test_if_not_modified_since(self):

        account_info = self.get_account_info()

        last_modified = account_info['Last-Modified']

        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])

        # Check unmodified

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

        t1_formats = map(t1.strftime, DATE_FORMATS)

        for t in t1_formats:

            r = self.get('/v1/%s' % self.user, HTTP_IF_UNMODIFIED_SINCE=t)

            self.assertEqual(r.status_code, 200)

            self.assertEqual(

                r.content.split('\n')[:-1],

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

        # modify account: add container

        _time.sleep(2)

        self.create_container('c1')

        account_info = self.get_account_info()

        last_modified = account_info['Last-Modified']

        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])

        t2 = t - datetime.timedelta(seconds=1)

        t2_formats = map(t2.strftime, DATE_FORMATS)

        # Check modified

        for t in t2_formats:

            r = self.get('/v1/%s' % self.user, HTTP_IF_UNMODIFIED_SINCE=t)

            self.assertEqual(r.status_code, 412)

        # modify account: update account meta

        _time.sleep(1)

        self.update_account_meta({'foo': 'bar'})

        account_info = self.get_account_info()

        last_modified = account_info['Last-Modified']

        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])

        t3 = t - datetime.timedelta(seconds=1)

        t3_formats = map(t3.strftime, DATE_FORMATS)

        # Check modified

        for t in t3_formats:

            r = self.get('/v1/%s' % self.user, HTTP_IF_UNMODIFIED_SINCE=t)

            self.assertEqual(r.status_code, 412)

    def test_if_unmodified_since_invalid_date(self):

        r = self.get('/v1/%s' % self.user, HTTP_IF_UNMODIFIED_SINCE='Monday')

        self.assertEqual(r.status_code, 200)

        self.assertEqual(

            r.content.split('\n')[:-1],

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

class AccountPost(PithosAPITest):

    def setUp(self):

        PithosAPITest.setUp(self)

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

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

        # set account meta

        self.update_account_meta({'foo': 'bar'})

    def test_update_meta(self):

        with AssertMappingInvariant(self.get_account_groups):

            initial = self.get_account_meta()

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

            kwargs = dict(('HTTP_X_ACCOUNT_META_%s' % k, str(v))

                          for k, v in meta.items())

            r = self.post('/v1/%s?update=' % self.user, **kwargs)

            self.assertEqual(r.status_code, 202)

            meta.update(initial)

            account_meta = self.get_account_meta()

            (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for

                k in meta.keys())

            (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for

                k, v in meta.items())

    def test_reset_meta(self):

        with AssertMappingInvariant(self.get_account_groups):

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

            self.update_account_meta(meta)

            new_meta = {'test': 'test33'}

            kwargs = dict((

                'HTTP_X_ACCOUNT_META_%s' % k, str(v)

            ) for k, v in new_meta.items())

            r = self.post('/v1/%s' % self.user, **kwargs)

            self.assertEqual(r.status_code, 202)

            account_meta = self.get_account_meta()

            (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for

                k in new_meta.keys())

            (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for

                k, v in new_meta.items())

            (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for

                k in meta.keys())

    def test_delete_meta(self):

        with AssertMappingInvariant(self.get_account_groups):

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

            self.update_account_meta(meta)

            kwargs = dict(

                ('HTTP_X_ACCOUNT_META_%s' % k, '') for k, v in meta.items())

            r = self.post('/v1/%s?update=' % self.user, **kwargs)

            self.assertEqual(r.status_code, 202)

            account_meta = self.get_account_meta()

            (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for

                k in meta.keys())

    def test_set_account_groups(self):

        with AssertMappingInvariant(self.get_account_meta):

            pithosdevs = ['verigak', 'gtsouk', 'chazapis']

            r = self.post(

                '/v1/%s?update=' % self.user,

                HTTP_X_ACCOUNT_GROUP_PITHOSDEV=','.join(pithosdevs))

            self.assertEqual(r.status_code, 202)

            account_groups = self.get_account_groups()

            self.assertTrue(

                'X-Account-Group-Pithosdev' in self.get_account_groups())

            self.assertEqual(

                account_groups['X-Account-Group-Pithosdev'],

                ','.join(sorted(pithosdevs)))

            clientdevs = ['pkanavos', 'mvasilak']

            r = self.post(

                '/v1/%s?update=' % self.user,

                HTTP_X_ACCOUNT_GROUP_CLIENTSDEV=','.join(clientdevs))

            self.assertEqual(r.status_code, 202)

            account_groups = self.get_account_groups()

            self.assertTrue(

                'X-Account-Group-Pithosdev' in account_groups)

            self.assertTrue(

                'X-Account-Group-Clientsdev' in account_groups)

            self.assertEqual(

                account_groups['X-Account-Group-Pithosdev'],

                ','.join(sorted(pithosdevs)))

            self.assertEqual(

                account_groups['X-Account-Group-Clientsdev'],

                ','.join(sorted(clientdevs)))

            clientdevs = ['mvasilak']

            r = self.post(

                '/v1/%s?update=' % self.user,

                HTTP_X_ACCOUNT_GROUP_CLIENTSDEV=''.join(clientdevs))

            self.assertEqual(r.status_code, 202)

            account_groups = self.get_account_groups()

            self.assertTrue(

                'X-Account-Group-Pithosdev' in account_groups)

            self.assertTrue(

                'X-Account-Group-Clientsdev' in account_groups)

            self.assertEqual(

                account_groups['X-Account-Group-Pithosdev'],

                ','.join(sorted(pithosdevs)))

            self.assertEqual(

                account_groups['X-Account-Group-Clientsdev'],

                ','.join(sorted(clientdevs)))

    def test_reset_account_groups(self):

        with AssertMappingInvariant(self.get_account_meta):

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

                      'clientsdev': ['pkanavos', 'mvasilak']}

            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))

                           for k, v in groups.iteritems())

            r = self.post('/v1/%s?update=' % self.user, **headers)

            self.assertEqual(r.status_code, 202)

            groups = {'pithosdev': ['verigak',

                                    'gtsouk',

                                    'chazapis',

                                    'papagian']}

            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))

                           for k, v in groups.iteritems())

            account_meta = self.get_account_meta()

            headers.update(dict(('HTTP_%s' % k.upper().replace('-', '_'), v)

                           for k, v in account_meta.iteritems()))

            r = self.post('/v1/%s' % self.user, **headers)

            self.assertEqual(r.status_code, 202)

            account_groups = self.get_account_groups()

            self.assertTrue(

                'X-Account-Group-Pithosdev' in account_groups)

            self.assertTrue(

                'X-Account-Group-Clientsdev' not in account_groups)

            self.assertEqual(

                account_groups['X-Account-Group-Pithosdev'],

                ','.join(sorted(groups['pithosdev'])))

    def test_delete_account_groups(self):

        with AssertMappingInvariant(self.get_account_meta):

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

                      'clientsdev': ['pkanavos', 'mvasilak']}

            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))

                           for k, v in groups.iteritems())

            self.post('/v1/%s?update=' % self.user, **headers)

            kwargs = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, '')

                          for k, v in groups.items())

            r = self.post('/v1/%s?update=' % self.user, **kwargs)

            self.assertEqual(r.status_code, 202)

            account_groups = self.get_account_groups()

            self.assertTrue(

                'X-Account-Group-Pithosdev' not in account_groups)

            self.assertTrue(

                'X-Account-Group-Clientsdev' not in account_groups)

#!/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,\

    strnextling, pithos_settings, pithos_test_settings

from pithos.backends.random_word import get_random_word

import django.utils.simplejson as json

from django.http import urlencode

from xml.dom import minidom

from urllib import quote

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

        r = self.head('/v1/%s/apples' % self.user)

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

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

        # share an object

        cname = self.cnames[0]

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

        oname = onames.pop()

        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),

                      content_type='',

                      HTTP_X_OBJECT_SHARING='read=*')

        self.assertEqual(r.status_code, 202)

        # publish another object

        other = onames.pop()

        r = self.post('/v1/%s/%s/%s' % (self.user, cname, other),

                      content_type='',

                      HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        # list shared and assert only the shared object is returned

        r = self.get('/v1/%s/%s?shared=' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

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

        objects.remove('')

        self.assertEqual([oname], objects)

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

        r = self.get('/v1/%s/%s?shared=&format=json' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

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

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

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

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

        # shared objects

        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),

                      content_type='',

                      HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        r = self.get('/v1/%s/%s?shared=&format=json' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

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

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

        # TODO

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

        # create child object

        descendant = strnextling(oname)

        self.upload_object(cname, descendant)

        # request shared and assert child obejct is not listed

        r = self.get('/v1/%s/%s?shared=' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

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

        objects.remove('')

        self.assertTrue(oname in objects)

        self.assertTrue(descendant not in objects)

        # check folder inheritance

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

        # create child object

        descendant = '%s/%s' % (oname, get_random_word(8))

        self.upload_object(cname, descendant)

        # request shared

        r = self.get('/v1/%s/%s?shared=' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

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

        objects.remove('')

        self.assertTrue(oname in objects)

        self.assertTrue(descendant in objects)

    def test_list_public(self):

        # publish an object

        cname = self.cnames[0]

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

        oname = onames.pop()

        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),

                      content_type='',

                      HTTP_X_OBJECT_PUBLIC='true')

        self.assertEqual(r.status_code, 202)

        # share another

        other = onames.pop()

        r = self.post('/v1/%s/%s/%s' % (self.user, cname, other),

                      content_type='',

                      HTTP_X_OBJECT_SHARING='read=alice')

        self.assertEqual(r.status_code, 202)

        # list public and assert only the public object is returned

        r = self.get('/v1/%s/%s?public=' % (self.user, cname))

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

        self.assertEqual(r.status_code, 200)

        self.assertTrue(oname in r.content.split('\n'))

        (self.assertTrue(o not in objects) for o in o_names[1:])

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

        r = self.get('/v1/%s/%s?public=&format=json' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

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

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

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

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

        # public objects

        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),

                      content_type='',

                      HTTP_X_OBJECT_SHARING='read=alice')

        self.assertEqual(r.status_code, 202)

        r = self.get('/v1/%s/%s?public=&format=json' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

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

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

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

        # Assert listing the container public contents is forbidden to not

        # shared users

        r = self.get('/v1/%s/%s?public=&format=json' % (

            self.user, cname), user='bob')

        self.assertEqual(r.status_code, 403)

        # Assert listing the container public contents to shared users

        r = self.get('/v1/%s/%s?public=&format=json' % (

            self.user, cname), user='alice')

        self.assertEqual(r.status_code, 200)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

        # TODO

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

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

        # assert public is not returned though

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

        # create child object

        descendant = strnextling(oname)

        self.upload_object(cname, descendant)

        # request public and assert child obejct is not listed

        r = self.get('/v1/%s/%s?public=' % (self.user, cname))

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

        objects.remove('')

        self.assertEqual(r.status_code, 200)

        self.assertTrue(oname in objects)

        (self.assertTrue(o not in objects) for o in o_names[1:])

        # test folder inheritance

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

        # create child object

        descendant = '%s/%s' % (oname, get_random_word(8))

        self.upload_object(cname, descendant)

        # request public

        r = self.get('/v1/%s/%s?public=' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

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

        self.assertTrue(oname in objects)

        self.assertTrue(descendant not in objects)

#    def test_list_shared_public(self):

#        # publish an object

#        cname = self.cnames[0]

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

#        oname = onames.pop()

#        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),

#                      content_type='',

#                      HTTP_X_OBJECT_PUBLIC='true')

#        self.assertEqual(r.status_code, 202)

#

#        # share another

#        other = onames.pop()

#        r = self.post('/v1/%s/%s/%s' % (self.user, cname, other),

#                      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('/v1/%s/%s?shared=&public=&format=json' % (

#            self.user, cname))

#        self.assertEqual(r.status_code, 200)

#        objects = json.loads(r.content)

#        self.assertEqual([o['name'] for o in objects], sorted([oname, other]))

#        for o in objects:

#            if o['name'] == oname:

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

#            elif o['name'] == other:

#                self.assertTrue('x_object_sharing' in objects[1])

#

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

#        r = self.get('/v1/%s/%s?shared=&public=&format=json' % (

#            self.user, cname), user='bob')

#        self.assertEqual(r.status_code, 403)

#

#        # assert listing shared and public to a shared user

#        r = self.get('/v1/%s/%s?shared=&public=&format=json' % (

#            self.user, cname), user='alice')

#        self.assertEqual(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([oname, other]))

#

#        # create child object

#        descentant1 = strnextling(oname)

#        self.upload_object(cname, descendant1)

#        descentant2 = strnextling(other)

#        self.upload_object(cname, descendant2)

#        r = self.get('/v1/%s/%s?shared=&public=&format=json' % (

#            self.user, cname), user='alice')

#        self.assertEqual(r.status_code, 200)

#        try:

#            objects = json.loads(r.content)

#        except:

#            self.fail('json format expected')

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

#

#        # test inheritance

#        oname1, _ = self.create_folder(cname,

#                                       HTTP_X_OBJECT_SHARING='read=alice')

#        # create child object

#        descendant1 = '%s/%s' % (oname, get_random_word(8))

#        self.upload_object(cname, descendant1)

#

#        oname2, _ = self.create_folder(cname,

#                                       HTTP_X_OBJECT_PUBLIC='true')

#        # create child object

#        descendant2 = '%s/%s' % (oname, get_random_word(8))

#        self.upload_object(cname, descendant2)

#

#

#        o = self.upload_random_data(self.container[1], 'folder2/object')

#        objs = self.client.list_objects(self.container[1], shared=True, public=True)

#        self.assertEqual(objs, ['folder1', 'folder1/object', 'folder2'])

#        objs = cl.list_objects(

#            self.container[1], shared=True, public=True, account=get_user()

#        )

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

#

    def test_list_objects(self):

        cname = self.cnames[0]

        r = self.get('/v1/%s/%s' % (self.user, cname))

        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', '/objectname')

        r = self.get('/v1/%s/test' % self.user)

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

        if '' in objects:

            objects.remove('')

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

        r = self.get('/v1/%s/test?format=json' % self.user)

        try:

            objects = json.loads(r.content)

        except:

            self.fail('json format expected')

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

        r = self.get('/v1/%s/test?format=xml' % self.user)

        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]

        r = self.get('/v1/%s/%s?limit=qwert' % (self.user, cname))

        self.assertTrue(r.status_code != 500)

        r = self.get('/v1/%s/%s?limit=2' % (self.user, cname))

        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('/v1/%s/%s?limit=%s&marker=%s' % (

                self.user, cname, 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')

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

            self.upload_object('container')

        r = self.get('/v1/%s/container?format=json' % self.user)

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

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

        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(

            '/v1/%s/apples?prefix=photos/animals&delimiter=/' % self.user)

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

        if '' in objects:

            objects.remove('')

        self.assertEquals(

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

        r = self.get('/v1/%s/apples?path=photos' % self.user)

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

        if '' in objects:

            objects.remove('')

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

    def test_extended_list_json(self):

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

                  'delimiter': '/'}

        r = self.get('/v1/%s/apples?%s' % (self.user, 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):

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

                  'delimiter': '/'}

        r = self.get('/v1/%s/apples?%s' % (self.user, 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'

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

        self.post('/v1/%s/%s/%s' % (self.user, cname, oname),

                  content_type='', **headers)

        # list objects that satisfy the criteria

        r = self.get('/v1/%s/%s?meta=Quality,Stock' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertEqual(objects, [oname])

    def tearDown(self):

        pass

    def test_list_using_meta(self):

        # update object meta

        cname = 'apples'

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

        self.post('/v1/%s/%s/%s' % (self.user, cname, oname1),

                  content_type='', HTTP_X_OBJECT_META_QUALITY='aaa')

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

        self.post('/v1/%s/%s/%s' % (self.user, cname, oname2),

                  content_type='', HTTP_X_OBJECT_META_QUALITY='ab')

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

        self.post('/v1/%s/%s/%s' % (self.user, cname, oname3),

                  content_type='', HTTP_X_OBJECT_META_STOCK='100')

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

        self.post('/v1/%s/%s/%s' % (self.user, cname, oname4),

                  content_type='', HTTP_X_OBJECT_META_STOCK='200')

        # test multiple existence criteria matches

        r = self.get('/v1/%s/%s?meta=Quality,Stock' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

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

        if '' in objects:

            objects.remove('')

        self.assertTrue(objects, sorted([oname1, oname2, oname3, oname4]))

        # list objects that satisfy the existence criteria

        r = self.get('/v1/%s/%s?meta=Stock' % (self.user, cname))

        self.assertEqual(r.status_code, 200)

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

        if '' in objects: