Synnefo

# Copyright 2011 GRNET S.A. All rights reserved.

# 

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

# 

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

# 

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

# 

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

# 

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from django.test.client import Client

from django.test import TestCase

from django.utils import simplejson as json

from xml.dom import minidom

import types

import hashlib

import os

import mimetypes

import random

import datetime

import string

from pithos.backends import backend

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

class AaiClient(Client):

    def request(self, **request):

        request['HTTP_X_AUTH_TOKEN'] = '0000'

        return super(AaiClient, self).request(**request)

class BaseTestCase(TestCase):

    #TODO unauthorized request

    def setUp(self):

        self.client = AaiClient()

        self.headers = {

            'account':(

                'X-Account-Container-Count',

                'X-Account-Bytes-Used',

                'Last-Modified',

                'Content-Length',

                'Date',

                'Content-Type',),

            'container':(

                'X-Container-Object-Count',

                'X-Container-Bytes-Used',

                'Content-Type',

                'Last-Modified',

                'Content-Length',

                'Date',

                'X-Container-Block-Size',

                'X-Container-Block-Hash',

                'X-Container-Policy-Quota',

                'X-Container-Policy-Versioning',),

            'object':(

                'ETag',

                'Content-Length',

                'Content-Type',

                'Content-Encoding',

                'Last-Modified',

                'Date',

                'X-Object-Manifest',

                'Content-Range',

                'X-Object-Modified-By',

                'X-Object-Version',

                'X-Object-Version-Timestamp',)}

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

                             'json':'application/json',

                             '':'text/plain'}

        self.extended = {

            'container':(

                'name',

                'count',

                'bytes',

                'last_modified'),

            'object':(

                'name',

                'hash',

                'bytes',

                'content_type',

                'content_encoding',

                'last_modified',)}

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

    def assertFault(self, response, status_code, name):

        self.assertEqual(response.status_code, status_code)

    def assertBadRequest(self, response):

        self.assertFault(response, 400, 'badRequest')

    def assertItemNotFound(self, response):

        self.assertFault(response, 404, 'itemNotFound')

    def assertUnauthorized(self, response):

        self.assertFault(response, 401, 'unauthorized')

    def assertServiceUnavailable(self, response):

        self.assertFault(response, 503, 'serviceUnavailable')

    def assertNonEmpty(self, response):

        self.assertFault(response, 409, 'nonEmpty')

    def assert_status(self, response, codes):

        l = [elem for elem in self.return_codes]

        if type(codes) == types.ListType:

            l.extend(codes)

        else:

            l.append(codes)

        self.assertTrue(response.status_code in l)

    def get_account_meta(self, account, exp_meta={}):

        path = '/v1/%s' % account

        response = self.client.head(path)

        self.assert_status(response, 204)

        self.assert_headers(response, 'account', exp_meta)

        return response

    def list_containers(self, account, limit=10000, marker='', format='',

                        **headers):

        params = locals()

        params.pop('self')

        params.pop('account')

        path = '/v1/%s' % account

        response = self.client.get(path, params, **headers)

        self.assert_status(response, [200, 204, 304, 412])

        response.content = response.content.strip()

        if format:

            self.assert_extended(response, format, 'container', limit)

        else:

            names = get_content_splitted(response)

            self.assertTrue(len(names) <= limit)

        return response

    def update_account_meta(self, account, **metadata):

        path = '/v1/%s' % account

        response = self.client.post(path, **metadata)

        response.content = response.content

        self.assert_status(response, 202)

        return response

    def get_container_meta(self, account, container, exp_meta={}):

        params = locals()

        params.pop('self')

        params.pop('account')

        params.pop('container')

        path = '/v1/%s/%s' %(account, container)

        response = self.client.head(path, params)

        response.content = response.content

        self.assert_status(response, 204)

        if response.status_code == 204:

            self.assert_headers(response, 'container', exp_meta)

        return response

    def list_objects(self, account, container, limit=10000, marker='',

                     prefix='', format='', path='', delimiter='', meta='',

                     **headers):

        params = locals()

        params.pop('self')

        params.pop('account')

        params.pop('container')

        path = '/v1/%s/%s' % (account, container)

        response = self.client.get(path, params, **headers)

        response.content = response.content.strip()

        if format:

            self.assert_extended(response, format, 'object', limit)

        self.assert_status(response, [200, 204, 304, 412])

        return response

    def create_container(self, account, name, **meta):

        path = '/v1/%s/%s' %(account, name)

        response = self.client.put(path, **meta)

        response.content = response.content

        self.assert_status(response, [201, 202])

        return response

    def update_container_meta(self, account, name, **meta):

        path = '/v1/%s/%s' %(account, name)

        response = self.client.post(path,

                                    data=None,

                                    content_type='text/xml',

                                    follow=False, **meta)

        response.content = response.content

        self.assert_status(response, 202)

        return response

    def delete_container(self, account, container):

        path = '/v1/%s/%s' %(account, container)

        response = self.client.delete(path)

        response.content = response.content

        self.assert_status(response, [204, 409])

        return response

    def get_object_meta(self, account, container, name):

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

        response = self.client.head(path)

        response.content = response.content

        self.assert_status(response, 200)

        return response

    def get_object(self, account, container, name, format='', **headers):

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

        response = self.client.get(path, {'format':format}, **headers)

        response.content = response.content

        self.assert_status(response, [200, 206, 304, 412, 416])

        if response.status_code in [200, 206]:

            self.assert_headers(response, 'object')

        return response

    def upload_object(self, account, container, name, data, content_type='',

                      **headers):

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

        response = self.client.put(path, data, content_type, **headers)

        response.content = response.content

        self.assert_status(response, [201, 411, 422])

        if response.status_code == 201:

            self.assertTrue(response['Etag'])

        return response

    def copy_object(self, account, container, name, src, **headers):

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

        headers['HTTP_X_COPY_FROM'] = src

        response = self.client.put(path, **headers)

        response.content = response.content

        self.assert_status(response, 201)

        return response

    def move_object(self, account, container, name, src, **headers):

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

        headers['HTTP_X_MOVE_FROM'] = src

        response = self.client.put(path, **headers)

        response.content = response.content

        self.assert_status(response, 201)

        return response

    def update_object(self, account, container, name, data='',

                      content_type='', **headers):

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

        response = self.client.post(path, data, content_type, **headers)

        response.content = response.content

        self.assert_status(response, [202, 204, 416])

        return response

    def delete_object(self, account, container, name):

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

        response = self.client.delete(path)

        response.content = response.content

        self.assert_status(response, 204)

        return response

    def assert_headers(self, response, type, exp_meta={}):

        entities = ['Account', 'Container', 'Container-Object', 'Object']

        user_defined_meta = ['X-%s-Meta' %elem for elem in entities]

        headers = [item for item in response._headers.values()]

        t = tuple(user_defined_meta)

        system_headers = [h for h in headers if not h[0].startswith(t)]

        for h in system_headers:

            self.assertTrue(h[0] in self.headers[type])

            if exp_meta:

                self.assertEqual(h[1], exp_meta[h[0]])

    def assert_extended(self, response, format, type, size):

        exp_content_type = self.contentTypes[format]

        self.assertEqual(response['Content-Type'].find(exp_content_type), 0)

        if format == 'xml':

            self.assert_xml(response, type, size)

        elif format == 'json':

            self.assert_json(response, type, size)

    def assert_json(self, response, type, size):

        convert = lambda s: s.lower()

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

        data = json.loads(response.content)

        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, response, type, size):

        convert = lambda s: s.lower()

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

        try:

            info.remove('content_encoding')

        except ValueError:

            pass

        xml = minidom.parseString(response.content)

        for item in info:

            nodes = xml.getElementsByTagName(item)

            self.assertTrue(nodes)

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

    def upload_os_file(self, account, container, fullpath, meta={}):

        try:

            f = open(fullpath, 'r')

            data = f.read()

            name = os.path.split(fullpath)[-1]

            return self.upload_data(account, container, name, data)    

        except IOError:

            return

    def upload_random_data(self, account, container, name, length=1024,

                           meta={}):

        data = get_random_data(length)

        return self.upload_data(account, container, name, data, meta)

    def upload_data(self, account, container, name, data, meta={}):

        obj = {}

        obj['name'] = name

        try:

            obj['data'] = data

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

            meta.update({'HTTP_X_OBJECT_META_TEST':'test1',

                         'HTTP_ETAG':obj['hash']})

            type, enc = mimetypes.guess_type(name)

            meta['HTTP_CONTENT_TYPE'] = type and type or 'plain/text'

            if enc:

                meta['HTTP_CONTENT_ENCODING'] = enc

            obj['meta'] = meta

            r = self.upload_object(account,

                               container,

                               obj['name'],

                               obj['data'],

                               meta['HTTP_CONTENT_TYPE'],

                               **meta)

            if r.status_code == 201:

                return obj

        except IOError:

            return

