Synnefo » ./kamaki

# Copyright 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 unittest import TestCase

from mock import patch, call, Mock

from kamaki.clients import ClientError

from kamaki.clients.pithos import PithosClient

from kamaki.clients.astakos import AstakosClient

from kamaki.clients.connection.kamakicon import KamakiHTTPConnection as C

user_id = 'ac0un7-1d-5tr1ng'

account_info = {

    'content-language': 'en-us',

    'content-type': 'text/html; charset=utf-8',

    'date': 'Wed, 06 Mar 2013 13:25:51 GMT',

    'last-modified': 'Mon, 04 Mar 2013 18:22:31 GMT',

    'server': 'gunicorn/0.14.5',

    'vary': 'Accept-Language',

    'x-account-bytes-used': '751615526',

    'x-account-container-count': 7,

    'x-account-policy-quota': 53687091200,

    'x-account-policy-versioning': 'auto'}

class Pithos(TestCase):

    class FR(object):

        """FR stands for Fake Response"""

        json = dict()

        headers = {}

        content = json

        status = None

        status_code = 204

        headers = dict()

        def release(self):

            pass

    files = []

    def assert_dicts_are_equal(self, d1, d2):

        for k, v in d1.items():

            self.assertTrue(k in d2)

            if isinstance(v, dict):

                self.assert_dicts_are_equal(v, d2[k])

            else:

                self.assertEqual(unicode(v), unicode(d2[k]))

    def setUp(self):

        self.url = 'https://www.example.com/pithos'

        self.token = 'p17h0570k3n'

        self.client = PithosClient(self.url, self.token)

        self.client.account = user_id

        self.client.container = 'c0nt@1n3r_i'

    def tearDown(self):

        self.FR.headers = dict()

    def test_get_account_info(self):

        self.FR.headers = account_info

        with patch.object(C, 'perform_request', return_value=self.FR()):

            r = self.client.get_account_info()

            self.assertEqual(self.client.http_client.url, self.url)

            self.assertEqual(self.client.http_client.path, '/%s' % user_id)

            self.assert_dicts_are_equal(r, account_info)

            PithosClient.set_param = Mock()

            untils = ['date 1', 'date 2', 'date 3']

            for unt in untils:

                r = self.client.get_account_info(until=unt)

                self.assert_dicts_are_equal(r, account_info)

            for i in range(len(untils)):

                self.assertEqual(

                    PithosClient.set_param.mock_calls[i],

                    call('until', untils[i], iff=untils[i]))

    def test_replace_account_meta(self):

        self.FR.status_code = 202

        metas = dict(k1='v1', k2='v2', k3='v3')

        PithosClient.set_header = Mock()

        with patch.object(C, 'perform_request', return_value=self.FR()):

            self.client.replace_account_meta(metas)

            self.assertEqual(self.client.http_client.url, self.url)

            self.assertEqual(self.client.http_client.path, '/%s' % user_id)

            prfx = 'X-Account-Meta-'

            expected = [call('%s%s' % (prfx, k), v) for k, v in metas.items()]

            self.assertEqual(PithosClient.set_header.mock_calls, expected)