openstack/nova
Code Issues Proposed changes

Merge "Move policy enforcement into REST API layer for v2.1 lock server"

Browse Source
This commit is contained in:
Jenkins 2015-02-12 20:05:32 +00:00 committed by Gerrit Code Review
commit bcc900577f
2 changed files with 41 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
nova/api/openstack/compute/plugins/v3/lock_server.py
View File

@ -20,16 +20,13 @@ from nova import compute
ALIAS = "os-lock-server" ALIAS = "os-lock-server"
authorize = extensions.os_compute_authorizer(ALIAS)
def authorize(context, action_name):
action = 'v3:%s:%s' % (ALIAS, action_name)
extensions.extension_authorizer('compute', action)(context)
class LockServerController(wsgi.Controller): class LockServerController(wsgi.Controller):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super(LockServerController, self).__init__(*args, **kwargs) super(LockServerController, self).__init__(*args, **kwargs)
self.compute_api = compute.API() self.compute_api = compute.API(skip_policy_check=True)
@wsgi.response(202) @wsgi.response(202)
@extensions.expected_errors(404) @extensions.expected_errors(404)
@ -37,7 +34,7 @@ class LockServerController(wsgi.Controller):
def _lock(self, req, id, body): def _lock(self, req, id, body):
"""Lock a server instance.""" """Lock a server instance."""
context = req.environ['nova.context'] context = req.environ['nova.context']
authorize(context, 'lock') authorize(context, action='lock')
instance = common.get_instance(self.compute_api, context, id, instance = common.get_instance(self.compute_api, context, id,
want_objects=True) want_objects=True)
self.compute_api.lock(context, instance) self.compute_api.lock(context, instance)
@ -48,7 +45,7 @@ class LockServerController(wsgi.Controller):
def _unlock(self, req, id, body): def _unlock(self, req, id, body):
"""Unlock a server instance.""" """Unlock a server instance."""
context = req.environ['nova.context'] context = req.environ['nova.context']
authorize(context, 'unlock') authorize(context, action='unlock')
instance = common.get_instance(self.compute_api, context, id, instance = common.get_instance(self.compute_api, context, id,
want_objects=True) want_objects=True)
self.compute_api.unlock(context, instance) self.compute_api.unlock(context, instance)

41
nova/tests/unit/api/openstack/compute/contrib/test_lock_server.py
View File

@ -13,11 +13,12 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from nova.api.openstack.compute.contrib import admin_actions as \ from nova.api.openstack.compute.contrib import (admin_actions as
lock_server_v2 lock_server_v2)
from nova.api.openstack.compute.plugins.v3 import lock_server as \ from nova.api.openstack.compute.plugins.v3 import (lock_server as
lock_server_v21 lock_server_v21)
from nova import exception from nova import exception
from nova import test
from nova.tests.unit.api.openstack.compute import admin_only_action_common from nova.tests.unit.api.openstack.compute import admin_only_action_common
from nova.tests.unit.api.openstack import fakes from nova.tests.unit.api.openstack import fakes
@ -79,3 +80,35 @@ class LockServerTestsV2(LockServerTestsV21):
def _get_app(self): def _get_app(self):
return fakes.wsgi_app(init_only=('servers',), return fakes.wsgi_app(init_only=('servers',),
fake_auth_context=self.context) fake_auth_context=self.context)
class LockServerPolicyEnforcementV21(test.NoDBTestCase):
def setUp(self):
super(LockServerPolicyEnforcementV21, self).setUp()
self.controller = lock_server_v21.LockServerController()
self.req = fakes.HTTPRequest.blank('')
def test_lock_policy_failed(self):
rule_name = "compute_extension:v3:os-lock-server:lock"
self.policy.set_rules({rule_name: "project:non_fake"})
exc = self.assertRaises(
exception.PolicyNotAuthorized,
self.controller._lock, self.req,
fakes.FAKE_UUID,
body={'lock': {}})
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule_name,
exc.format_message())
def test_unlock_policy_failed(self):
rule_name = "compute_extension:v3:os-lock-server:unlock"
self.policy.set_rules({rule_name: "project:non_fake"})
exc = self.assertRaises(
exception.PolicyNotAuthorized,
self.controller._unlock, self.req,
fakes.FAKE_UUID,
body={'unlock': {}})
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule_name,
exc.format_message())