← Back to branch summary

~ubuntu-branches/ubuntu/saucy/keystone/saucy-proposed

« back to all changes in this revision

Viewing changes to keystone/test/unit/test_services.py

  • Committer: Bazaar Package Importer
  • Author(s): Chuck Short
  • Date: 2011-08-23 10:18:22 UTC
  • Revision ID: james.westby@ubuntu.com-20110823101822-enve6zceb3lqhuvj
Tags: upstream-1.0~d4~20110823.1078
Import upstream version 1.0~d4~20110823.1078

Show diffs side-by-side

added added

removed removed

Lines of Context:
keystone/test/unit/test_services.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
# Copyright (c) 2010-2011 OpenStack, LLC.
 
3
#
 
4
# Licensed under the Apache License, Version 2.0 (the "License");
 
5
# you may not use this file except in compliance with the License.
 
6
# You may obtain a copy of the License at
 
7
#
 
8
#    http://www.apache.org/licenses/LICENSE-2.0
 
9
#
 
10
# Unless required by applicable law or agreed to in writing, software
 
11
# distributed under the License is distributed on an "AS IS" BASIS,
 
12
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 
13
# implied.
 
14
# See the License for the specific language governing permissions and
 
15
# limitations under the License.
 
16
 
 
17
 
 
18
import httplib2
 
19
import json
 
20
from lxml import etree
 
21
import os
 
22
import sys
 
23
sys.path.append(os.path.abspath(os.path.join(os.path.abspath(__file__),
 
24
                                '..', '..', '..', '..', '..', 'keystone')))
 
25
import unittest2 as unittest
 
26
 
 
27
import test_common as utils
 
28
from test_common import URL_V2
 
29
 
 
30
from keystone.logic.types import fault
 
31
 
 
32
 
 
33
class ServicesTest(unittest.TestCase):
 
34
    def setUp(self):
 
35
        self.auth_token = utils.get_auth_token()
 
36
        self.service_token = utils.get_service_token()
 
37
        self.missing_token = utils.get_none_token()
 
38
        self.invalid_token = utils.get_non_existing_token()
 
39
        self.disabled_token = utils.get_disabled_token()
 
40
        self.exp_auth_token = utils.get_exp_auth_token()
 
41
        self.user = utils.get_user()
 
42
        self.tenant = utils.get_tenant()
 
43
        utils.create_tenant(self.tenant, str(self.auth_token))
 
44
        utils.create_user(self.tenant, self.user, self.auth_token)
 
45
        self.sample_service = 'sampleservice'
 
46
        self.test_service = 'test_service'
 
47
        self.token = utils.get_token(self.user, 'secrete', self.tenant,
 
48
                                     'token')
 
49
        utils.create_service(self.sample_service, str(self.auth_token))
 
50
 
 
51
    def tearDown(self):
 
52
        utils.delete_user(self.user, self.auth_token)
 
53
        utils.delete_tenant(self.tenant, self.auth_token)
 
54
        utils.delete_service(self.sample_service, str(self.auth_token))
 
55
 
 
56
 
 
57
class GetServicesTest(ServicesTest):
 
58
        def test_get_services_using_keystone_admin_token_json(self):
 
59
            resp, content = utils.get_services(self.auth_token)
 
60
            self.assertEqual(200, int(resp['status']))
 
61
            #verify content
 
62
            obj = json.loads(content)
 
63
            if not "services" in obj:
 
64
                raise self.fail("Expecting Services")
 
65
            services = obj["services"]["values"]
 
66
            if len(services) < 1:
 
67
                self.fail("Services not of required length.")
 
68
            is_service_found = None
 
69
            for service in services:
 
70
                if service["id"] in [self.sample_service]:
 
71
                    is_service_found = True
 
72
            if not is_service_found:
 
73
                raise self.fail("Service not found")
 
74
 
 
75
        def test_get_services_using_keystone_admin_token_xml(self):
 
76
            resp, content = utils.get_services_xml(self.auth_token)
 
77
            self.assertEqual(200, int(resp['status']))
 
78
            # verify content
 
