test.py

# 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 mock import patch, call
from logging import getLogger
from unittest import TestCase

from kamaki.clients import ClientError


example = dict(
    access=dict(
         token=dict(
            expires="2013-07-14T10:07:42.481134+00:00",
            id="ast@k0sT0k3n==",
            tenant=dict(
                id="42",
                name="Simple Name 0")
        ),
        serviceCatalog=[
            dict(name='service name 1', type='compute', endpoints=[
                dict(versionId='v1', publicUrl='http://1.1.1.1/v1'),
                dict(versionId='v2', publicUrl='http://1.1.1.1/v2')]),
            dict(name='service name 3', type='object-storage', endpoints=[
                dict(versionId='v2', publicUrl='http://1.1.1.1/v2'),
                dict(versionId='v2.1', publicUrl='http://1.1.1.1/v2/xtra')])
            ],
        user=dict(
            name='Simple Name 0',
            username='User Full Name 0',
            auth_token_expires='1362585796000',
            auth_token_created='1359931796000',
            email=['user0@example.gr'],
            id=42,
            uuid='aus3r-uu1d-507-73s71ng-as7ak0s')
        )
    )


class FR(object):
    json = example
    headers = {}
    content = json
    status = None
    status_code = 200

astakos_pkg = 'kamaki.clients.astakos'


class AstakosClient(TestCase):

    cached = False

    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://astakos.example.com'
        self.token = 'ast@k0sT0k3n=='
        from kamaki.clients.astakos import AstakosClient as AC
        self.client = AC(self.url, self.token)

    def tearDown(self):
        FR.json = example

    @patch(
        '%s.SynnefoAstakosClient.__init__' % astakos_pkg, return_value=None)
    @patch(
        '%s.SynnefoAstakosClient.get_endpoints' % astakos_pkg,
        return_value=example)
    def _authenticate(self, get_endpoints, sac):
        r = self.client.authenticate()
        self.assertEqual(
            sac.mock_calls[-1], call(self.token, self.url,
                logger=getLogger('astakosclient')))
        self.assertEqual(get_endpoints.mock_calls[-1], call())
        return r

    def test_authenticate(self):
        r = self._authenticate()
        self.assert_dicts_are_equal(r, example)
        uuid = example['access']['user']['id']
        self.assert_dicts_are_equal(self.client._uuids, {self.token: uuid})
        self.assert_dicts_are_equal(self.client._cache, {uuid: r})
        from astakosclient import AstakosClient as SAC
        self.assertTrue(isinstance(self.client._astakos[uuid], SAC))
        self.assert_dicts_are_equal(self.client._uuids2usernames, {uuid: {}})
        self.assert_dicts_are_equal(self.client._usernames2uuids, {uuid: {}})

    def test_get_client(self):
        if not self.cached:
            self._authenticate()
        from astakosclient import AstakosClient as SNFAC
        self.assertTrue(self.client.get_client(), SNFAC)

    def test_get_token(self):
        self._authenticate()
        uuid = self.client._uuids.values()[0]
        self.assertEqual(self.client.get_token(uuid), self.token)

    def test_get_services(self):
        if not self.cached:
            self._authenticate()
        slist = self.client.get_services()
        self.assertEqual(slist, example['access']['serviceCatalog'])

    def test_get_service_details(self):
        if not self.cached:
            self._authenticate()
        stype = '#FAIL'
        self.assertRaises(ClientError, self.client.get_service_details, stype)
        stype = 'compute'
        expected = [s for s in example['access']['serviceCatalog'] if (
            s['type'] == stype)]
        self.assert_dicts_are_equal(
            self.client.get_service_details(stype), expected[0])

    def test_get_service_endpoints(self):
        if not self.cached:
            self._authenticate()
        stype, version = 'compute', 'V2'
        self.assertRaises(
            ClientError, self.client.get_service_endpoints, stype)
        expected = [s for s in example['access']['serviceCatalog'] if (
            s['type'] == stype)]
        expected = [e for e in expected[0]['endpoints'] if (
            e['versionId'] == version.lower())]
        self.assert_dicts_are_equal(
            self.client.get_service_endpoints(stype, version), expected[0])

    def test_user_info(self):
        if not self.cached:
            self._authenticate()
        self.assertTrue(set(example['access']['user'].keys()).issubset(
            self.client.user_info().keys()))

    def test_item(self):
        if not self.cached:
            self._authenticate()
        for term, val in example['access']['user'].items():
            self.assertEqual(self.client.term(term, self.token), val)
        self.assertTrue(
            example['access']['user']['email'][0] in self.client.term('email'))

    def test_list_users(self):
        if not self.cached:
            self._authenticate()
        FR.json = example
        self._authenticate()
        r = self.client.list_users()
        self.assertTrue(len(r) == 1)
        self.assertEqual(r[0]['auth_token'], self.token)

    @patch(
        '%s.SynnefoAstakosClient.get_usernames' % astakos_pkg,
        return_value={42: 'username42', 43: 'username43'})
    def test_uuids2usernames(self, get_usernames):
        from astakosclient import AstakosClientException
        self.assertRaises(
            AstakosClientException, self.client.uuids2usernames, [42, 43])
        with patch(
                '%s.SynnefoAstakosClient.__init__' % astakos_pkg,
                return_value=None) as sac:
            with patch(
                    '%s.SynnefoAstakosClient.get_endpoints' % astakos_pkg,
                    return_value=example) as get_endpoints:
                r = self.client.uuids2usernames([42, 43])
                self.assert_dicts_are_equal(
                    r, {42: 'username42', 43: 'username43'})
                self.assertEqual(sac.mock_calls[-1], call(
                    self.token, self.url, logger=getLogger('astakosclient')))
                self.assertEqual(get_endpoints.mock_calls[-1], call())
                self.assertEqual(get_usernames.mock_calls[-1], call([42, 43]))

    @patch(
        '%s.SynnefoAstakosClient.get_uuids' % astakos_pkg,
        return_value={'username42': 42, 'username43': 43})
    def test_usernames2uuids(self, get_uuids):
        from astakosclient import AstakosClientException
        self.assertRaises(
            AstakosClientException, self.client.usernames2uuids, ['u1', 'u2'])
        with patch(
                '%s.SynnefoAstakosClient.__init__' % astakos_pkg,
                return_value=None) as sac:
            with patch(
                    '%s.SynnefoAstakosClient.get_endpoints' % astakos_pkg,
                    return_value=example) as get_endpoints:
                r = self.client.usernames2uuids(['u1', 'u2'])
                self.assert_dicts_are_equal(
                    r, {'username42': 42, 'username43': 43})
                self.assertEqual(sac.mock_calls[-1], call(
                    self.token, self.url, logger=getLogger('astakosclient')))
                self.assertEqual(get_endpoints.mock_calls[-1], call())
                self.assertEqual(get_uuids.mock_calls[-1], call(['u1', 'u2']))


if __name__ == '__main__':
    from sys import argv
    from kamaki.clients.test import runTestCase
    runTestCase(AstakosClient, 'AstakosClient', argv[1:])