Synnefo

#!/usr/bin/env python

#coding=utf8

# Copyright 2011-2012 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.tools.lib.client import Pithos_Client, Fault

from pithos.tools.lib.util import get_user, get_auth, get_url

from xml.dom import minidom

from StringIO import StringIO

from hashlib import new as newhasher

from binascii import hexlify

from httplib import HTTPConnection, HTTPSConnection

from urlparse import urlparse

import json

import unittest

import time as _time

import types

import hashlib

import mimetypes

import random

import datetime

import string

import re

DATE_FORMATS = ["%a %b %d %H:%M:%S %Y",

                "%A, %d-%b-%y %H:%M:%S GMT",

                "%a, %d %b %Y %H:%M:%S GMT"]

from pithos.api.settings import AUTHENTICATION_USERS

AUTHENTICATION_USERS = AUTHENTICATION_USERS or {}

OTHER_ACCOUNTS = AUTHENTICATION_USERS.copy()

OTHER_ACCOUNTS.pop(get_auth())

class BaseTestCase(unittest.TestCase):

    #TODO unauthorized request

    def setUp(self):

        self.client = Pithos_Client(get_url(), get_auth(), get_user())

        #keep track of initial containers

        self.initial_containers = self.client.list_containers()

        if self.initial_containers == '':

            self.initial_containers = []

        self._clean_account()

        self.invalid_client = Pithos_Client(get_url(), get_auth(), 'invalid')

        #keep track of initial account groups

        self.initial_groups = self.client.retrieve_account_groups()

        #keep track of initial account meta

        self.initial_meta = self.client.retrieve_account_metadata(restricted=True)

        self.extended = {

            'container':(

                'name',

                'count',

                'bytes',

                'last_modified',

                'x_container_policy'),

            'object':(

                'name',

                'hash',

                'bytes',

                'content_type',

                'content_encoding',

                'last_modified',)}

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

    def tearDown(self):

        #delete additionally created meta

        l = []

        for m in self.client.retrieve_account_metadata(restricted=True):

            if m not in self.initial_meta:

                l.append(m)

        self.client.delete_account_metadata(l)

        #delete additionally created groups

        l = []

        for g in self.client.retrieve_account_groups():

            if g not in self.initial_groups:

                l.append(g)

        self.client.unset_account_groups(l)

        self._clean_account()

    def _clean_account(self):

        for c in self.client.list_containers():

