openstack
/
magnum
Code Issues Proposed changes

WSGI enfore fails should return 403 instead of 500 

When user is not authorized to perform operations defined in policy
file, it should return a 403 error. The 500 error is incorrect.
This patch do the following changes:
1. Raise a PolicyNotAuthorized 403 exception when normal user
without admin privilege run command 'magnum service-list'.
2. Remove unnecessary hacking rule M301 'decorator must be
the first decorator on a method'.
3. Fix failed enforcement test cases introduced by 403
PolicyNotAuthorized exception.

Change-Id: Ie5a7d138cdb8b226686c189ae86f251c0a1329c8
Closes-Bug: #1520311
This commit is contained in:
houming-wang 2016-01-04 02:51:54 -05:00 committed by Hongbin Lu
parent 2e7a5f8c15
commit 7b754ae390
14 changed files with 120 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
HACKING.rst
View File

@ -8,7 +8,6 @@ Magnum Style Commandments
Magnum Specific Commandments
----------------------------
- [M301] policy.enforce_wsgi decorator must be the first decorator on a method.
- [M310] timeutils.utcnow() wrapper must be used instead of direct calls to
datetime.datetime.utcnow() to make it easy to override its return value.
- [M318] Change assertEqual(A, None) or assertEqual(None, A) by optimal assert

2
magnum/api/controllers/v1/magnum_services.py
View File

