Statistics
| Branch: | Tag: | Revision:

root / kamaki / clients / astakos / test.py @ 736072e2

History | View | Annotate | Download (9 kB)

1
# Copyright 2013 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.
33

    
34
from mock import patch, call
35
from logging import getLogger
36
from unittest import TestCase
37

    
38
from kamaki.clients import ClientError
39

    
40

    
41
example = dict(
42
    access=dict(
43
         token=dict(
44
            expires="2013-07-14T10:07:42.481134+00:00",
45
            id="ast@k0sT0k3n==",
46
            tenant=dict(
47
                id="42",
48
                name="Simple Name 0")
49
        ),
50
        serviceCatalog=[
51
            dict(name='service name 1', type='compute', endpoints=[
52
                dict(versionId='v1', publicUrl='http://1.1.1.1/v1'),
53
                dict(versionId='v2', publicUrl='http://1.1.1.1/v2')]),
54
            dict(name='service name 3', type='object-storage', endpoints=[
55
                dict(versionId='v2', publicUrl='http://1.1.1.1/v2'),
56
                dict(versionId='v2.1', publicUrl='http://1.1.1.1/v2/xtra')])
57
            ],
58
        user=dict(
59
            name='Simple Name 0',
60
            username='User Full Name 0',
61
            auth_token_expires='1362585796000',
62
            auth_token_created='1359931796000',
63
            email=['user0@example.gr'],
64
            id=42,
65
            uuid='aus3r-uu1d-507-73s71ng-as7ak0s')
66
        )
67
    )
68

    
69

    
70
class FR(object):
71
    json = example
72
    headers = {}
73
    content = json
74
    status = None
75
    status_code = 200
76

    
77
astakos_pkg = 'kamaki.clients.astakos'
78

    
79

    
80
class AstakosClient(TestCase):
81

    
82
    cached = False
83

    
84
    def assert_dicts_are_equal(self, d1, d2):
85
        for k, v in d1.items():
86
            self.assertTrue(k in d2)
87
            if isinstance(v, dict):
88
                self.assert_dicts_are_equal(v, d2[k])
89
            else:
90
                self.assertEqual(unicode(v), unicode(d2[k]))
91

    
92
    def setUp(self):
93
        self.url = 'https://astakos.example.com'
94
        self.token = 'ast@k0sT0k3n=='
95
        from kamaki.clients.astakos import AstakosClient as AC
96
        self.client = AC(self.url, self.token)
97

    
98
    def tearDown(self):
99
        FR.json = example
100

    
101
    @patch('%s.LoggedAstakosClient.__init__' % astakos_pkg, return_value=None)
102
    @patch(
103
        '%s.LoggedAstakosClient.get_endpoints' % astakos_pkg,
104
        return_value=example)
105
    def _authenticate(self, get_endpoints, sac):
106
        r = self.client.authenticate()
107
        self.assertEqual(
108
            sac.mock_calls[-1], call(self.token, self.url,
109
                logger=getLogger('astakosclient')))
110
        self.assertEqual(get_endpoints.mock_calls[-1], call())
111
        return r
112

    
113
    def test_authenticate(self):
114
        r = self._authenticate()
115
        self.assert_dicts_are_equal(r, example)
116
        uuid = example['access']['user']['id']
117
        self.assert_dicts_are_equal(self.client._uuids, {self.token: uuid})
118
        self.assert_dicts_are_equal(self.client._cache, {uuid: r})
119
        from astakosclient import AstakosClient as SAC
120
        self.assertTrue(isinstance(self.client._astakos[uuid], SAC))
121
        self.assert_dicts_are_equal(self.client._uuids2usernames, {uuid: {}})
122
        self.assert_dicts_are_equal(self.client._usernames2uuids, {uuid: {}})
123

    
124
    def test_get_client(self):
125
        if not self.cached:
126
            self._authenticate()
127
        from astakosclient import AstakosClient as SNFAC
128
        self.assertTrue(self.client.get_client(), SNFAC)
129

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

    
135
    def test_get_services(self):
136
        if not self.cached:
137
            self._authenticate()
138
        slist = self.client.get_services()
139
        self.assertEqual(slist, example['access']['serviceCatalog'])
