test.py 6.13 KB
Newer Older
1
# Copyright 2013 GRNET S.A. All rights reserved.
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
#
# 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.

34
from mock import patch, call
35
from unittest import TestCase
36

37
from kamaki.clients import astakos, ClientError
38
39
40


example = dict(
41
42
43
44
45
46
47
48
49
    serviceCatalog=[
        dict(name='service name 1', type='compute', endpoints=[
            dict(version_id='v1', publicUrl='http://1.1.1.1/v1'),
            dict(version_id='v2', publicUrl='http://1.1.1.1/v2')]),
        dict(name='service name 2', type='image', endpoints=[
            dict(version_id='v2', publicUrl='http://1.1.1.1/v2'),
            dict(version_id='v2.1', publicUrl='http://1.1.1.1/v2/xtra')])
        ],
    user=dict(
50
51
52
53
54
55
56
        name='Simple Name',
        username='User Full Name',
        auth_token_expires='1362583796000',
        auth_token_created='1359991796000',
        email=['user@example.gr'],
        id=42,
        uuid='aus3r-uu1d-f0r-73s71ng-as7ak0s')
57
    )
58
59

example0 = dict(
60
61
62
63
64
65
66
67
68
    serviceCatalog=[
        dict(name='service name 1', type='compute', endpoints=[
            dict(version_id='v1', publicUrl='http://1.1.1.1/v1'),
            dict(version_id='v2', publicUrl='http://1.1.1.1/v2')]),
        dict(name='service name 3', type='object-storage', endpoints=[
            dict(version_id='v2', publicUrl='http://1.1.1.1/v2'),
            dict(version_id='v2.1', publicUrl='http://1.1.1.1/v2/xtra')])
        ],
    user=dict(
69
70
        name='Simple Name 0',
        username='User Full Name 0',
71
72
        auth_token_expires='1362585796000',
        auth_token_created='1359931796000',
73
        email=['user0@example.gr'],
74
75
76
        id=42,
        uuid='aus3r-uu1d-507-73s71ng-as7ak0s')
    )
77
78


79
80
81
82
83
84
85
class FR(object):
    json = example
    headers = {}
    content = json
    status = None
    status_code = 200

86
astakos_pkg = 'kamaki.clients.astakos.AstakosClient'
87
88


89
class AstakosClient(TestCase):
90

91
92
    cached = False

93
94
95
96
97
98
99
100
    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]))

101
102
103
    def setUp(self):
        self.url = 'https://astakos.example.com'
        self.token = 'ast@k0sT0k3n=='
104
        self.client = astakos.AstakosClient(self.url, self.token)
105

106
107
108
    def tearDown(self):
        FR.json = example

109
110
    @patch('%s.get' % astakos_pkg, return_value=FR())
    def _authenticate(self, get):
111
        r = self.client.authenticate()
112
        self.assertEqual(get.mock_calls[-1], call('/tokens'))
113
114
        self.cached = True
        return r
115

116
117
    def test_authenticate(self):
        r = self._authenticate()
118
119
120
121
122
123
124
        self.assert_dicts_are_equal(r, example)

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

126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
    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['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['serviceCatalog'] if s['type'] == stype]
        expected = [e for e in expected[0]['endpoints'] if (
            e['version_id'] == version.lower())]
        self.assert_dicts_are_equal(
            self.client.get_service_endpoints(stype, version), expected[0])

    def test_user_info(self):
149
150
151
        if not self.cached:
            self._authenticate()
        self.assertTrue(set(
152
            example['user'].keys()).issubset(self.client.user_info().keys()))
153

154
    def test_item(self):
155
156
        if not self.cached:
            self._authenticate()
157
        for term, val in example['user'].items():
158
            self.assertEqual(self.client.term(term, self.token), val)
159
160
        self.assertTrue(
            example['user']['email'][0] in self.client.term('email'))
161

162
    def test_list_users(self):
163
164
        if not self.cached:
            self._authenticate
165
166
        FR.json = example0
        self._authenticate()
167
        r = self.client.list_users()
168
169
        self.assertTrue(len(r) == 1)
        self.assertEqual(r[0]['auth_token'], self.token)
170
171
172
173

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