1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2013 OpenStack Foundation
5
# Licensed under the Apache License, Version 2.0 (the "License"); you may
6
# not use this file except in compliance with the License. You may obtain
7
# a copy of the License at
9
# http://www.apache.org/licenses/LICENSE-2.0
11
# Unless required by applicable law or agreed to in writing, software
12
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14
# License for the specific language governing permissions and limitations
20
from keystone.common.sql import migration
21
from keystone import contrib
22
from keystone.openstack.common import importutils
23
from keystone.tests import core as test
28
class TestExtensionCase(test_v3.RestfulTestCase):
30
EXTENSION_NAME = 'endpoint_filter'
31
EXTENSION_TO_ADD = 'endpoint_filter_extension'
33
def setup_database(self):
34
self.conf_files = super(TestExtensionCase, self).config_files()
35
self.conf_files.append(
36
test.testsdir('test_associate_project_endpoint_extension.conf'))
37
super(TestExtensionCase, self).setup_database()
38
package_name = "%s.%s.migrate_repo" % (contrib.__name__,
40
package = importutils.import_module(package_name)
41
self.repo_path = os.path.abspath(
42
os.path.dirname(package.__file__))
43
migration.db_version_control(version=None, repo_path=self.repo_path)
44
migration.db_sync(version=None, repo_path=self.repo_path)
47
super(TestExtensionCase, self).setUp()
48
self.default_request_url = (
49
'/OS-EP-FILTER/projects/%(project_id)s'
50
'/endpoints/%(endpoint_id)s' % {
51
'project_id': self.default_domain_project_id,
52
'endpoint_id': self.endpoint_id})
55
super(TestExtensionCase, self).tearDown()
59
class AssociateEndpointProjectFilterCRUDTestCase(TestExtensionCase):
60
"""Test OS-EP-FILTER endpoint to project associations extension."""
62
# endpoint-project associations crud tests
64
def test_create_endpoint_project_assoc(self):
65
"""PUT /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
67
Valid endpoint and project id test case.
70
self.put(self.default_request_url,
74
def test_create_endpoint_project_assoc_noproj(self):
75
"""PUT OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
77
Invalid project id test case.
80
self.put('/OS-EP-FILTER/projects/%(project_id)s'
81
'/endpoints/%(endpoint_id)s' % {
82
'project_id': uuid.uuid4().hex,
83
'endpoint_id': self.endpoint_id},
87
def test_create_endpoint_project_assoc_noendp(self):
88
"""PUT /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
90
Invalid endpoint id test case.
93
self.put('/OS-EP-FILTER/projects/%(project_id)s'
94
'/endpoints/%(endpoint_id)s' % {
95
'project_id': self.default_domain_project_id,
96
'endpoint_id': uuid.uuid4().hex},
100
def test_create_endpoint_project_assoc_unexpected_body(self):
101
"""PUT /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
103
Unexpected body in request. The body should be ignored.
106
self.put(self.default_request_url,
107
body={'project_id': self.default_domain_project_id},
111
def test_check_endpoint_project_assoc(self):
112
"""HEAD /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
114
Valid project and endpoint id test case.
117
self.put(self.default_request_url,
120
self.head('/OS-EP-FILTER/projects/%(project_id)s'
121
'/endpoints/%(endpoint_id)s' % {
122
'project_id': self.default_domain_project_id,
123
'endpoint_id': self.endpoint_id},
126
def test_check_endpoint_project_assoc_noproj(self):
127
"""HEAD /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
129
Invalid project id test case.
132
self.put(self.default_request_url)
133
self.head('/OS-EP-FILTER/projects/%(project_id)s'
134
'/endpoints/%(endpoint_id)s' % {
135
'project_id': uuid.uuid4().hex,
136
'endpoint_id': self.endpoint_id},
140
def test_check_endpoint_project_assoc_noendp(self):
141
"""HEAD /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
143
Invalid endpoint id test case.
146
self.put(self.default_request_url)
147
self.head('/OS-EP-FILTER/projects/%(project_id)s'
148
'/endpoints/%(endpoint_id)s' % {
149
'project_id': self.default_domain_project_id,
150
'endpoint_id': uuid.uuid4().hex},
155
def test_get_endpoint_project_assoc(self):
156
"""GET /OS-EP-FILTER/projects/{project_id}/endpoints success."""
157
self.put(self.default_request_url)
158
r = self.get('/OS-EP-FILTER/projects/%(project_id)s/endpoints' % {
159
'project_id': self.default_domain_project_id})
160
self.assertValidEndpointListResponse(r, self.endpoint)
162
def test_get_endpoint_project_assoc_noproj(self):
163
"""GET /OS-EP-FILTER/projects/{project_id}/endpoints no project."""
164
self.put(self.default_request_url)
165
self.get('/OS-EP-FILTER/projects/%(project_id)s/endpoints' % {
166
'project_id': uuid.uuid4().hex},
171
def test_remove_endpoint_project_assoc(self):
172
"""DELETE /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
174
Valid project id and endpoint id test case.
177
self.put(self.default_request_url)
178
self.delete('/OS-EP-FILTER/projects/%(project_id)s'
179
'/endpoints/%(endpoint_id)s' % {
180
'project_id': self.default_domain_project_id,
181
'endpoint_id': self.endpoint_id},
184
def test_remove_endpoint_project_assoc_noproj(self):
185
"""DELETE /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
187
Invalid project id test case.
190
self.put(self.default_request_url)
191
self.delete('/OS-EP-FILTER/projects/%(project_id)s'
192
'/endpoints/%(endpoint_id)s' % {
193
'project_id': uuid.uuid4().hex,
194
'endpoint_id': self.endpoint_id},
198
def test_remove_endpoint_project_assoc_noendp(self):
199
"""DELETE /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id}
201
Invalid endpoint id test case.
204
self.put(self.default_request_url)
205
self.delete('/OS-EP-FILTER/projects/%(project_id)s'
206
'/endpoints/%(endpoint_id)s' % {
207
'project_id': self.default_domain_project_id,
208
'endpoint_id': uuid.uuid4().hex},
213
class AssociateProjectEndpointFilterTokenRequestTestCase(TestExtensionCase):
214
"""Test OS-EP-FILTER catalog filtering extension."""
216
def test_default_project_id_scoped_token_with_user_id_ep_filter(self):
217
# create a second project to work with
218
ref = self.new_project_ref(domain_id=self.domain_id)
219
r = self.post('/projects', body={'project': ref})
220
project = self.assertValidProjectResponse(r, ref)
222
# grant the user a role on the project
224
'/projects/%(project_id)s/users/%(user_id)s/roles/%(role_id)s' % {
225
'user_id': self.user['id'],
226
'project_id': project['id'],
227
'role_id': self.role['id']})
229
# set the user's preferred project
230
body = {'user': {'default_project_id': project['id']}}
231
r = self.patch('/users/%(user_id)s' % {
232
'user_id': self.user['id']},
234
self.assertValidUserResponse(r)
236
# add one endpoint to the project
237
self.put('/OS-EP-FILTER/projects/%(project_id)s'
238
'/endpoints/%(endpoint_id)s' % {
239
'project_id': project['id'],
240
'endpoint_id': self.endpoint_id},
244
# attempt to authenticate without requesting a project
245
auth_data = self.build_authentication_request(
246
user_id=self.user['id'],
247
password=self.user['password'])
248
r = self.post('/auth/tokens', body=auth_data)
249
self.assertValidProjectScopedTokenResponse(
251
require_catalog=True,
252
endpoint_filter=True,
254
self.assertEqual(r.result['token']['project']['id'], project['id'])
256
def test_implicit_project_id_scoped_token_with_user_id_ep_filter(self):
257
# attempt to authenticate without requesting a project
259
# add one endpoint to default project
260
self.put('/OS-EP-FILTER/projects/%(project_id)s'
261
'/endpoints/%(endpoint_id)s' % {
262
'project_id': self.project['id'],
263
'endpoint_id': self.endpoint_id},
267
auth_data = self.build_authentication_request(
268
user_id=self.user['id'],
269
password=self.user['password'],
270
project_id=self.project['id'])
271
r = self.post('/auth/tokens', body=auth_data)
272
self.assertValidProjectScopedTokenResponse(
274
require_catalog=True,
275
endpoint_filter=True,
277
self.assertEqual(r.result['token']['project']['id'],
280
def test_default_project_id_scoped_token_ep_filter_no_catalog(self):
281
# create a second project to work with
282
ref = self.new_project_ref(domain_id=self.domain_id)
283
r = self.post('/projects', body={'project': ref})
284
project = self.assertValidProjectResponse(r, ref)
286
# grant the user a role on the project
288
'/projects/%(project_id)s/users/%(user_id)s/roles/%(role_id)s' % {
289
'user_id': self.user['id'],
290
'project_id': project['id'],
291
'role_id': self.role['id']})
293
# set the user's preferred project
294
body = {'user': {'default_project_id': project['id']}}
295
r = self.patch('/users/%(user_id)s' % {
296
'user_id': self.user['id']},
298
self.assertValidUserResponse(r)
300
# add one endpoint to the project
301
self.put('/OS-EP-FILTER/projects/%(project_id)s'
302
'/endpoints/%(endpoint_id)s' % {
303
'project_id': project['id'],
304
'endpoint_id': self.endpoint_id},
308
# attempt to authenticate without requesting a project
309
auth_data = self.build_authentication_request(
310
user_id=self.user['id'],
311
password=self.user['password'])
312
r = self.post('/auth/tokens?nocatalog', body=auth_data)
313
self.assertValidProjectScopedTokenResponse(
315
require_catalog=False,
316
endpoint_filter=True,
318
self.assertEqual(r.result['token']['project']['id'], project['id'])
320
def test_implicit_project_id_scoped_token_ep_filter_no_catalog(self):
321
# attempt to authenticate without requesting a project
323
# add one endpoint to default project
324
self.put('/OS-EP-FILTER/projects/%(project_id)s'
325
'/endpoints/%(endpoint_id)s' % {
326
'project_id': self.project['id'],
327
'endpoint_id': self.endpoint_id},
331
auth_data = self.build_authentication_request(
332
user_id=self.user['id'],
333
password=self.user['password'],
334
project_id=self.project['id'])
335
r = self.post('/auth/tokens?nocatalog', body=auth_data)
336
self.assertValidProjectScopedTokenResponse(
338
require_catalog=False,
339
endpoint_filter=True,
341
self.assertEqual(r.result['token']['project']['id'],
344
def test_default_project_id_scoped_token_ep_filter_full_catalog(self):
345
# create a second project to work with
346
ref = self.new_project_ref(domain_id=self.domain_id)
347
r = self.post('/projects', body={'project': ref})
348
project = self.assertValidProjectResponse(r, ref)
350
# grant the user a role on the project
352
'/projects/%(project_id)s/users/%(user_id)s/roles/%(role_id)s' % {
353
'user_id': self.user['id'],
354
'project_id': project['id'],
355
'role_id': self.role['id']})
357
# set the user's preferred project
358
body = {'user': {'default_project_id': project['id']}}
359
r = self.patch('/users/%(user_id)s' % {
360
'user_id': self.user['id']},
362
self.assertValidUserResponse(r)
364
# attempt to authenticate without requesting a project
365
auth_data = self.build_authentication_request(
366
user_id=self.user['id'],
367
password=self.user['password'])
368
r = self.post('/auth/tokens?nocatalog', body=auth_data)
369
self.assertValidProjectScopedTokenResponse(
371
require_catalog=False,
372
endpoint_filter=True)
373
self.assertEqual(r.result['token']['project']['id'], project['id'])
375
def test_implicit_project_id_scoped_token_ep_filter_full_catalog(self):
376
# attempt to authenticate without requesting a project
378
auth_data = self.build_authentication_request(
379
user_id=self.user['id'],
380
password=self.user['password'],
381
project_id=self.project['id'])
382
r = self.post('/auth/tokens?nocatalog', body=auth_data)
383
self.assertValidProjectScopedTokenResponse(
385
require_catalog=False,
386
endpoint_filter=True,)
387
self.assertEqual(r.result['token']['project']['id'],
390
def test_implicit_project_id_scoped_token_handling_bad_reference(self):
391
# handling the case with an endpoint that is not associate with
393
# add first endpoint to default project
394
self.put('/OS-EP-FILTER/projects/%(project_id)s'
395
'/endpoints/%(endpoint_id)s' % {
396
'project_id': self.project['id'],
397
'endpoint_id': self.endpoint_id},
401
# create a second temporary endpoint
402
self.endpoint_id2 = uuid.uuid4().hex
403
self.endpoint2 = self.new_endpoint_ref(service_id=self.service_id)
404
self.endpoint2['id'] = self.endpoint_id2
405
self.catalog_api.create_endpoint(
407
self.endpoint2.copy())
409
# add second endpoint to default project
410
self.put('/OS-EP-FILTER/projects/%(project_id)s'
411
'/endpoints/%(endpoint_id)s' % {
412
'project_id': self.project['id'],
413
'endpoint_id': self.endpoint_id2},
417
# remove the temporary reference
418
# this will create inconsistency in the endpoint filter table
419
# which is fixed during the catalog creation for token request
420
self.catalog_api.delete_endpoint(self.endpoint_id2)
422
auth_data = self.build_authentication_request(
423
user_id=self.user['id'],
424
password=self.user['password'],
425
project_id=self.project['id'])
426
r = self.post('/auth/tokens', body=auth_data)
427
self.assertValidProjectScopedTokenResponse(
429
require_catalog=True,
430
endpoint_filter=True,
432
self.assertEqual(r.result['token']['project']['id'],