79
            # Validate Returned Content
 
80
            dom = etree.Element("root")
 
81
            dom.append(etree.fromstring(content))
 
82
            services = dom.find(
 
83
                "{http://docs.openstack.org/identity/api/v2.0}" \
 
84
                "services")
 
85
            if services == None:
 
86
                self.fail("Expecting Services")
 
87
            services = services.findall(
 
88
                "{http://docs.openstack.org/identity/api/v2.0}" \
 
89
                "service")
 
90
            if len(services) < 1:
 
91
                self.fail("Not the expected Service count")
 
92
            for service in services:
 
93
                if service.get("id") in [self.sample_service]:
 
94
                    is_service_found = True
 
95
            if not is_service_found:
 
96
                raise self.fail("Service not found")
 
97
 
 
98
        def test_get_services_using_service_admin_token(self):
 
99
            resp, content = utils.get_services(self.service_token)
 
100
            self.assertEqual(200, int(resp['status']))
 
101
            #verify content
 
102
            obj = json.loads(content)
 
103
            if not "services" in obj:
 
104
                raise self.fail("Expecting Services")
 
105
            services = obj["services"]["values"]
 
106
            if len(services) < 1:
 
107
                self.fail("Services not of required length.")
 
108
            is_service_found = None
 
109
            for service in services:
 
110
                if service["id"] in [self.sample_service]:
 
111
                    is_service_found = True
 
112
            if not is_service_found:
 
113
                raise self.fail("Service not found")
 
114
 
 
115
        def test_get_services_using_service_admin_token_xml(self):
 
116
            resp, content = utils.get_services_xml(self.service_token)
 
117
            self.assertEqual(200, int(resp['status']))
 
118
            # Verify content
 
119
            # Validate Returned Content
 
120
            dom = etree.Element("root")
 
121
            dom.append(etree.fromstring(content))
 
122
            services = dom.find(
 
123
                "{http://docs.openstack.org/identity/api/v2.0}" \
 
124
                "services")
 
125
            if services == None:
 
126
                self.fail("Expecting Services")
 
127
            services = services.findall(
 
128
                "{http://docs.openstack.org/identity/api/v2.0}" \
 
129
                "service")
 
130
            if len(services) < 1:
 
131
                self.fail("Not the expected Service count")
 
132
            for service in services:
 
133
                if service.get("id") in [self.sample_service]:
 
134
                    is_service_found = True
 
135
            if not is_service_found:
 
136
                raise self.fail("Service not found")
 
137
 
 
138
        def test_get_services_using_invalid_tokens(self):
 
139
            resp, content = utils.get_services(self.disabled_token)
 
140
            self.assertEqual(403, int(resp['status']))
 
141
            resp, content = utils.get_services(self.missing_token)
 
142
            self.assertEqual(401, int(resp['status']))
 
143
            resp, content = utils.get_services(self.exp_auth_token)
 
144
            self.assertEqual(403, int(resp['status']))
 
145
            resp, content = utils.get_services(self.invalid_token)
 
146
            self.assertEqual(404, int(resp['status']))
 
147
 
 
148
 
 
149
class GetServiceTest(ServicesTest):
 
150
    def test_service_get_json(self):
 
151
        resp, _content = utils.get_service(
 
152
            self.sample_service, str(self.auth_token))
 
153
        resp_val = int(resp['status'])
 
154
        self.assertEqual(200, resp_val)
 
155
 
 
156
    def test_service_get_xml(self):
 
157
        resp, _content = utils.get_service_xml(
 
158
            self.sample_service, str(self.auth_token))
 
159
        resp_val = int(resp['status'])
 
160
        self.assertEqual(200, resp_val)
 
161
 
 
162
    def test_service_get_using_expired_token(self):
 
163
        resp, _content = utils.get_service(
 
164
            self.sample_service, str(self.exp_auth_token))
 
165
        resp_val = int(resp['status'])
 
166
        self.assertEqual(403, resp_val)
 
167
 
 
168
    def test_service_get_using_disabled_token(self):
 
169
        resp, _content = utils.get_service(
 
170
            self.sample_service, str(self.disabled_token))
 