#             if c not in self.initial_containers:

                self.client.delete_container(c, delimiter='/')

                self.client.delete_container(c)

    def assert_status(self, status, codes):

        l = [elem for elem in self.return_codes]

        if type(codes) == types.ListType:

            l.extend(codes)

        else:

            l.append(codes)

        self.assertTrue(status in l)

    def assert_extended(self, data, format, type, size=10000):

        if format == 'xml':

            self._assert_xml(data, type, size)

        elif format == 'json':

            self._assert_json(data, type, size)

    def _assert_json(self, data, type, size):

        convert = lambda s: s.lower()

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

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

        for item in info:

            for i in data:

                if 'subdir' in i.keys():

                    continue

                self.assertTrue(item in i.keys())

    def _assert_xml(self, data, type, size):

        convert = lambda s: s.lower()

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

        try:

            info.remove('content_encoding')

        except ValueError:

            pass

        xml = data

        entities = xml.getElementsByTagName(type)

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

        for e in entities:

            for item in info:

                self.assertTrue(e.getElementsByTagName(item))

    def assert_raises_fault(self, status, callableObj, *args, **kwargs):

        """

        asserts that a Fault with a specific status is raised

        when callableObj is called with the specific arguments

        """

        try:

            r = callableObj(*args, **kwargs)

            self.fail('Should never reach here')

        except Fault, f:

            if type(status) == types.ListType:

                self.failUnless(f.status in status)

            else:

                self.failUnless(f.status == status)

    def assert_not_raises_fault(self, status, callableObj, *args, **kwargs):

        """

        asserts that a Fault with a specific status is not raised

        when callableObj is called with the specific arguments

        """

        try:

            r = callableObj(*args, **kwargs)

        except Fault, f:

            self.failIfEqual(f.status, status)

    def assert_container_exists(self, container):

        """

        asserts the existence of a container

        """

        try:

            self.client.retrieve_container_metadata(container)

        except Fault, f:

            self.failIf(f.status == 404)

    def assert_container_not_exists(self, container):

        """

        asserts there is no such a container

        """

        self.assert_raises_fault(404, self.client.retrieve_container_metadata,

                                 container)

    def assert_object_exists(self, container, object):

        """

        asserts the existence of an object

        """

        try:

            self.client.retrieve_object_metadata(container, object)

        except Fault, f:

            self.failIf(f.status == 404)

    def assert_object_not_exists(self, container, object):

        """

        asserts there is no such an object

        """

        self.assert_raises_fault(404, self.client.retrieve_object_metadata,

                                 container, object)

    def assert_versionlist_structure(self, versionlist):

        self.assertTrue(type(versionlist) == types.ListType)

        for elem in versionlist:

            self.assertTrue(type(elem) == types.ListType)

            self.assertEqual(len(elem), 2)

    def upload_random_data(self, container, name, length=1024, type=None,

                           enc=None, **meta):

        data = get_random_data(length)

        return self.upload_data(container, name, data, type, enc, **meta)

    def upload_data(self, container, name, data, type=None, enc=None, etag=None,

                    **meta):

        obj = {}

        obj['name'] = name

        try:

            obj['data'] = data

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

            args = {}

            args['etag'] = etag if etag else obj['hash']

            try:

                guess = mimetypes.guess_type(name)

                type = type if type else guess[0]

                enc = enc if enc else guess[1]

            except:

                pass

            args['content_type'] = type if type else 'plain/text'

            args['content_encoding'] = enc if enc else None

            obj['meta'] = args

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

            self.client.create_object(container, name, f=StringIO(obj['data']),

                                      meta=meta, **args)

            return obj

        except IOError:

            return

