x
/
kingbird
Code Issues Proposed changes

Fix to show version information from root API. 

When the user hits http://serv_api:8118 instead of version information
kingbird returns 401 authentication.Introduced a new section
pecan configurations and pecan hooks to fix this error.
Added test-cases, made trivial changes for tempest tests and refactored
some code.

Closes-Bug: #1673383
Change-Id: Ic87b078a73bd0505b309f099cffdfb7ee31ba8e3
This commit is contained in:
Goutham Pratapa 2017-03-21 12:18:57 +05:30
parent e5dfc05067
commit 8e0acbdcc5
10 changed files with 211 additions and 241 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
doc/source/installation.rst
View File

@ -259,14 +259,11 @@ are running:
Post-installation activities
----------------------------
Run the following commands to check whether kingbird-api is serving, please
replace $token to the token you get from "openstack token issue":
Run the following commands to check whether kingbird-api is serving.
.. code-block:: bash
openstack token issue
curl -H "Content-Type: application/json" -H "X-Auth-Token: $token" \
http://127.0.0.1:8118/
curl http://127.0.0.1:8118/
If the response looks like following: {"versions": [{"status": "CURRENT",
"updated": "2016-03-07", "id": "v1.0", "links": [{"href":

28
kingbird/api/app.py
View File

@ -20,11 +20,13 @@ from oslo_config import cfg
from oslo_middleware import request_id
from oslo_service import service
from kingbird.common import exceptions as k_exc
from kingbird.common import context as ctx
from kingbird.common.i18n import _
def setup_app(*args, **kwargs):
opts = cfg.CONF.pecan
config = {
'server': {
'port': cfg.CONF.bind_port,
@ -33,12 +35,15 @@ def setup_app(*args, **kwargs):
'app': {
'root': 'kingbird.api.controllers.root.RootController',
'modules': ['kingbird.api'],
"debug": opts.debug,
"auth_enable": opts.auth_enable,
'errors': {
400: '/error',
'__force_dict__': True
}
}
}
}
pecan_config = pecan.configuration.conf_from_dict(config)
# app_hooks = [], hook collection will be put here later
@ -48,7 +53,7 @@ def setup_app(*args, **kwargs):
debug=False,
wrap_app=_wrap_app,
force_canonical=False,
hooks=[],
hooks=lambda: [ctx.AuthHook()],
guess_content_type_from_ext=True
)
@ -57,16 +62,17 @@ def setup_app(*args, **kwargs):
def _wrap_app(app):
app = request_id.RequestId(app)
if cfg.CONF.pecan.auth_enable and cfg.CONF.auth_strategy == 'keystone':
conf = dict(cfg.CONF.keystone_authtoken)
# Change auth decisions of requests to the app itself.
conf.update({'delay_auth_decision': True})
if cfg.CONF.auth_strategy == 'noauth':
pass
elif cfg.CONF.auth_strategy == 'keystone':
app = auth_token.AuthProtocol(app, {})
# NOTE: Policy enforcement works only if Keystone
# authentication is enabled. No support for other authentication
# types at this point.
return auth_token.AuthProtocol(app, conf)
else:
raise k_exc.InvalidConfigurationOption(
opt_name='auth_strategy', opt_value=cfg.CONF.auth_strategy)
return app
return app
_launcher = None

33
kingbird/common/config.py
View File

@ -25,6 +25,33 @@ global_opts = [
default=60,
help='Seconds between running periodic reporting tasks.'),
]
# Pecan_opts
pecan_opts = [
cfg.StrOpt(
'root',
default='kingbird.api.controllers.root.RootController',
help='Pecan root controller'
),
cfg.ListOpt(
'modules',
default=["kingbird.api"],
help='A list of modules where pecan will search for applications.'
),
cfg.BoolOpt(
'debug',
default=False,
help='Enables the ability to display tracebacks in the browser and'
'interactively debug during development.'
),
cfg.BoolOpt(
'auth_enable',
default=True,
help='Enables user authentication in pecan.'
)
]
# Global nova quotas for all projects
nova_quotas = [
cfg.IntOpt('quota_instances',
@ -150,9 +177,12 @@ common_opts = [
scheduler_opt_group = cfg.OptGroup('scheduler',
title='Scheduler options for periodic job')
# The group stores Kingbird global limit for all the projects
# The group stores Kingbird global limit for all the projects.
default_quota_group = cfg.OptGroup(name='kingbird_global_limit',
title='Global quota limit for all projects')
# The group stores the pecan configurations.
pecan_group = cfg.OptGroup(name='pecan',
title='Pecan options')
cache_opt_group = cfg.OptGroup(name='cache',
title='OpenStack Credentials')
@ -164,6 +194,7 @@ def list_opts():
yield default_quota_group.name, cinder_quotas
yield cache_opt_group.name, cache_opts
yield scheduler_opt_group.name, scheduler_opts
yield pecan_group.name, pecan_opts
yield None, global_opts
yield None, common_opts

23
kingbird/common/context.py
View File

@ -10,12 +10,17 @@
# License for the specific language governing permissions and limitations
# under the License.
import pecan
from pecan import hooks
from oslo_context import context as base_context
from oslo_utils import encodeutils
from kingbird.common import policy
from kingbird.db import api as db_api
ALLOWED_WITHOUT_AUTH = ['/', '/v1.0']
class RequestContext(base_context.RequestContext):
'''Stores information about the security context.
@ -122,3 +127,21 @@ def get_service_context(**args):
in an OpenStack cloud.
'''
pass
class AuthHook(hooks.PecanHook):
def before(self, state):
if state.request.path in ALLOWED_WITHOUT_AUTH:
return
req = state.request
identity_status = req.headers.get('X-Identity-Status')
service_identity_status = req.headers.get('X-Service-Identity-Status')
if (identity_status == 'Confirmed' or
service_identity_status == 'Confirmed'):
return
if req.headers.get('X-Auth-Token'):
msg = 'Auth token is invalid: %s' % req.headers['X-Auth-Token']
else:
msg = 'Authentication required'
msg = "Failed to validate access token: %s" % str(msg)
pecan.abort(status_code=401, detail=msg)

6
kingbird/tests/tempest/scenario/quota_management/client_tests/test_quota_management_api.py
View File

@ -44,10 +44,11 @@ class KingbirdQMTestJSON(base.BaseKingbirdTest):
def test_kingbird_put_method(self):
new_quota = {"quota_set": {"instances": 15, "cores": 10}}
actual_value = self.create_custom_kingbird_quota(
response = self.create_custom_kingbird_quota(
self.resource_ids["project_id"],
self.resource_ids["target_project_id"],
new_quota)
actual_value = response.json()
expected_value = {
self.resource_ids["target_project_id"]: new_quota["quota_set"]
}
@ -154,8 +155,7 @@ class KingbirdQMTestJSON(base.BaseKingbirdTest):
self.resource_ids["project_id"],
self.resource_ids["target_project_id"],
new_quota)
result = eval(response).get('error').get('code')
self.assertEqual(result, 401)
self.assertEqual(response.status_code, 401)
def test_quota_sync_for_project(self):
# Delete custom quota if there are any for this project

2
kingbird/tests/tempest/scenario/quota_management/sync_client.py
View File

@ -139,7 +139,7 @@ def create_custom_kingbird_quota(token, project_id, target_project_id,
quota_api_url)
url_string = url_string + target_project_id
response = requests.put(url_string, headers=headers, data=body)
return response.text
return response
def get_custom_kingbird_quota(token, project_id, target_project_id):

7
kingbird/tests/unit/api/controllers/test_sync_manager.py
View File

@ -32,11 +32,12 @@ FAKE_RESOURCE_ID = 'fake_id'
FAKE_RESOURCE_TYPE = 'keypair'
FAKE_TENANT = utils.UUID1
FAKE_JOB = utils.UUID2
FAKE_URL = '/v1.0/' + FAKE_TENANT + '/os-sync/'
WRONG_URL = '/v1.0/wrong/os-sync/'
FAKE_URL = '/v1.0/' + FAKE_TENANT + '/os-sync'
WRONG_URL = '/v1.0/wrong/os-sync'
fake_user = utils.UUID3
FAKE_STATUS = consts.JOB_PROGRESS
FAKE_HEADERS = {'X-Tenant-Id': FAKE_TENANT, 'X_ROLE': 'admin'}
FAKE_HEADERS = {'X-Tenant-Id': FAKE_TENANT, 'X_ROLE': 'admin',
'X-Identity-Status': 'Confirmed'}
NON_ADMIN_HEADERS = {'X-Tenant-Id': FAKE_TENANT}

81
kingbird/tests/unit/api/test_quota_class.py
View File

@ -16,15 +16,20 @@ import mock
import webtest
from oslo_config import cfg
from oslo_utils import uuidutils
from kingbird.api.controllers.v1 import quota_class
from kingbird.common import config
from kingbird.tests.unit.api import testroot
from kingbird.tests import utils
config.register_options()
OPT_GROUP_NAME = 'keystone_authtoken'
cfg.CONF.import_group(OPT_GROUP_NAME, "keystonemiddleware.auth_token")
FAKE_TENANT = utils.UUID1
ADMIN_HEADERS = {'X-Tenant-Id': FAKE_TENANT, 'X_ROLE': 'admin',
'X-Identity-Status': 'Confirmed'}
NON_ADMIN_HEADERS = {'X-Tenant-Id': FAKE_TENANT,
'X-Identity-Status': 'Confirmed'}
class Result(object):
@ -42,32 +47,29 @@ class TestQuotaClassController(testroot.KBApiTest):
@mock.patch.object(quota_class, 'db_api')
def test_get_all_admin(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
result = Result('class1', 'ram', 100)
mock_db_api.quota_class_get_all_by_name.return_value = \
{"class_name": result.class_name,
result.resource: result.hard_limit}
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % fake_tenant
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % FAKE_TENANT
response = self.app.get(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual({'quota_class_set': {'id': 'class1', 'ram': 100}},
eval(response.text))
def test_get_invalid_req(self):
FAKE_URL = '/v1.0/dummy/os-quota-class-sets/default'
FAKE_HEADERS = {'X_ROLE': 'nonadmin'}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.get, FAKE_URL,
headers=FAKE_HEADERS)
headers=NON_ADMIN_HEADERS)
def test_get_invalid_req_with_admin(self):
FAKE_URL = '/v1.0/dummy/os-quota-class-sets/default'
FAKE_HEADERS = {'X_ROLE': 'admin'}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.get, FAKE_URL,
headers=FAKE_HEADERS)
headers=ADMIN_HEADERS)
@mock.patch.object(quota_class, 'db_api')
def test_put_admin(self, mock_db_api):
@ -76,11 +78,10 @@ class TestQuotaClassController(testroot.KBApiTest):
{"class_name": result.class_name,
result.resource: result.hard_limit}
data = {"quota_class_set": {result.resource: result.hard_limit}}
fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % fake_tenant
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % FAKE_TENANT
response = self.app.put_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'},
headers=ADMIN_HEADERS,
params=data)
self.assertEqual(response.status_int, 200)
self.assertEqual({'quota_class_set': {'id': 'class1', 'cores': 10}},
@ -89,61 +90,47 @@ class TestQuotaClassController(testroot.KBApiTest):
@mock.patch.object(quota_class, 'db_api')
def test_delete_all_admin(self, mock_db_api):
result = Result('class1', 'cores', 10)
fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % fake_tenant
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % FAKE_TENANT
mock_db_api.quota_destroy_all.return_value = result
response = self.app.delete_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
def test_delete_all_non_admin(self):
fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % fake_tenant
try:
self.app.delete_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant})
except webtest.app.AppError as admin_exception:
self.assertIn('Admin required', admin_exception.message)
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % FAKE_TENANT
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.delete_json, fake_url,
headers=NON_ADMIN_HEADERS)
def test_delete_invalid_req_nonadmin(self):
FAKE_URL = '/v1.0/dummy/os-quota-class-sets/default'
FAKE_HEADERS = {'X_ROLE': 'nonadmin'}
fake_url = '/v1.0/dummy/os-quota-class-sets/default'
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.delete, FAKE_URL,
headers=FAKE_HEADERS)
self.app.delete, fake_url,
headers=NON_ADMIN_HEADERS)
def test_delete_invalid_req_admin(self):
FAKE_URL = '/v1.0/dummy/os-quota-class-sets/default'
FAKE_HEADERS = {'X_ROLE': 'admin'}
fake_url = '/v1.0/dummy/os-quota-class-sets/default'
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.delete, FAKE_URL,
headers=FAKE_HEADERS)
self.app.delete, fake_url,
headers=ADMIN_HEADERS)
def test_put_non_admin(self):
result = Result('class1', 'cores', 10)
data = {"quota_class_set": {result.resource: result.hard_limit}}
fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % fake_tenant
try:
self.app.put_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant},
params=data)
except webtest.app.AppError as admin_exception:
self.assertIn('Admin required', admin_exception.message)
fake_url = '/v1.0/%s/os-quota-class-sets/class1' % FAKE_TENANT
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.delete, fake_url,
headers=NON_ADMIN_HEADERS, params=data)
def test_put_invalid_req_non_admin(self):
FAKE_URL = '/v1.0/dummy/os-quota-class-sets/default'
FAKE_HEADERS = {'X_ROLE': 'non-admin'}
fake_url = '/v1.0/dummy/os-quota-class-sets/default'
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put, FAKE_URL,
headers=FAKE_HEADERS)
self.app.put, fake_url,
headers=NON_ADMIN_HEADERS)
def test_put_invalid_req_with_admin(self):
FAKE_URL = '/v1.0/dummy/os-quota-class-sets/default'
FAKE_HEADERS = {'X_ROLE': 'admin'}
fake_url = '/v1.0/dummy/os-quota-class-sets/default'
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put, FAKE_URL,
headers=FAKE_HEADERS)
self.app.put, fake_url,
headers=ADMIN_HEADERS)

