openstack/magnum
Code Issues Proposed changes

Merge "WSGI enfore fails should return 403 instead of 500"

Browse Source
This commit is contained in:
Jenkins 2016-01-07 09:54:31 +00:00 committed by Gerrit Code Review
commit 19d7496a6f
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