140

    
141
    def test_get_service_details(self):
142
        if not self.cached:
143
            self._authenticate()
144
        stype = '#FAIL'
145
        self.assertRaises(ClientError, self.client.get_service_details, stype)
146
        stype = 'compute'
147
        expected = [s for s in example['access']['serviceCatalog'] if (
148
            s['type'] == stype)]
149
        self.assert_dicts_are_equal(
150
            self.client.get_service_details(stype), expected[0])
151

    
152
    def test_get_service_endpoints(self):
153
        if not self.cached:
154
            self._authenticate()
155
        stype, version = 'compute', 'V2'
156
        self.assertRaises(
157
            ClientError, self.client.get_service_endpoints, stype)
158
        expected = [s for s in example['access']['serviceCatalog'] if (
159
            s['type'] == stype)]
160
        expected = [e for e in expected[0]['endpoints'] if (
161
            e['versionId'] == version.lower())]
162
        self.assert_dicts_are_equal(
163
            self.client.get_service_endpoints(stype, version), expected[0])
164

    
165
    def test_user_info(self):
166
        if not self.cached:
167
            self._authenticate()
168
        self.assertTrue(set(example['access']['user'].keys()).issubset(
169
            self.client.user_info().keys()))
170

    
171
    def test_item(self):
172
        if not self.cached:
173
            self._authenticate()
174
        for term, val in example['access']['user'].items():
175
            self.assertEqual(self.client.term(term, self.token), val)
176
        self.assertTrue(
177
            example['access']['user']['email'][0] in self.client.term('email'))
178

    
179
    def test_list_users(self):
180
        if not self.cached:
181
            self._authenticate()
182
        FR.json = example
183
        self._authenticate()
184
        r = self.client.list_users()
185
        self.assertTrue(len(r) == 1)
186
        self.assertEqual(r[0]['auth_token'], self.token)
187

    
188
    @patch(
189
        '%s.LoggedAstakosClient.get_usernames' % astakos_pkg,
190
        return_value={42: 'username42', 43: 'username43'})
191
    def test_uuids2usernames(self, get_usernames):
192
        from astakosclient import AstakosClientException
193
        self.assertRaises(
194
            AstakosClientException, self.client.uuids2usernames, [42, 43])
195
        with patch(
196
                '%s.LoggedAstakosClient.__init__' % astakos_pkg,
197
                return_value=None) as sac:
198
            with patch(
199
                    '%s.LoggedAstakosClient.get_endpoints' % astakos_pkg,
200
                    return_value=example) as get_endpoints:
201
                r = self.client.uuids2usernames([42, 43])
202
                self.assert_dicts_are_equal(
203
                    r, {42: 'username42', 43: 'username43'})
204
                self.assertEqual(sac.mock_calls[-1], call(
205
                    self.token, self.url, logger=getLogger('astakosclient')))
206
                self.assertEqual(get_endpoints.mock_calls[-1], call())
207
                self.assertEqual(get_usernames.mock_calls[-1], call([42, 43]))
208

    
209
    @patch(
210
        '%s.LoggedAstakosClient.get_uuids' % astakos_pkg,
211
        return_value={'username42': 42, 'username43': 43})
212
    def test_usernames2uuids(self, get_uuids):
213
        from astakosclient import AstakosClientException
214
        self.assertRaises(
215
            AstakosClientException, self.client.usernames2uuids, ['u1', 'u2'])
216
        with patch(
217
                '%s.LoggedAstakosClient.__init__' % astakos_pkg,
218
                return_value=None) as sac:
219
            with patch(
220
                    '%s.LoggedAstakosClient.get_endpoints' % astakos_pkg,
221
                    return_value=example) as get_endpoints:
222
                r = self.client.usernames2uuids(['u1', 'u2'])
223
                self.assert_dicts_are_equal(
224
                    r, {'username42': 42, 'username43': 43})
225
                self.assertEqual(sac.mock_calls[-1], call(
226
                    self.token, self.url, logger=getLogger('astakosclient')))
227
                self.assertEqual(get_endpoints.mock_calls[-1], call())
228
                self.assertEqual(get_uuids.mock_calls[-1], call(['u1', 'u2']))
229

    
230

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