class AccountHead(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.account = 'test'

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

        for item in self.containers:

            self.create_container(self.account, item)

    def tearDown(self):

        for c in  get_content_splitted(self.list_containers(self.account)):

            self.delete_container(self.account, c)

    def test_get_account_meta(self):

        response = self.get_account_meta(self.account)

        r2 = self.list_containers(self.account)

        containers =  get_content_splitted(r2)

        l = str(len(containers))

        self.assertEqual(response['X-Account-Container-Count'], l)

        size = 0

        for c in containers:

            r = self.get_container_meta(self.account, c)

            size = size + int(r['X-Container-Bytes-Used'])

        self.assertEqual(response['X-Account-Bytes-Used'], str(size))

    #def test_get_account_401(self):

    #    response = self.get_account_meta('non-existing-account')

    #    print response

    #    self.assertEqual(response.status_code, 401)

class AccountGet(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.account = 'test'

        #create some containers

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

        for item in self.containers:

            self.create_container(self.account, item)

    def tearDown(self):

        for c in get_content_splitted(self.list_containers(self.account)):

            response = self.delete_container(self.account, c)

    def test_list(self):

        #list containers

        response = self.list_containers(self.account)

        containers = get_content_splitted(response)

        self.assertEquals(self.containers, containers)

    #def test_list_204(self):

    #    response = self.list_containers('non-existing-account')

    #    self.assertEqual(response.status_code, 204)

    def test_list_with_limit(self):

        limit = 2

        response = self.list_containers(self.account, limit=limit)

        containers = get_content_splitted(response)

        self.assertEquals(len(containers), limit)

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

    def test_list_with_marker(self):

        l = 2

        m = 'bananas'

        response = self.list_containers(self.account, limit=l, marker=m)

        containers =  get_content_splitted(response)

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

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

        m = 'oranges'

        response = self.list_containers(self.account, limit=l, marker=m)

        containers = get_content_splitted(response)

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

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

    #def test_extended_list(self):

    #    self.list_containers(self.account, limit=3, format='xml')

    #    self.list_containers(self.account, limit=3, format='json')

    def test_list_json_with_marker(self):

        l = 2

        m = 'bananas'

        response = self.list_containers(self.account, limit=l, marker=m,

                                        format='json')

        containers = json.loads(response.content)

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

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

    def test_list_xml_with_marker(self):

        l = 2

        m = 'oranges'

        response = self.list_containers(self.account, limit=l, marker=m,

                                        format='xml')

        xml = minidom.parseString(response.content)

        nodes = xml.getElementsByTagName('name')

        self.assertEqual(len(nodes), 1)

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

    def test_if_modified_since(self):

        t = datetime.datetime.utcnow()

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

        #add a new container

        self.create_container(self.account,

                              'dummy')

        for f in DATE_FORMATS:

            past = t2.strftime(f)

            headers = {'HTTP_IF_MODIFIED_SINCE':'%s' %past}

            r = self.list_containers(self.account, **headers)

            #assert get success

            self.assertEqual(r.status_code, 200)

    def test_if_modified_since_invalid_date(self):

        headers = {'HTTP_IF_MODIFIED_SINCE':''}

        r = self.list_containers(self.account, **headers)

        #assert get success

        self.assertEqual(r.status_code, 200)

    def test_if_not_modified_since(self):

        now = datetime.datetime.utcnow()

        since = now + datetime.timedelta(1)

        for f in DATE_FORMATS:

            headers = {'HTTP_IF_MODIFIED_SINCE':'%s' %since.strftime(f)}

            r = self.list_containers(self.account, **headers)

            #assert not modified

            self.assertEqual(r.status_code, 304)

    def test_if_unmodified_since(self):

        now = datetime.datetime.utcnow()

        since = now + datetime.timedelta(1)

        for f in DATE_FORMATS:

            headers = {'HTTP_IF_UNMODIFIED_SINCE':'%s' %since.strftime(f)}

            r = self.list_containers(self.account, **headers)

            #assert success

            self.assertEqual(r.status_code, 200)

            self.assertEqual(self.containers, get_content_splitted(r))

    def test_if_unmodified_since_precondition_failed(self):

        t = datetime.datetime.utcnow()

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

        #add a new container

        self.create_container(self.account,

                              'dummy')

        for f in DATE_FORMATS:

            past = t2.strftime(f)

            headers = {'HTTP_IF_UNMODIFIED_SINCE':'%s' %past}

            r = self.list_containers(self.account, **headers)

            #assert get success

            self.assertEqual(r.status_code, 412)

class AccountPost(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.account = 'test'

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

        for item in self.containers:

            self.create_container(self.account, item)

    def tearDown(self):

        for c in  get_content_splitted(self.list_containers(self.account)):

            self.delete_container(self.account, c)

    def test_update_meta(self):

        meta = {'HTTP_X_ACCOUNT_META_TEST':'test',

                'HTTP_X_ACCOUNT_META_TOST':'tost'}

        response = self.update_account_meta(self.account, **meta)

        response = self.get_account_meta(self.account)

        for k,v in meta.items():

            key = '-'.join(elem.capitalize() for elem in k.split('_')[1:])

            self.assertTrue(response[key])

            self.assertEqual(response[key], v)

    #def test_invalid_account_update_meta(self):

    #    with AssertInvariant(self.get_account_meta, self.account):

    #        meta = {'HTTP_X_ACCOUNT_META_TEST':'test',

    #               'HTTP_X_ACCOUNT_META_TOST':'tost'}

    #        response = self.update_account_meta('non-existing-account', **meta)

class ContainerHead(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.account = 'test'

        self.container = 'apples'

        self.create_container(self.account, self.container)

    def tearDown(self):

        for o in self.list_objects(self.account, self.container):

            self.delete_object(self.account, self.container, o)

        self.delete_container(self.account, self.container)

    def test_get_meta(self):

        headers = {'HTTP_X_OBJECT_META_TRASH':'true'}

        t1 = datetime.datetime.utcnow()

        o = self.upload_random_data(self.account,

                                self.container,

                                'McIntosh.jpg',

                                meta=headers)

        if o:

            r = self.get_container_meta(self.account,

                                        self.container)

            self.assertEqual(r['X-Container-Object-Count'], '1')

            self.assertEqual(r['X-Container-Bytes-Used'], str(len(o['data'])))

            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('Trash' in r['X-Container-Object-Meta'])

class ContainerGet(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.account = 'test'

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

        for c in self.container:

            self.create_container(self.account, c)

        self.obj = []

        for o in o_names[:8]:

            self.obj.append(self.upload_random_data(self.account,

                                                    self.container[0],

                                                    o))

        for o in o_names[8:]:

            self.obj.append(self.upload_random_data(self.account,

                                                    self.container[1],

                                                    o))

    def tearDown(self):

        for c in self.container:

            for obj in get_content_splitted(self.list_objects(self.account, c)):

                self.delete_object(self.account, c, obj)

            self.delete_container(self.account, c)

    def test_list_objects(self):

        response = self.list_objects(self.account, self.container[0])

        objects = get_content_splitted(response)

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

        l.sort()

        self.assertEqual(objects, l)

    def test_list_objects_with_limit_marker(self):

        response = self.list_objects(self.account, self.container[0], limit=2)

        objects = get_content_splitted(response)

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

        l.sort()

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

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

                   'moms_birthday.jpg']

        limit = 4

        for m in markers:

            response = self.list_objects(self.account, self.container[0],

                                         limit=limit, marker=m)

            objects = get_content_splitted(response)

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

            l.sort()

            start = l.index(m) + 1

            end = start + limit

            end = len(l) >= end and end or len(l)

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

    def test_list_pseudo_hierarchical_folders(self):

        response = self.list_objects(self.account, self.container[1],

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

        objects = get_content_splitted(response)

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

                           'photos/plants/'], objects)

        response = self.list_objects(self.account, self.container[1],

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

        objs = get_content_splitted(response)

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

        self.assertEquals(l, objs)

        response = self.list_objects(self.account, self.container[1],

                                     path='photos')

        objects = get_content_splitted(response)

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

    def test_extended_list_json(self):

        response = self.list_objects(self.account,

                                     self.container[1],

                                     format='json', limit=2,

                                     prefix='photos/animals',

                                     delimiter='/')

        objects = json.loads(response.content)

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

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

    def test_extended_list_xml(self):

        response = self.list_objects(self.account, self.container[1],

                                     format='xml', limit=4, prefix='photos',

                                     delimiter='/')

        xml = minidom.parseString(response.content)

        dirs = xml.getElementsByTagName('subdir')

        self.assertEqual(len(dirs), 2)

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

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

        objects = xml.getElementsByTagName('name')

        self.assertEqual(len(objects), 1)

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

    def test_list_meta_double_matching(self):

        meta = {'HTTP_X_OBJECT_META_QUALITY':'aaa',

                'HTTP_X_OBJECT_META_STOCK':'true'}

        r = self.update_object(self.account,

                                    self.container[0],

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

                                    **meta)

        r = self.list_objects(self.account,

                          self.container[0],

                          meta='Quality,Stock')

        self.assertEqual(r.status_code, 200)

        obj = get_content_splitted(r)

        self.assertEqual(len(obj), 1)

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

    def test_list_using_meta(self):

        meta = {'HTTP_X_OBJECT_META_QUALITY':'aaa'}

        for o in self.obj[:2]:

            r = self.update_object(self.account,

                                    self.container[0],

                                    o['name'],

                                    **meta)

        meta = {'HTTP_X_OBJECT_META_STOCK':'true'}

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

            r = self.update_object(self.account,

                                    self.container[0],

                                    o['name'],

                                    **meta)

        r = self.list_objects(self.account,

                          self.container[0],

                          meta='Quality')

        self.assertEqual(r.status_code, 200)

        obj = get_content_splitted(r)

        self.assertEqual(len(obj), 2)

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

        # test case insensitive

        r = self.list_objects(self.account,

                          self.container[0],

                          meta='quality')

        self.assertEqual(r.status_code, 200)

        obj = get_content_splitted(r)

        self.assertEqual(len(obj), 2)

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

        # test multiple matches

        r = self.list_objects(self.account,

                          self.container[0],

                          meta='Quality,Stock')

        self.assertEqual(r.status_code, 200)

        obj = get_content_splitted(r)

        self.assertEqual(len(obj), 4)

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