@ -86,8 +86,8 @@ class MagnumServiceController(rest.RestController):
super(MagnumServiceController, self).__init__()
self.servicegroup_api = svcgrp_api.ServiceGroup()
@policy.enforce_wsgi("magnum-service", "get_all")
@expose.expose(MagnumServiceCollection)
@policy.enforce_wsgi("magnum-service")
def get_all(self):
"""Retrieve a list of magnum-services.

4
magnum/common/exception.py
View File

@ -332,6 +332,10 @@ class NotAuthorized(MagnumException):
code = 403
class PolicyNotAuthorized(NotAuthorized):
message = _("Policy doesn't allow %(action)s to be performed.")
class NotAcceptable(MagnumException):
# TODO(yuntongjin): We need to set response headers
# in the API for this exception

13
magnum/common/policy.py
View File

@ -20,6 +20,8 @@ from oslo_config import cfg
from oslo_policy import policy
import pecan
from magnum.common import exception
_ENFORCER = None
CONF = cfg.CONF
@ -59,14 +61,14 @@ def init(policy_file=None, rules=None,
return _ENFORCER
def enforce(context, action=None, target=None,
def enforce(context, rule=None, target=None,
do_raise=True, exc=None, *args, **kwargs):
"""Checks authorization of a rule against the target and credentials.
:param dict context: As much information about the user performing the
action as possible.
:param action: The rule to evaluate.
:param rule: The rule to evaluate.
:param dict target: As much information about the object being operated
on as possible.
:param do_raise: Whether to raise an exception or not if check
@ -88,12 +90,10 @@ def enforce(context, action=None, target=None,
if target is None:
target = {'project_id': context.project_id,
'user_id': context.user_id}
return enforcer.enforce(action, target, credentials,
return enforcer.enforce(rule, target, credentials,
do_raise=do_raise, exc=exc, *args, **kwargs)
# NOTE(Shaohe Feng): This decorator MUST appear first (the outermost
# decorator) on an API method for it to work correctly
def enforce_wsgi(api_name, act=None):
"""This is a decorator to simplify wsgi action policy rule check.
@ -114,7 +114,8 @@ def enforce_wsgi(api_name, act=None):
@functools.wraps(fn)
def handle(self, *args, **kwargs):
enforce(pecan.request.context, action, None)
enforce(pecan.request.context, action,
exc=exception.PolicyNotAuthorized, action=action)
return fn(self, *args, **kwargs)
return handle
return wrapper

13
magnum/hacking/checks.py
View File

@ -30,8 +30,6 @@ Guidelines for writing new hacking checks
"""
enforce_re = re.compile(r"@policy.enforce_wsgi*")
decorator_re = re.compile(r"@.*")
mutable_default_args = re.compile(r"^\s*def .+\((.+=\{\}|.+=\[\])")
assert_equal_in_end_with_true_or_false_re = re.compile(
r"assertEqual\((\w|[][.'\"])+ in (\w|[][.'\", ])+, (True|False)\)")
@ -53,16 +51,6 @@ assert_true_isinstance_re = re.compile(
dict_constructor_with_list_copy_re = re.compile(r".*\bdict\((\[)?(\(|\[)")
def check_policy_enforce_decorator(logical_line,
previous_logical, blank_before,
filename):
msg = ("M301: the policy.enforce_wsgi decorator must be the "
"first decorator on a method.")
if (blank_before == 0 and re.match(enforce_re, logical_line)
and re.match(decorator_re, previous_logical)):
yield(0, msg)
def assert_equal_none(logical_line):
"""Check for assertEqual(A, None) or assertEqual(None, A) sentences
@ -146,7 +134,6 @@ def dict_constructor_with_list_copy(logical_line):
def factory(register):
register(check_policy_enforce_decorator)
register(no_mutable_default_args)
register(assert_equal_none)
register(assert_equal_true_or_false)

22
magnum/tests/unit/api/controllers/v1/test_bay.py
View File

@ -14,7 +14,6 @@ import datetime
import mock
from oslo_config import cfg
from oslo_policy import policy
from oslo_utils import timeutils
from six.moves.urllib import parse as urlparse
@ -710,22 +709,24 @@ class TestBayPolicyEnforcement(api_base.FunctionalTest):
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: "project:non_fake"})
exc = self.assertRaises(policy.PolicyNotAuthorized,
exc = self.assertRaises(exception.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith("disallowed by policy"))
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule,
exc.format_message())
def test_policy_disallow_get_all(self):
self._common_policy_check(
"bay:get_all", self.get_json, '/bays')
"bay:get_all", self.get_json, '/bays', expect_errors=True)
def test_policy_disallow_get_one(self):
self._common_policy_check(
"bay:get", self.get_json, '/bays/111-222-333')
"bay:get", self.get_json, '/bays/111-222-333', expect_errors=True)
def test_policy_disallow_detail(self):
self._common_policy_check(
"bay:detail", self.get_json, '/bays/111-222-333/detail')
"bay:detail", self.get_json, '/bays/111-222-333/detail',
expect_errors=True)
def test_policy_disallow_update(self):
self.bay = obj_utils.create_test_bay(self.context,
@ -733,12 +734,13 @@ class TestBayPolicyEnforcement(api_base.FunctionalTest):
node_count=3)
self._common_policy_check(
"bay:update", self.patch_json, '/bays/%s' % self.bay.name,
[{'path': '/name', 'value': "new_name", 'op': 'replace'}])
[{'path': '/name', 'value': "new_name", 'op': 'replace'}],
expect_errors=True)
def test_policy_disallow_create(self):
bdict = apiutils.bay_post_data(name='bay_example_A')
self._common_policy_check(
"bay:create", self.post_json, '/bays', bdict)
"bay:create", self.post_json, '/bays', bdict, expect_errors=True)
def _simulate_rpc_bay_delete(self, bay_uuid):
bay = objects.Bay.get_by_uuid(self.context, bay_uuid)
@ -750,4 +752,4 @@ class TestBayPolicyEnforcement(api_base.FunctionalTest):
self.mock_bay_delete.side_effect = self._simulate_rpc_bay_delete
self.addCleanup(p.stop)
self._common_policy_check(
"bay:delete", self.delete, '/bays/test_bay')
"bay:delete", self.delete, '/bays/test_bay', expect_errors=True)

27
magnum/tests/unit/api/controllers/v1/test_baymodel.py
View File

@ -14,7 +14,6 @@ import datetime
import mock
from oslo_config import cfg
from oslo_policy import policy
from oslo_utils import timeutils
from six.moves.urllib import parse as urlparse
from webtest.app import AppError
@ -653,7 +652,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/baymodels', bdict)
self.assertFalse(response.json['public'])
# policy enforcement is called only once for enforce_wsgi
mock_policy.assert_called_once_with(mock.ANY, mock.ANY, None)
self.assertEqual(1, mock_policy.call_count)
cc_mock.assert_called_once_with(mock.ANY)
self.assertNotIn('id', cc_mock.call_args[0][0])
self.assertFalse(cc_mock.call_args[0][0]['public'])
@ -814,22 +813,26 @@ class TestBayModelPolicyEnforcement(api_base.FunctionalTest):
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: "project:non_fake"})
exc = self.assertRaises(policy.PolicyNotAuthorized,
exc = self.assertRaises(exception.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith("disallowed by policy"))
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule,
exc.format_message())
def test_policy_disallow_get_all(self):
self._common_policy_check(
"baymodel:get_all", self.get_json, '/baymodels')
"baymodel:get_all", self.get_json, '/baymodels',
expect_errors=True)
def test_policy_disallow_get_one(self):
self._common_policy_check(
"baymodel:get", self.get_json, '/baymodels/111-222-333')
"baymodel:get", self.get_json, '/baymodels/111-222-333',
expect_errors=True)
def test_policy_disallow_detail(self):
self._common_policy_check(
"baymodel:detail", self.get_json, '/baymodels/111-222-333/detail')
"baymodel:detail", self.get_json, '/baymodels/111-222-333/detail',
expect_errors=True)
def test_policy_disallow_update(self):
baymodel = obj_utils.create_test_baymodel(self.context,
@ -838,16 +841,18 @@ class TestBayModelPolicyEnforcement(api_base.FunctionalTest):
self._common_policy_check(
"baymodel:update", self.patch_json,
'/baymodels/%s' % baymodel.name,
[{'name': '/name', 'value': "new_name", 'op': 'replace'}])
[{'name': '/name', 'value': "new_name", 'op': 'replace'}],
expect_errors=True)
def test_policy_disallow_create(self):
bdict = apiutils.baymodel_post_data(name='bay_model_example_A')
self._common_policy_check(
"baymodel:create", self.post_json, '/baymodels', bdict)
"baymodel:create", self.post_json, '/baymodels', bdict,
expect_errors=True)
def test_policy_disallow_delete(self):
baymodel = obj_utils.create_test_baymodel(self.context,
uuid='137-246-789')
self._common_policy_check(
"baymodel:delete", self.delete,
'/baymodels/%s' % baymodel.uuid)
'/baymodels/%s' % baymodel.uuid, expect_errors=True)

15
magnum/tests/unit/api/controllers/v1/test_certificate.py
View File

@ -11,9 +11,9 @@
# limitations under the License.
import mock
from oslo_policy import policy
from magnum.api.controllers.v1 import certificate as api_cert
from magnum.common import exception
from magnum.common import utils
from magnum.tests import base
from magnum.tests.unit.api import base as api_base
@ -169,17 +169,20 @@ class TestCertPolicyEnforcement(api_base.FunctionalTest):
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: "project:non_fake"})
exc = self.assertRaises(policy.PolicyNotAuthorized,
exc = self.assertRaises(exception.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith("disallowed by policy"))
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule,
exc.format_message())
def test_policy_disallow_get_one(self):
self._common_policy_check(
"certificate:get", self.get_json,
'/certificates/ce5da569-4f65-4272-9199-fac8c9fbc9d4')
'/certificates/ce5da569-4f65-4272-9199-fac8c9fbc9d4',
expect_errors=True)
def test_policy_disallow_create(self):
cert = apiutils.cert_post_data()
self._common_policy_check(
"certificate:create", self.post_json, '/certificates', cert)
"certificate:create", self.post_json, '/certificates', cert,
expect_errors=True)

28
magnum/tests/unit/api/controllers/v1/test_container.py
View File

@ -10,6 +10,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from magnum.common import exception
from magnum.common import utils as comm_utils
from magnum import objects
from magnum.objects import fields
@ -17,8 +18,6 @@ from magnum.tests.unit.api import base as api_base
from magnum.tests.unit.db import utils
from magnum.tests.unit.objects import utils as obj_utils
from oslo_policy import policy
import mock
from mock import patch
from webtest.app import AppError
@ -637,24 +636,28 @@ class TestContainerEnforcement(api_base.FunctionalTest):
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: 'project:non_fake'})
exc = self.assertRaises(policy.PolicyNotAuthorized,
exc = self.assertRaises(exception.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith('disallowed by policy'))
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule,
exc.format_message())
def test_policy_disallow_get_all(self):
self._common_policy_check(
'container:get_all', self.get_json, '/containers')
'container:get_all', self.get_json, '/containers',
expect_errors=True)
def test_policy_disallow_get_one(self):
self._common_policy_check(
'container:get', self.get_json, '/containers/111-222-333')
'container:get', self.get_json, '/containers/111-222-333',
expect_errors=True)
def test_policy_disallow_detail(self):
self._common_policy_check(
'container:detail',
self.get_json,
'/containers/111-222-333/detail')
'/containers/111-222-333/detail',
expect_errors=True)
def test_policy_disallow_update(self):
test_container = utils.get_test_container()
@ -664,7 +667,8 @@ class TestContainerEnforcement(api_base.FunctionalTest):
'op': 'replace'}]
self._common_policy_check(
'container:update', self.app.patch_json,
'/v1/containers/%s' % container_uuid, params)
'/v1/containers/%s' % container_uuid, params,
expect_errors=True)
def test_policy_disallow_create(self):
params = ('{"name": "' + 'i' * 256 + '", "image": "ubuntu",'
@ -672,9 +676,11 @@ class TestContainerEnforcement(api_base.FunctionalTest):
'"bay_uuid": "fff114da-3bfa-4a0f-a123-c0dffad9718e"}')
self._common_policy_check(
'container:create', self.app.post, '/v1/containers', params)
'container:create', self.app.post, '/v1/containers', params,
expect_errors=True)
def test_policy_disallow_delete(self):
self._common_policy_check(
'container:delete', self.app.delete,
'/v1/containers/%s' % comm_utils.generate_uuid())
'/v1/containers/%s' % comm_utils.generate_uuid(),
expect_errors=True)

24
magnum/tests/unit/api/controllers/v1/test_node.py
View File

@ -14,12 +14,12 @@ import datetime
import mock
from oslo_config import cfg
from oslo_policy import policy
from oslo_utils import timeutils
from six.moves.urllib import parse as urlparse
from wsme import types as wtypes
from magnum.api.controllers.v1 import node as api_node
from magnum.common import exception
from magnum.common import utils
from magnum.tests import base
from magnum.tests.unit.api import base as api_base
@ -295,22 +295,25 @@ class TestNodePolicyEnforcement(api_base.FunctionalTest):
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: "project:non_fake"})
exc = self.assertRaises(policy.PolicyNotAuthorized,
exc = self.assertRaises(exception.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith("disallowed by policy"))
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule,
exc.format_message())
def test_policy_disallow_get_all(self):
self._common_policy_check(
"node:get_all", self.get_json, '/nodes')
"node:get_all", self.get_json, '/nodes', expect_errors=True)
def test_policy_disallow_get_one(self):
self._common_policy_check(
"node:get", self.get_json, '/nodes/111-222-333')
"node:get", self.get_json, '/nodes/111-222-333',
expect_errors=True)
def test_policy_disallow_detail(self):
self._common_policy_check(
"node:detail", self.get_json, '/nodes/111-222-333/detail')
"node:detail", self.get_json, '/nodes/111-222-333/detail',
expect_errors=True)
def test_policy_disallow_update(self):
node = obj_utils.create_test_node(self.context,
@ -319,16 +322,17 @@ class TestNodePolicyEnforcement(api_base.FunctionalTest):
self._common_policy_check(
"node:update", self.patch_json,
'/nodes/%s' % node.uuid,
[{'type': '/type', 'value': "new_type", 'op': 'replace'}])
[{'type': '/type', 'value': "new_type", 'op': 'replace'}],
expect_errors=True)
def test_policy_disallow_create(self):
bdict = apiutils.node_post_data(name='node_example_A')
self._common_policy_check(
"node:create", self.post_json, '/nodes', bdict)
"node:create", self.post_json, '/nodes', bdict, expect_errors=True)
def test_policy_disallow_delete(self):
node = obj_utils.create_test_node(self.context,
uuid='137-246-789')
self._common_policy_check(
"node:delete", self.delete,
'/nodes/%s' % node.uuid)
'/nodes/%s' % node.uuid, expect_errors=True)

23
magnum/tests/unit/api/controllers/v1/test_pod.py
View File

@ -14,11 +14,11 @@ import datetime
import mock
from oslo_config import cfg
from oslo_policy import policy
from six.moves.urllib import parse as urlparse
from wsme import types as wtypes
from magnum.api.controllers.v1 import pod as api_pod
from magnum.common import exception
from magnum.common.pythonk8sclient.swagger_client import rest
from magnum.common import utils
from magnum.conductor import api as rpcapi
@ -576,22 +576,24 @@ class TestPodPolicyEnforcement(api_base.FunctionalTest):
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: 'project:non_fake'})
exc = self.assertRaises(policy.PolicyNotAuthorized,
exc = self.assertRaises(exception.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith('disallowed by policy'))
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule,
exc.format_message())
def test_policy_disallow_get_all(self):
self._common_policy_check(
'pod:get_all', self.get_json, '/pods')
'pod:get_all', self.get_json, '/pods', expect_errors=True)
def test_policy_disallow_get_one(self):
self._common_policy_check(
'pod:get', self.get_json, '/pods/111-222-333')
'pod:get', self.get_json, '/pods/111-222-333', expect_errors=True)
def test_policy_disallow_detail(self):
self._common_policy_check(
'pod:detail', self.get_json, '/pods/111-222-333/detail')
'pod:detail', self.get_json, '/pods/111-222-333/detail',
expect_errors=True)
def test_policy_disallow_update(self):
pod = obj_utils.create_test_pod(self.context,
@ -601,12 +603,13 @@ class TestPodPolicyEnforcement(api_base.FunctionalTest):
self._common_policy_check(
'pod:update', self.patch_json,
'/pods/%s' % pod.uuid,
[{'path': '/desc', 'value': 'new test pod', 'op': 'replace'}])
[{'path': '/desc', 'value': 'new test pod', 'op': 'replace'}],
expect_errors=True)
def test_policy_disallow_create(self):
pdict = apiutils.pod_post_data()
self._common_policy_check(
'pod:create', self.post_json, '/pods', pdict)
'pod:create', self.post_json, '/pods', pdict, expect_errors=True)
def test_policy_disallow_delete(self):
pod = obj_utils.create_test_pod(self.context,
@ -614,4 +617,4 @@ class TestPodPolicyEnforcement(api_base.FunctionalTest):
uuid=utils.generate_uuid())
self._common_policy_check(
'pod:delete', self.delete,
'/pods/%s' % pod.uuid)
'/pods/%s' % pod.uuid, expect_errors=True)

23
magnum/tests/unit/api/controllers/v1/test_replicationcontroller.py
View File

@ -14,11 +14,11 @@ import datetime
import mock
from oslo_config import cfg
from oslo_policy import policy
from six.moves.urllib import parse as urlparse
from wsme import types as wtypes
from magnum.api.controllers.v1 import replicationcontroller as api_rc
from magnum.common import exception
from magnum.common.pythonk8sclient.swagger_client import rest
from magnum.common import utils
from magnum.conductor import api as rpcapi
@ -586,22 +586,24 @@ class TestRCEnforcement(api_base.FunctionalTest):
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: 'project:non_fake'})
exc = self.assertRaises(policy.PolicyNotAuthorized,
exc = self.assertRaises(exception.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith('disallowed by policy'))
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule,
exc.format_message())
def test_policy_disallow_get_all(self):
self._common_policy_check(
'rc:get_all', self.get_json, '/rcs')
'rc:get_all', self.get_json, '/rcs', expect_errors=True)
def test_policy_disallow_get_one(self):
self._common_policy_check(
'rc:get', self.get_json, '/rcs/111-222-333')
'rc:get', self.get_json, '/rcs/111-222-333', expect_errors=True)
def test_policy_disallow_detail(self):
self._common_policy_check(
'rc:detail', self.get_json, '/rcs/111-222-333/detail')
'rc:detail', self.get_json, '/rcs/111-222-333/detail',
expect_errors=True)
def test_policy_disallow_update(self):
rc = obj_utils.create_test_rc(self.context,
@ -612,12 +614,13 @@ class TestRCEnforcement(api_base.FunctionalTest):
self._common_policy_check(
'rc:update', self.patch_json,
'/rcs/%s' % rc.uuid,
[{'path': '/images/0', 'value': new_image, 'op': 'replace'}])
[{'path': '/images/0', 'value': new_image, 'op': 'replace'}],
expect_errors=True)
def test_policy_disallow_create(self):
pdict = apiutils.rc_post_data()
self._common_policy_check(
'rc:create', self.post_json, '/rcs', pdict)
'rc:create', self.post_json, '/rcs', pdict, expect_errors=True)
def test_policy_disallow_delete(self):
rc = obj_utils.create_test_rc(self.context,
@ -625,4 +628,4 @@ class TestRCEnforcement(api_base.FunctionalTest):
uuid=utils.generate_uuid())
self._common_policy_check(
'rc:delete', self.delete,
'/rcs/%s' % rc.uuid)
'/rcs/%s' % rc.uuid, expect_errors=True)

24
magnum/tests/unit/api/controllers/v1/test_service.py
View File

@ -14,11 +14,11 @@ import datetime
import mock
from oslo_config import cfg
from oslo_policy import policy
from six.moves.urllib import parse as urlparse
from wsme import types as wtypes
from magnum.api.controllers.v1 import service as api_service
from magnum.common import exception
from magnum.common.pythonk8sclient.swagger_client import rest
from magnum.common import utils
from magnum.conductor import api as rpcapi
@ -598,22 +598,25 @@ class TestServiceEnforcement(api_base.FunctionalTest):
def _common_policy_check(self, rule, func, *arg, **kwarg):
self.policy.set_rules({rule: 'project:non_fake'})
exc = self.assertRaises(policy.PolicyNotAuthorized,
exc = self.assertRaises(exception.PolicyNotAuthorized,
func, *arg, **kwarg)
self.assertTrue(exc.message.startswith(rule))
self.assertTrue(exc.message.endswith('disallowed by policy'))
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule,
exc.format_message())
def test_policy_disallow_get_all(self):
self._common_policy_check(
'service:get_all', self.get_json, '/services')
'service:get_all', self.get_json, '/services', expect_errors=True)
def test_policy_disallow_get_one(self):
self._common_policy_check(
'service:get', self.get_json, '/services/111-222-333')
'service:get', self.get_json, '/services/111-222-333',
expect_errors=True)
def test_policy_disallow_detail(self):
self._common_policy_check(
'service:detail', self.get_json, '/services/111-222-333/detail')
'service:detail', self.get_json, '/services/111-222-333/detail',
expect_errors=True)
def test_policy_disallow_update(self):
service = obj_utils.create_test_service(self.context,
@ -625,12 +628,13 @@ class TestServiceEnforcement(api_base.FunctionalTest):
'/services/%s' % service.uuid,
[{'path': '/bay_uuid',
'value': utils.generate_uuid(),
'op': 'replace'}])
'op': 'replace'}], expect_errors=True)
def test_policy_disallow_create(self):
pdict = apiutils.service_post_data()
self._common_policy_check(
'service:create', self.post_json, '/services', pdict)
'service:create', self.post_json, '/services', pdict,
expect_errors=True)
def test_policy_disallow_delete(self):
service = obj_utils.create_test_service(self.context,
@ -638,4 +642,4 @@ class TestServiceEnforcement(api_base.FunctionalTest):
uuid=utils.generate_uuid())
self._common_policy_check(
'service:delete', self.delete,
'/services/%s' % service.uuid)
'/services/%s' % service.uuid, expect_errors=True)

10
magnum/tests/unit/test_hacking.py
View File

@ -76,16 +76,6 @@ class HackingTestCase(base.TestCase):
def _assert_has_no_errors(self, code, checker, filename=None):
self._assert_has_errors(code, checker, filename=filename)
def test_policy_enforce_decorator(self):
code = """
@some_other_decorator
@policy.enforce_wsgi("bay", "create")
def my_method():
pass
"""
self._assert_has_errors(code, checks.check_policy_enforce_decorator,
expected_errors=[(2, 0, "M301")])
def test_assert_equal_in(self):
errors = [(1, 0, "M338")]
check = checks.assert_equal_in