255
kingbird/tests/unit/api/test_quota_manager.py
View File

@ -17,15 +17,22 @@ import mock
import webtest
from oslo_config import cfg
from oslo_utils import uuidutils
from kingbird.api.controllers import quota_manager
from kingbird.common import config
from kingbird.rpc import client as rpc_client
from kingbird.tests.unit.api import testroot
from kingbird.tests import utils
config.register_options()
OPT_GROUP_NAME = 'keystone_authtoken'
cfg.CONF.import_group(OPT_GROUP_NAME, "keystonemiddleware.auth_token")
FAKE_TENANT = utils.UUID1
TARGET_FAKE_TENANT = utils.UUID2
ADMIN_HEADERS = {'X-Tenant-Id': FAKE_TENANT, 'X_ROLE': 'admin',
'X-Identity-Status': 'Confirmed'}
NON_ADMIN_HEADERS = {'X-Tenant-Id': TARGET_FAKE_TENANT,
'X-Identity-Status': 'Confirmed'}
class Result(object):
@ -44,20 +51,18 @@ class TestQuotaManager(testroot.KBApiTest):
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_get_all_admin(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
Res = Result(fake_tenant, 'ram', 100)
Res = Result(TARGET_FAKE_TENANT, 'ram', 100)
mock_db_api.quota_get_all_by_project.return_value = \
{"project_id": Res.project_id,
Res.resource: Res.hard_limit}
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
response = self.app.get(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual({'quota_set':
{'project_id': fake_tenant, 'ram': 100}},
{'project_id': TARGET_FAKE_TENANT, 'ram': 100}},
eval(response.text))
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@ -65,12 +70,11 @@ class TestQuotaManager(testroot.KBApiTest):
def test_get_default_admin(self, mock_db_api):
mock_db_api.quota_class_get_default.return_value = \
{'class_name': 'default'}
fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/defaults'\
% (fake_tenant)
% (FAKE_TENANT)
response = self.app.get(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
result = eval(response.text)
for resource in result['quota_set']:
@ -83,29 +87,25 @@ class TestQuotaManager(testroot.KBApiTest):
expected_usage = {"ram": 10}
mock_rpc_client().get_total_usage_for_tenant.return_value = \
expected_usage
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s/detail'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
response = self.app.get(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual(eval(response.body), {"quota_set": expected_usage})
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_put_admin(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
Res = Result(target_fake_tenant, 'cores', 10)
Res = Result(TARGET_FAKE_TENANT, 'cores', 10)
mock_db_api.quota_update.return_value = Res
data = {"quota_set": {Res.resource: Res.hard_limit}}
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
response = self.app.put_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'},
headers=ADMIN_HEADERS,
params=data)
self.assertEqual(response.status_int, 200)
self.assertEqual({Res.project_id: {Res.resource: Res.hard_limit}},
@ -114,16 +114,14 @@ class TestQuotaManager(testroot.KBApiTest):
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_delete_admin(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
Res = Result(target_fake_tenant, 'cores', 10)
Res = Result(TARGET_FAKE_TENANT, 'cores', 10)
mock_db_api.quota_destroy.return_value = Res
data = {"quota_set": [Res.resource]}
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
response = self.app.delete_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'},
headers=ADMIN_HEADERS,
params=data)
self.assertEqual(response.status_int, 200)
self.assertEqual({'Deleted quota limits': [Res.resource]},
@ -132,101 +130,76 @@ class TestQuotaManager(testroot.KBApiTest):
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_delete_all_admin(self, mock_db_api):
Res = Result('tenant_1', 'cores', 10)
Res = Result(TARGET_FAKE_TENANT, 'cores', 10)
mock_db_api.quota_destroy_all.return_value = Res
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
response = self.app.delete_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual('Deleted all quota limits for the given project',
eval(response.text))
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_quota_sync_admin(self):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s/sync'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
response = self.app.put_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant,
'X_ROLE': 'admin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual("triggered quota sync for " + target_fake_tenant,
self.assertEqual("triggered quota sync for " + TARGET_FAKE_TENANT,
eval(response.text))
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_put_nonadmin(self):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
Res = Result('tenant_1', 'cores', 10)
Res = Result(TARGET_FAKE_TENANT, 'cores', 10)
data = {"quota_set": {Res.resource: Res.hard_limit}}
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
try:
self.app.put_json(fake_url,
headers={'X-Tenant-Id': fake_tenant},
params=data)
except webtest.app.AppError as admin_exception:
self.assertIn('Admin required', admin_exception.message)
% (TARGET_FAKE_TENANT, FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.put_json, fake_url,
headers=NON_ADMIN_HEADERS,
params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_delete_all_nonadmin(self):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
try:
self.app.delete_json(fake_url,
headers={'X-Tenant-Id': fake_tenant})
except webtest.app.AppError as admin_exception:
self.assertIn('Admin required', admin_exception.message)
% (TARGET_FAKE_TENANT, FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.delete_json, fake_url,
headers=NON_ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_delete_nonadmin(self):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
Res = Result(target_fake_tenant, 'cores', 10)
Res = Result(TARGET_FAKE_TENANT, 'cores', 10)
data = {"quota_set": {Res.resource: Res.hard_limit}}
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
try:
self.app.delete_json(fake_url,
headers={'X-Tenant-Id': fake_tenant},
params=data)
except webtest.app.AppError as admin_exception:
self.assertIn('Admin required', admin_exception.message)
% (TARGET_FAKE_TENANT, FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.delete_json, fake_url,
headers=NON_ADMIN_HEADERS,
params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_quota_sync_nonadmin(self):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s/sync'\
% (fake_tenant, target_fake_tenant)
try:
self.app.put_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant})
except webtest.app.AppError as admin_exception:
self.assertIn('Admin required', admin_exception.message)
% (TARGET_FAKE_TENANT, FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.put_json, fake_url,
headers=NON_ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_get_default_nonadmin(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/defaults'\
% (fake_tenant)
% (FAKE_TENANT)
mock_db_api.quota_class_get_default.return_value = \
{'class_name': 'default'}
response = self.app.get(
fake_url,
headers={'X_TENANT_ID': fake_tenant, 'X_USER_ID': 'nonadmin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
result = eval(response.text)
for resource in result['quota_set']:
@ -236,120 +209,77 @@ class TestQuotaManager(testroot.KBApiTest):
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_quota_sync_bad_request(self):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-ssdfets/%s/sync'\
% (fake_tenant, target_fake_tenant)
try:
self.app.post_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant,
'X_ROLE': 'admin'})
except webtest.app.AppError as bad_method_exception:
self.assertIn('Bad response: 404 Not Found',
bad_method_exception.message)
% (TARGET_FAKE_TENANT, FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "404 *",
self.app.post_json, fake_url,
headers=NON_ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_put_invalid_payload(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
Res = Result(fake_tenant, 'cores', 10)
Res = Result(FAKE_TENANT, 'cores', 10)
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
mock_db_api.quota_update.return_value = Res
data = {'quota': {Res.resource: Res.hard_limit}}
try:
self.app.put_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'},
params=data)
except webtest.app.AppError as invalid_payload_exception:
self.assertIn('400 Bad Request',
invalid_payload_exception.message)
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put_json, fake_url,
headers=ADMIN_HEADERS, params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_put_invalid_input(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
Res = Result(fake_tenant, 'cores', -10)
Res = Result(FAKE_TENANT, 'cores', -10)
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
mock_db_api.quota_update.return_value = Res
data = {"quota_set": {Res.resource: Res.hard_limit}}
try:
self.app.put_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'},
params=data)
except webtest.app.AppError as invalid_input_exception:
self.assertIn('400 Bad Request',
invalid_input_exception.message)
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put_json, fake_url,
headers=ADMIN_HEADERS, params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_delete_invalid_quota(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
Res = Result('tenant_1', 'invalid_quota', 10)
mock_db_api.quota_destroy.return_value = Res
data = {"quota_set": [Res.resource]}
try:
self.app.delete_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant, 'X_ROLE': 'admin'},
params=data)
except webtest.app.AppError as invalid_quota_exception:
self.assertIn('The resource could not be found',
invalid_quota_exception.message)
self.assertRaisesRegexp(webtest.app.AppError, "404 *",
self.app.delete_json, fake_url,
headers=ADMIN_HEADERS, params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_quota_sync_bad_action(self):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s/syncing'\
% (fake_tenant, target_fake_tenant)
try:
self.app.put_json(
fake_url,
headers={'X-Tenant-Id': fake_tenant,
'X_ROLE': 'admin'})
except webtest.app.AppError as bad_method_exception:
self.assertIn('Invalid action, only sync is allowed',
bad_method_exception.message)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "404 *",
self.app.delete_json, fake_url,
headers=ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_get_all_another_tenant_nonadmin(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
FAKE_HEADERS = {'X_TENANT_ID': fake_tenant,
'X_USER_ID': 'nonadmin'}
% (TARGET_FAKE_TENANT, FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.get, fake_url,
headers=FAKE_HEADERS)
headers=NON_ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_get_all_another_tenant_with_admin(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
Res = Result('tenant_1', 'ram', 100)
mock_db_api.quota_get_all_by_project.return_value = \
{"project_id": Res.project_id,
Res.resource: Res.hard_limit}
response = self.app.get(
fake_url,
headers={'X_TENANT_ID': fake_tenant, 'X_ROLE': 'admin',
'X_USER_ID': 'nonadmin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual({'quota_set': {'project_id': 'tenant_1', 'ram': 100}},
eval(response.text))
@ -357,76 +287,63 @@ class TestQuotaManager(testroot.KBApiTest):
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_get_usages_another_tenant_no_admin(self, mock_db_api):
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s/detail'\
% (fake_tenant, target_fake_tenant)
FAKE_HEADERS = {'X_TENANT_ID': fake_tenant,
'X_USER_ID': 'nonadmin'}
% (TARGET_FAKE_TENANT, FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.get, fake_url,
headers=FAKE_HEADERS)
headers=NON_ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
def test_get_usages_another_tenant_admin(self, mock_rpc_client):
expected_usage = {"ram": 10}
fake_tenant = uuidutils.generate_uuid()
target_fake_tenant = uuidutils.generate_uuid()
fake_url = '/v1.0/%s/os-quota-sets/%s/detail'\
% (fake_tenant, target_fake_tenant)
% (FAKE_TENANT, TARGET_FAKE_TENANT)
mock_rpc_client().get_total_usage_for_tenant.return_value = \
expected_usage
response = self.app.get(
fake_url,
headers={'X_TENANT_ID': fake_tenant, 'X_ROLE': 'admin',
'X_USER_ID': 'admin'})
headers=ADMIN_HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual(eval(response.body), {"quota_set": expected_usage})
@mock.patch.object(rpc_client, 'EngineClient')
def test_get_invalid_curl_req_nonadmin(self, mock_rpc_client):
FAKE_URL = '/v1.0/dummy/os-quota-sets/defaults'
FAKE_HEADERS = {'X_ROLE': 'nonadmin'}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.get, FAKE_URL,
headers=FAKE_HEADERS)
headers=NON_ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
def test_get_invalid_curl_req_admin(self, mock_rpc_client):
FAKE_URL = '/v1.0/dummy/os-quota-sets/defaults'
FAKE_HEADERS = {'X_ROLE': 'admin'}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.get, FAKE_URL,
headers=FAKE_HEADERS)
headers=ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
def test_put_invalid_curl_req_nonadmin(self, mock_rpc_client):
FAKE_URL = '/v1.0/dummy/os-quota-sets/dummy2/sync'
FAKE_HEADERS = {'X_ROLE': 'nonadmin'}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put, FAKE_URL,
headers=FAKE_HEADERS)
headers=NON_ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
def test_put_invalid_curl_req_admin(self, mock_rpc_client):
FAKE_URL = '/v1.0/dummy/os-quota-sets/dummy2'
FAKE_HEADERS = {'X_ROLE': 'admin'}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put, FAKE_URL,
headers=FAKE_HEADERS)
headers=ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
def test_delete_invalid_curl_req_nonadmin(self, mock_rpc_client):
FAKE_URL = '/v1.0/dummy/os-quota-sets/dummy2'
FAKE_HEADERS = {'X_ROLE': 'nonadmin'}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.delete, FAKE_URL,
headers=FAKE_HEADERS)
headers=NON_ADMIN_HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
def test_delete_invalid_curl_req_admin(self, mock_rpc_client):
FAKE_URL = '/v1.0/dummy/os-quota-sets/dummy2'
FAKE_HEADERS = {'X_ROLE': 'admin'}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.delete, FAKE_URL,
headers=FAKE_HEADERS)
headers=NON_ADMIN_HEADERS)

10
kingbird/tests/unit/api/testroot.py
View File

@ -188,6 +188,14 @@ class TestKeystoneAuth(KBApiTest):
self.app = self._make_app()
def test_auth_not_enforced_for_root(self):
response = self.app.get('/')
self.assertEqual(response.status_int, 200)
def test_auth_not_enforced_for_v1(self):
response = self.app.get('/v1.0')
self.assertEqual(response.status_int, 200)
def test_auth_enforced(self):
response = self.app.get('/', expect_errors=True)
response = self.app.get('/v1.0/', expect_errors=True)
self.assertEqual(response.status_int, 401)