171
        resp_val = int(resp['status'])
 
172
        self.assertEqual(403, resp_val)
 
173
 
 
174
    def test_service_get_json_using_missing_token(self):
 
175
        resp, _content = utils.get_service(
 
176
            self.sample_service, str(self.missing_token))
 
177
        resp_val = int(resp['status'])
 
178
        self.assertEqual(401, resp_val)
 
179
 
 
180
    def test_service_get_json_using_invalid_token(self):
 
181
        resp, _content = utils.get_service(
 
182
            self.sample_service, str(self.invalid_token))
 
183
        resp_val = int(resp['status'])
 
184
        self.assertEqual(404, resp_val)
 
185
 
 
186
    def test_get_service_using_invalid_tokens(self):
 
187
        resp, content = utils.get_service(
 
188
            self.sample_service, self.disabled_token)
 
189
        self.assertEqual(403, int(resp['status']))
 
190
        resp, content = utils.get_service(
 
191
            self.sample_service, self.missing_token)
 
192
        self.assertEqual(401, int(resp['status']))
 
193
        resp, content = utils.get_service(
 
194
            self.sample_service, self.exp_auth_token)
 
195
        self.assertEqual(403, int(resp['status']))
 
196
        resp, content = utils.get_service(
 
197
            self.sample_service, self.invalid_token)
 
198
        self.assertEqual(404, int(resp['status']))
 
199
 
 
200
    def test_get_service_using_invalid_tokens(self):
 
201
        resp, content = utils.get_service(
 
202
            self.sample_service, self.disabled_token)
 
203
        self.assertEqual(403, int(resp['status']))
 
204
 
 
205
        resp, content = utils.get_service(
 
206
            self.sample_service, self.missing_token)
 
207
        self.assertEqual(401, int(resp['status']))
 
208
 
 
209
        resp, content = utils.get_service(
 
210
            self.sample_service, self.exp_auth_token)
 
211
        self.assertEqual(403, int(resp['status']))
 
212
 
 
213
        resp, content = utils.get_service(
 
214
            self.sample_service, self.invalid_token)
 
215
        self.assertEqual(404, int(resp['status']))
 
216
 
 
217
 
 
218
class CreateServiceTest(ServicesTest):
 
219
    def test_service_create_json(self):
 
220
        resp, _content = utils.create_service(
 
221
            self.test_service, str(self.auth_token))
 
222
        resp_val = int(resp['status'])
 
223
        self.assertEqual(201, resp_val)
 
224
        utils.delete_service(self.test_service, self.auth_token)
 
225
 
 
226
    def test_service_create_xml(self):
 
227
        resp, _content = utils.create_service_xml(
 
228
            self.test_service, str(self.auth_token))
 
229
        resp_val = int(resp['status'])
 
230
        self.assertEqual(201, resp_val)
 
231
 
 
232
    def test_service_create_duplicate_json(self):
 
233
        resp, _content = utils.create_service(
 
234
            self.test_service, str(self.auth_token))
 
235
        resp_val = int(resp['status'])
 
236
        self.assertEqual(201, resp_val)
 
237
        resp, _content = utils.create_service(
 
238
            self.test_service, str(self.auth_token))
 
239
        resp_val = int(resp['status'])
 
240
        self.assertEqual(409, resp_val)
 
241
        utils.delete_service(self.test_service, self.auth_token)
 
242
 
 
243
    def test_service_create_using_expired_token(self):
 
244
        resp, _content = utils.create_service(
 
245
            self.test_service, str(self.exp_auth_token))
 
246
        resp_val = int(resp['status'])
 
247
        self.assertEqual(403, resp_val)
 
248
 
 
249
    def test_service_create_using_disabled_token(self):
 
250
        resp, _content = utils.create_service(
 
251
            self.test_service, str(self.disabled_token))
 
252
        resp_val = int(resp['status'])
 
253
        self.assertEqual(403, resp_val)
 
254
 
 
255
    def test_service_create_json_using_missing_token(self):
 