class AccountHead(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.containers = list(set(self.initial_containers + ['apples', 'bananas', 'kiwis', 'oranges', 'pears']))

        self.containers.sort()

        for item in self.containers:

            self.client.create_container(item)

        meta = {'foo':'bar'}

        self.client.update_account_metadata(**meta)

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

    def test_get_account_meta(self):

        meta = self.client.retrieve_account_metadata()

        containers = self.client.list_containers()

        l = str(len(containers))

        self.assertEqual(meta['x-account-container-count'], l)

        size1 = 0

        size2 = 0

        for c in containers:

            m = self.client.retrieve_container_metadata(c)

            csum = sum([o['bytes'] for o in self.client.list_objects(c, format='json')])

            self.assertEqual(int(m['x-container-bytes-used']), csum)

            size1 += int(m['x-container-bytes-used'])

            size2 += csum

        self.assertEqual(meta['x-account-bytes-used'], str(size1))

        self.assertEqual(meta['x-account-bytes-used'], str(size2))

    def test_get_account_403(self):

        self.assert_raises_fault(403,

                                 self.invalid_client.retrieve_account_metadata)

    def test_get_account_meta_until(self):

        t = datetime.datetime.utcnow()

        past = t - datetime.timedelta(minutes=15)

        past = int(_time.mktime(past.timetuple()))

        meta = {'premium':True}

        self.client.update_account_metadata(**meta)

        meta = self.client.retrieve_account_metadata(restricted=True,

                                                     until=past)

        self.assertTrue('premium' not in meta)

        meta = self.client.retrieve_account_metadata(restricted=True)

        self.assertTrue('premium' in meta)

    def test_get_account_meta_until_invalid_date(self):

        meta = {'premium':True}

        self.client.update_account_metadata(**meta)

        meta = self.client.retrieve_account_metadata(restricted=True,

                                                     until='kshfksfh')

        self.assertTrue('premium' in meta)

class AccountGet(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        #create some containers

        self.containers = list(set(self.initial_containers + ['apples', 'bananas', 'kiwis', 'oranges', 'pears']))

        self.containers.sort()

        for item in self.containers:

            self.client.create_container(item)

    def test_list(self):

        #list containers

        containers = self.client.list_containers()

        self.assertEquals(self.containers, containers)

    def test_list_403(self):

        self.assert_raises_fault(403, self.invalid_client.list_containers)

    def test_list_with_limit(self):

        limit = 2

        containers = self.client.list_containers(limit=limit)

        self.assertEquals(len(containers), limit)

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

    def test_list_with_marker(self):

        l = 2

        m = 'bananas'

        containers = self.client.list_containers(limit=l, marker=m)

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

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

        m = 'oranges'

        containers = self.client.list_containers(limit=l, marker=m)

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

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

    def test_list_json_with_marker(self):

        l = 2

        m = 'bananas'

        containers = self.client.list_containers(limit=l, marker=m, format='json')

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

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

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

    def test_list_xml_with_marker(self):

        l = 2

        m = 'oranges'

        xml = self.client.list_containers(limit=l, marker=m, format='xml')

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

        nodes = xml.getElementsByTagName('name')

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

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

        self.assertTrue('pears' in names or 'pears' > name for name in names)

    def test_if_modified_since(self):

        t = datetime.datetime.utcnow()

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

        #add a new container

        self.client.create_container('dummy')

        for f in DATE_FORMATS:

            past = t2.strftime(f)

            try:

                c = self.client.list_containers(if_modified_since=past)

                self.assertEqual(len(c), len(self.containers) + 1)

            except Fault, f:

                self.failIf(f.status == 304) #fail if not modified

    def test_if_modified_since_invalid_date(self):

        c = self.client.list_containers(if_modified_since='')

        self.assertEqual(len(c), len(self.containers))

    def test_if_not_modified_since(self):

        now = datetime.datetime.utcnow()

        since = now + datetime.timedelta(1)

        for f in DATE_FORMATS:

            args = {'if_modified_since':'%s' %since.strftime(f)}

            #assert not modified

            self.assert_raises_fault(304, self.client.list_containers, **args)

    def test_if_unmodified_since(self):

        now = datetime.datetime.utcnow()

        since = now + datetime.timedelta(1)

        for f in DATE_FORMATS:

            c = self.client.list_containers(if_unmodified_since=since.strftime(f))

            #assert success

            self.assertEqual(self.containers, c)

    def test_if_unmodified_since_precondition_failed(self):

        t = datetime.datetime.utcnow()

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

        #add a new container

        self.client.create_container('dummy')

        for f in DATE_FORMATS:

            past = t2.strftime(f)

            args = {'if_unmodified_since':'%s' %past}

            #assert precondition failed

            self.assert_raises_fault(412, self.client.list_containers, **args)

class AccountPost(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.containers = list(set(self.initial_containers + ['apples', 'bananas', 'kiwis', 'oranges', 'pears']))

        self.containers.sort()

        for item in self.containers:

            self.client.create_container(item)

        meta = {'foo':'bar'}

        self.client.update_account_metadata(**meta)

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

    def test_update_meta(self):

        with AssertMappingInvariant(self.client.retrieve_account_groups):

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

            self.client.update_account_metadata(**meta)

            meta.update(self.initial_meta)

            self.assertEqual(meta,

                             self.client.retrieve_account_metadata(

                                restricted=True))

    def test_invalid_account_update_meta(self):

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

        self.assert_raises_fault(403,

                                 self.invalid_client.update_account_metadata,

                                 **meta)

    def test_reset_meta(self):

        with AssertMappingInvariant(self.client.retrieve_account_groups):

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

            self.client.update_account_metadata(**meta)

            meta = {'test':'test33'}

            self.client.reset_account_metadata(**meta)

            self.assertEqual(meta, self.client.retrieve_account_metadata(restricted=True))

    def test_delete_meta(self):

        with AssertMappingInvariant(self.client.retrieve_account_groups):

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

            self.client.update_account_metadata(**meta)

            self.client.delete_account_metadata(meta.keys())

            account_meta = self.client.retrieve_account_metadata(restricted=True)

            for m in meta:

                self.assertTrue(m not in account_meta.keys())

    def test_set_account_groups(self):

        with AssertMappingInvariant(self.client.retrieve_account_metadata):

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

            self.client.set_account_groups(**groups)

            self.assertEqual(set(groups['pithosdev']),

                             set(self.client.retrieve_account_groups()['pithosdev']))

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

            self.client.set_account_groups(**more_groups)

            groups.update(more_groups)

            self.assertEqual(set(groups['clientsdev']),

                             set(self.client.retrieve_account_groups()['clientsdev']))

    def test_reset_account_groups(self):

        with AssertMappingInvariant(self.client.retrieve_account_metadata):

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

                      'clientsdev':'pkanavos,mvasilak'}

            self.client.set_account_groups(**groups)

            self.assertEqual(set(groups['pithosdev'].split(',')),

                             set(self.client.retrieve_account_groups()['pithosdev'].split(',')))

            self.assertEqual(set(groups['clientsdev'].split(',')),

                             set(self.client.retrieve_account_groups()['clientsdev'].split(',')))

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

            self.client.reset_account_groups(**groups)

            self.assertEqual(set(groups['pithosdev'].split(',')),

                             set(self.client.retrieve_account_groups()['pithosdev'].split(',')))

    def test_delete_account_groups(self):

        with AssertMappingInvariant(self.client.retrieve_account_metadata):

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

                      'clientsdev':'pkanavos,mvasilak'}

            self.client.set_account_groups(**groups)

            self.client.unset_account_groups(groups.keys())

            self.assertEqual({}, self.client.retrieve_account_groups())

class ContainerHead(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

        self.container = 'apples'

        self.client.create_container(self.container)

    def test_get_meta(self):

        meta = {'trash':'true'}

        t1 = datetime.datetime.utcnow()

        o = self.upload_random_data(self.container, o_names[0], **meta)

        if o:

            headers = self.client.retrieve_container_metadata(self.container)

            self.assertEqual(headers['x-container-object-count'], '1')

            self.assertEqual(headers['x-container-bytes-used'], str(len(o['data'])))

            t2 = datetime.datetime.strptime(headers['last-modified'], DATE_FORMATS[2])

            delta = (t2 - t1)

            threashold = datetime.timedelta(seconds=1)

            self.assertTrue(delta < threashold)

            self.assertTrue(headers['x-container-object-meta'])

            self.assertTrue('Trash' in headers['x-container-object-meta'])

class ContainerGet(BaseTestCase):

    def setUp(self):

        BaseTestCase.setUp(self)

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

        for c in self.container:

            self.client.create_container(c)

        self.obj = []

        for o in o_names[:8]:

            self.obj.append(self.upload_random_data(self.container[0], o))

        for o in o_names[8:]:

            self.obj.append(self.upload_random_data(self.container[1], o))

    def test_list_shared(self):

        self.client.share_object(self.container[0], self.obj[0]['name'], ('*',))

        objs = self.client.list_objects(self.container[0], shared=True)

        self.assertEqual(objs, [self.obj[0]['name']])

        # create child object

        self.upload_random_data(self.container[0], strnextling(self.obj[0]['name']))

        objs = self.client.list_objects(self.container[0], shared=True)

        self.assertEqual(objs, [self.obj[0]['name']])

        # test inheritance

        self.client.create_folder(self.container[1], 'folder')

        self.client.share_object(self.container[1], 'folder', ('*',))

        self.upload_random_data(self.container[1], 'folder/object')

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

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

    def test_list_public(self):

        self.client.publish_object(self.container[0], self.obj[0]['name'])

        objs = self.client.list_objects(self.container[0], public=True)

        self.assertEqual(objs, [self.obj[0]['name']])

        # create child object

        self.upload_random_data(self.container[0], strnextling(self.obj[0]['name']))

        objs = self.client.list_objects(self.container[0], public=True)

        self.assertEqual(objs, [self.obj[0]['name']])

        # test inheritance

        self.client.create_folder(self.container[1], 'folder')

        self.client.publish_object(self.container[1], 'folder')

        self.upload_random_data(self.container[1], 'folder/object')

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

        self.assertEqual(objs, ['folder'])

    def test_list_shared_public(self):

        self.client.share_object(self.container[0], self.obj[0]['name'], ('*',))

        self.client.publish_object(self.container[0], self.obj[1]['name'])

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

        self.assertEqual(objs, [self.obj[0]['name'], self.obj[1]['name']])

        # create child object

        self.upload_random_data(self.container[0], strnextling(self.obj[0]['name']))

        self.upload_random_data(self.container[0], strnextling(self.obj[1]['name']))

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

        self.assertEqual(objs, [self.obj[0]['name'], self.obj[1]['name']])

        # test inheritance

        self.client.create_folder(self.container[1], 'folder1')

        self.client.share_object(self.container[1], 'folder1', ('*',))

        self.upload_random_data(self.container[1], 'folder1/object')

        self.client.create_folder(self.container[1], 'folder2')

        self.client.publish_object(self.container[1], 'folder2')

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

    def test_list_objects(self):

        objects = self.client.list_objects(self.container[0])

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

        l.sort()

        self.assertEqual(objects, l)

    def test_list_objects_containing_slash(self):

        self.client.create_container('test')

        self.upload_random_data('test', '/objectname')

        objects = self.client.list_objects('test')

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

        objects = self.client.list_objects('test', format='json')

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

        objects = self.client.list_objects('test', format='xml')

        self.assert_extended(objects, 'xml', 'object')

        node_name = objects.getElementsByTagName('name')[0]

        self.assertEqual(node_name.firstChild.data, '/objectname')

    def test_list_objects_with_limit_marker(self):

        objects = self.client.list_objects(self.container[0], limit=2)

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

        l.sort()

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

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

                   'moms_birthday.jpg']

        limit = 4

        for m in markers:

            objects = self.client.list_objects(self.container[0], limit=limit,

                                               marker=m)

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

            l.sort()

            start = l.index(m) + 1

            end = start + limit

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

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

    #takes too long

    def _test_list_limit_exceeds(self):

        self.client.create_container('pithos')

        for i in range(10001):

            self.client.create_zero_length_object('pithos', i)

        self.assertEqual(10000, len(self.client.list_objects('pithos')))

    def test_list_empty_params(self):

        objects = self.client.get('/%s/%s' % (get_user(), self.container[0]))[2]

        if objects:

            objects = objects.strip().split('\n')

        self.assertEqual(objects,

                         self.client.list_objects(self.container[0]))

    def test_list_pseudo_hierarchical_folders(self):

        objects = self.client.list_objects(self.container[1], prefix='photos',

                                           delimiter='/')

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

                           'photos/plants/'], objects)

        objects = self.client.list_objects(self.container[1],

                                           prefix='photos/animals',

                                           delimiter='/')

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

        self.assertEquals(l, objects)

        objects = self.client.list_objects(self.container[1], path='photos')

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

    def test_extended_list_json(self):

        objects = self.client.list_objects(self.container[1], format='json',

                                           limit=2, prefix='photos/animals',

                                           delimiter='/')

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

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

    def test_extended_list_xml(self):

        xml = self.client.list_objects(self.container[1], format='xml', limit=4,

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

        self.assert_extended(xml, 'xml', 'object', size=4)

        dirs = xml.getElementsByTagName('subdir')

        self.assertEqual(len(dirs), 2)

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

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

        objects = xml.getElementsByTagName('name')

        self.assertEqual(len(objects), 1)

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

    def test_list_meta_double_matching(self):

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

        self.client.update_object_metadata(self.container[0],

                                           self.obj[0]['name'], **meta)

        obj = self.client.list_objects(self.container[0], meta='Quality,Stock')