256
        resp, _content = utils.create_service(
 
257
            self.test_service, str(self.missing_token))
 
258
        resp_val = int(resp['status'])
 
259
        self.assertEqual(401, resp_val)
 
260
 
 
261
    def test_service_create_json_using_invalid_token(self):
 
262
        resp, _content = utils.create_service(
 
263
            self.test_service, str(self.invalid_token))
 
264
        resp_val = int(resp['status'])
 
265
        self.assertEqual(404, resp_val)
 
266
 
 
267
 
 
268
class DeleteServiceTest(ServicesTest):
 
269
    def test_service_delete(self):
 
270
        resp, _content = utils.delete_service(
 
271
            self.test_service, self.auth_token)
 
272
        resp_val = int(resp['status'])
 
273
        self.assertEqual(204, resp_val)
 
274
 
 
275
    def test_service_which_has_dependencies_delete(self):
 
276
        resp, _content = utils.create_service(
 
277
            self.test_service, str(self.auth_token))
 
278
        resp, _content = utils.create_role_mapped_to_service(
 
279
            self.test_service + ":test_role",
 
280
            self.auth_token, self.test_service)
 
281
        self.assertEqual(201, int(resp['status']))
 
282
        resp, _content = utils.create_role_ref(
 
283
            self.user, self.test_service +
 
284
            ":test_role", self.tenant, self.auth_token)
 
285
        self.assertEqual(201, int(resp['status']))
 
286
        resp, _content = utils.create_endpoint_template(\
 
287
            'DFW', self.test_service, 'public_url',\
 
288
            'admin_url', 'internal_url', True, False, self.auth_token)
 
289
        self.assertEqual(201, int(resp['status']))
 
290
 
 
291
        #verify content
 
292
        obj = json.loads(_content)
 
293
        if not "endpointTemplate" in obj:
 
294
            raise fault.BadRequestFault("Expecting endpointTemplate")
 
295
        endpoint_template = obj["endpointTemplate"]
 
296
        if not "id" in endpoint_template:
 
297
            endpoint_template_id = None
 
298
        else:
 
299
            endpoint_template_id = endpoint_template["id"]
 
300
        resp, _content = utils.create_endpoint_xml(self.tenant,
 
301
            endpoint_template_id,
 
302
            str(self.auth_token))
 
303
        resp_val = int(resp['status'])
 
304
        self.assertEqual(201, resp_val)
 
305
 
 
306
        resp, _content = utils.delete_service(
 
307
            self.test_service, self.auth_token)
 
308
        resp_val = int(resp['status'])
 
309
        self.assertEqual(204, resp_val)
 
310
 
 
311
    def test_service_delete_json_using_expired_token(self):
 
312
        resp, _content = utils.delete_service(
 
313
            self.test_service, str(self.exp_auth_token))
 
314
        resp_val = int(resp['status'])
 
315
        self.assertEqual(403, resp_val)
 
316
 
 
317
    def test_service_delete_json_using_disabled_token(self):
 
318
        resp, _content = utils.delete_service(
 
319
            self.test_service, str(self.disabled_token))
 
320
        resp_val = int(resp['status'])
 
321
        self.assertEqual(403, resp_val)
 
322
 
 
323
    def test_service_delete_json_using_missing_token(self):
 
324
        resp, _content = utils.delete_service(
 
325
            self.test_service, str(self.missing_token))
 
326
        resp_val = int(resp['status'])
 
327
        self.assertEqual(401, resp_val)
 
328
 
 
329
    def test_service_delete_json_using_invalid_token(self):
 
330
        resp, _content = utils.delete_service(
 
331
            self.test_service, str(self.invalid_token))
 
332
        resp_val = int(resp['status'])
 
333
        self.assertEqual(404, resp_val)
 
334
 
 
335
    def test_service_delete_json_using_disabled_token(self):
 
336
        resp, _content = utils.delete_service(
 
337
            self.test_service, str(self.disabled_token))
 
338
        resp_val = int(resp['status'])
 
339
        self.assertEqual(403, resp_val)
 
340
 
 
341
if __name__ == '__main__':
 
342
    unittest.main()