Fix missing policy enforcement in CFN API

Currently the CreateStack and UpdateStack actions aren't correctly
enforcing the policy, so add calls to _enforce which fixes that,
and rework the test to illustrate the issue.

Change-Id: Ia5c2e65512ab9d1c163171f8617c081ab96745e5
Closes-Bug: #1256049
This commit is contained in:
Steven Hardy 2013-11-28 16:55:50 +00:00 committed by Jeremy Stanley
parent dc21dc16ed
commit cfd967db3a
2 changed files with 91 additions and 71 deletions

View File

@ -267,9 +267,11 @@ class StackController(object):
)
def create(self, req):
self._enforce(req, 'CreateStack')
return self.create_or_update(req, self.CREATE_STACK)
def update(self, req):
self._enforce(req, 'UpdateStack')
return self.create_or_update(req, self.UPDATE_STACK)
def create_or_update(self, req, action=None):

View File

@ -38,6 +38,28 @@ class CfnStackControllerTest(HeatTestCase):
the endpoint processing API requests after they are routed
'''
def setUp(self):
super(CfnStackControllerTest, self).setUp()
opts = [
cfg.StrOpt('config_dir', default=policy_path),
cfg.StrOpt('config_file', default='foo'),
cfg.StrOpt('project', default='heat'),
]
cfg.CONF.register_opts(opts)
cfg.CONF.set_default('host', 'host')
self.topic = rpc_api.ENGINE_TOPIC
self.api_version = '1.0'
# Create WSGI controller instance
class DummyConfig():
bind_port = 8000
cfgopts = DummyConfig()
self.controller = stacks.StackController(options=cfgopts)
self.controller.policy.enforcer.policy_path = (policy_path +
'deny_stack_user.json')
self.addCleanup(self.m.VerifyAll)
def _dummy_GET_request(self, params={}):
# Mangle the params dict into a query string
qs = "&".join(["=".join([k, str(params[k])]) for k in params])
@ -46,6 +68,16 @@ class CfnStackControllerTest(HeatTestCase):
req.context = utils.dummy_context()
return req
def _stub_enforce(self, req, action, allowed=True):
self.m.StubOutWithMock(policy.Enforcer, 'enforce')
if allowed:
policy.Enforcer.enforce(req.context, action
).AndReturn(True)
else:
policy.Enforcer.enforce(req.context, action
).AndRaise(heat_exception.Forbidden)
self.m.ReplayAll()
# The tests
def test_stackid_addprefix(self):
self.m.ReplayAll()
@ -62,27 +94,21 @@ class CfnStackControllerTest(HeatTestCase):
expected = {'StackName': 'Foo',
'StackId': 'arn:openstack:heat::t:stacks/Foo/123'}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_enforce_default(self):
self.m.ReplayAll()
def test_enforce_ok(self):
params = {'Action': 'ListStacks'}
dummy_req = self._dummy_GET_request(params)
self.controller.policy.policy_path = None
self._stub_enforce(dummy_req, 'ListStacks')
response = self.controller._enforce(dummy_req, 'ListStacks')
self.assertEqual(response, None)
self.m.VerifyAll()
def test_enforce_denied(self):
self.m.ReplayAll()
params = {'Action': 'ListStacks'}
dummy_req = self._dummy_GET_request(params)
dummy_req.context.roles = ['heat_stack_user']
self.controller.policy.policy_path = (policy_path +
'deny_stack_user.json')
self._stub_enforce(dummy_req, 'ListStacks', False)
self.assertRaises(exception.HeatAccessDeniedError,
self.controller._enforce, dummy_req, 'ListStacks')
self.m.VerifyAll()
def test_enforce_ise(self):
params = {'Action': 'ListStacks'}
@ -90,21 +116,19 @@ class CfnStackControllerTest(HeatTestCase):
dummy_req.context.roles = ['heat_stack_user']
self.m.StubOutWithMock(policy.Enforcer, 'enforce')
policy.Enforcer.enforce(dummy_req.context, 'ListStacks', {}
policy.Enforcer.enforce(dummy_req.context, 'ListStacks'
).AndRaise(AttributeError)
self.m.ReplayAll()
self.controller.policy.policy_path = (policy_path +
'deny_stack_user.json')
self.assertRaises(exception.HeatInternalFailureError,
self.controller._enforce, dummy_req, 'ListStacks')
self.m.VerifyAll()
@mock.patch.object(rpc, 'call')
def test_list(self, mock_call):
# Format a dummy GET request to pass into the WSGI handler
params = {'Action': 'ListStacks'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'ListStacks')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = [{u'stack_identity': {u'tenant': u't',
@ -145,6 +169,7 @@ class CfnStackControllerTest(HeatTestCase):
def test_list_rmt_aterr(self, mock_call):
params = {'Action': 'ListStacks'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'ListStacks')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -165,6 +190,7 @@ class CfnStackControllerTest(HeatTestCase):
def test_list_rmt_interr(self, mock_call):
params = {'Action': 'ListStacks'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'ListStacks')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -186,6 +212,7 @@ class CfnStackControllerTest(HeatTestCase):
identity = dict(identifier.HeatIdentifier('t', stack_name, '6'))
params = {'Action': 'DescribeStacks', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStacks')
# Stub out the RPC call to the engine with a pre-canned response
# Note the engine returns a load of keys we don't actually use
@ -267,7 +294,6 @@ class CfnStackControllerTest(HeatTestCase):
'LastUpdatedTime': u'2012-07-09T09:13:11Z'}]}}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_describe_arn(self):
# Format a dummy GET request to pass into the WSGI handler
@ -277,6 +303,7 @@ class CfnStackControllerTest(HeatTestCase):
params = {'Action': 'DescribeStacks',
'StackName': stack_identifier.arn()}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStacks')
# Stub out the RPC call to the engine with a pre-canned response
# Note the engine returns a load of keys we don't actually use
@ -352,7 +379,6 @@ class CfnStackControllerTest(HeatTestCase):
'LastUpdatedTime': u'2012-07-09T09:13:11Z'}]}}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_describe_arn_invalidtenant(self):
# Format a dummy GET request to pass into the WSGI handler
@ -362,6 +388,7 @@ class CfnStackControllerTest(HeatTestCase):
params = {'Action': 'DescribeStacks',
'StackName': stack_identifier.arn()}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStacks')
self.m.StubOutWithMock(rpc, 'call')
rpc.call(dummy_req.context, self.topic,
@ -377,13 +404,13 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.describe(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_describe_aterr(self):
stack_name = "wordpress"
identity = dict(identifier.HeatIdentifier('t', stack_name, '6'))
params = {'Action': 'DescribeStacks', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStacks')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -405,12 +432,12 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.describe(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_describe_bad_name(self):
stack_name = "wibble"
params = {'Action': 'DescribeStacks', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStacks')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -427,7 +454,6 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.describe(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_get_template_int_body(self):
'''Test the internal _get_template function.'''
@ -453,6 +479,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {'timeout_mins': u'30', 'disable_rollback': 'true'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = {u'tenant': u't',
@ -484,7 +511,6 @@ class CfnStackControllerTest(HeatTestCase):
}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_create_rollback(self):
# Format a dummy request
@ -500,6 +526,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {'timeout_mins': u'30', 'disable_rollback': 'false'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = {u'tenant': u't',
@ -531,7 +558,6 @@ class CfnStackControllerTest(HeatTestCase):
}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_create_onfailure_true(self):
# Format a dummy request
@ -547,6 +573,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {'timeout_mins': u'30', 'disable_rollback': 'true'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = {u'tenant': u't',
@ -578,7 +605,6 @@ class CfnStackControllerTest(HeatTestCase):
}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_create_onfailure_false_delete(self):
# Format a dummy request
@ -594,6 +620,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {'timeout_mins': u'30', 'disable_rollback': 'false'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = {u'tenant': u't',
@ -625,7 +652,6 @@ class CfnStackControllerTest(HeatTestCase):
}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_create_onfailure_false_rollback(self):
# Format a dummy request
@ -641,6 +667,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {'timeout_mins': u'30', 'disable_rollback': 'false'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = {u'tenant': u't',
@ -672,7 +699,6 @@ class CfnStackControllerTest(HeatTestCase):
}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_create_onfailure_err(self):
# Format a dummy request
@ -689,6 +715,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {'timeout_mins': u'30', 'disable_rollback': 'false'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
self.assertRaises(exception.HeatInvalidParameterCombinationError,
self.controller.create, dummy_req)
@ -698,6 +725,7 @@ class CfnStackControllerTest(HeatTestCase):
stack_name = "wordpress"
params = {'Action': 'CreateStack', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
result = self.controller.create(dummy_req)
self.assertEqual(type(result), exception.HeatMissingParameterError)
@ -709,6 +737,7 @@ class CfnStackControllerTest(HeatTestCase):
params = {'Action': 'CreateStack', 'StackName': stack_name,
'TemplateBody': '%s' % json_template}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
result = self.controller.create(dummy_req)
self.assertEqual(type(result),
@ -728,10 +757,14 @@ class CfnStackControllerTest(HeatTestCase):
engine_args = {'timeout_mins': u'30'}
dummy_req = self._dummy_GET_request(params)
self.m.StubOutWithMock(policy.Enforcer, 'enforce')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
self.m.StubOutWithMock(rpc, 'call')
policy.Enforcer.enforce(dummy_req.context, 'CreateStack'
).AndReturn(True)
rpc.call(dummy_req.context, self.topic,
{'namespace': None,
'method': 'create_stack',
@ -742,6 +775,9 @@ class CfnStackControllerTest(HeatTestCase):
'args': engine_args},
'version': self.api_version}, None
).AndRaise(AttributeError())
policy.Enforcer.enforce(dummy_req.context, 'CreateStack'
).AndReturn(True)
rpc.call(dummy_req.context, self.topic,
{'namespace': None,
'method': 'create_stack',
@ -752,6 +788,9 @@ class CfnStackControllerTest(HeatTestCase):
'args': engine_args},
'version': self.api_version}, None
).AndRaise(heat_exception.UnknownUserParameter(key='test'))
policy.Enforcer.enforce(dummy_req.context, 'CreateStack'
).AndReturn(True)
rpc.call(dummy_req.context, self.topic,
{'namespace': None,
'method': 'create_stack',
@ -780,8 +819,6 @@ class CfnStackControllerTest(HeatTestCase):
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_create_err_exists(self):
# Format a dummy request
stack_name = "wordpress"
@ -795,6 +832,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {'timeout_mins': u'30'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -817,7 +855,6 @@ class CfnStackControllerTest(HeatTestCase):
self.assertEqual(type(result),
exception.AlreadyExistsError)
self.m.VerifyAll()
def test_create_err_engine(self):
# Format a dummy request
@ -832,6 +869,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {'timeout_mins': u'30'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'CreateStack')
# Stub out the RPC call to the engine with a pre-canned response
self.m.StubOutWithMock(rpc, 'call')
@ -854,7 +892,6 @@ class CfnStackControllerTest(HeatTestCase):
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_update(self):
# Format a dummy request
@ -868,6 +905,7 @@ class CfnStackControllerTest(HeatTestCase):
engine_parms = {u'InstanceType': u'm1.xlarge'}
engine_args = {}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'UpdateStack')
# Stub out the RPC call to the engine with a pre-canned response
identity = dict(identifier.HeatIdentifier('t', stack_name, '1'))
@ -903,7 +941,6 @@ class CfnStackControllerTest(HeatTestCase):
}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_update_bad_name(self):
stack_name = "wibble"
@ -914,6 +951,7 @@ class CfnStackControllerTest(HeatTestCase):
'Parameters.member.1.ParameterKey': 'InstanceType',
'Parameters.member.1.ParameterValue': 'm1.xlarge'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'UpdateStack')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -930,7 +968,6 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.update(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_create_or_update_err(self):
result = self.controller.create_or_update(req={}, action="dsdgfdf")
@ -943,6 +980,7 @@ class CfnStackControllerTest(HeatTestCase):
template = {u'Foo': u'bar'}
params = {'Action': 'GetTemplate', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'GetTemplate')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = template
@ -968,7 +1006,6 @@ class CfnStackControllerTest(HeatTestCase):
{'TemplateBody': template}}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_get_template_err_rpcerr(self):
stack_name = "wordpress"
@ -976,6 +1013,7 @@ class CfnStackControllerTest(HeatTestCase):
template = {u'Foo': u'bar'}
params = {'Action': 'GetTemplate', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'GetTemplate')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -998,12 +1036,12 @@ class CfnStackControllerTest(HeatTestCase):
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_get_template_bad_name(self):
stack_name = "wibble"
params = {'Action': 'GetTemplate', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'GetTemplate')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -1020,7 +1058,6 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.get_template(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_get_template_err_none(self):
stack_name = "wordpress"
@ -1028,6 +1065,7 @@ class CfnStackControllerTest(HeatTestCase):
template = {u'Foo': u'bar'}
params = {'Action': 'GetTemplate', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'GetTemplate')
# Stub out the RPC call to the engine to return None
# this test the "no such stack" error path
@ -1051,13 +1089,13 @@ class CfnStackControllerTest(HeatTestCase):
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_validate_err_no_template(self):
# Format a dummy request with a missing template field
stack_name = "wordpress"
params = {'Action': 'ValidateTemplate'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'ValidateTemplate')
result = self.controller.validate_template(dummy_req)
self.assertEqual(type(result), exception.HeatMissingParameterError)
@ -1068,6 +1106,7 @@ class CfnStackControllerTest(HeatTestCase):
params = {'Action': 'ValidateTemplate',
'TemplateBody': '%s' % json_template}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'ValidateTemplate')
result = self.controller.validate_template(dummy_req)
self.assertEqual(type(result),
@ -1088,6 +1127,7 @@ class CfnStackControllerTest(HeatTestCase):
response = {'Error': 'Resources must contain Resource. '
'Found a [string] instead'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'ValidateTemplate')
# Stub out the RPC call to the engine with a pre-canned response
self.m.StubOutWithMock(rpc, 'call')
@ -1105,7 +1145,6 @@ class CfnStackControllerTest(HeatTestCase):
'Resources must contain Resource. '
'Found a [string] instead'}}
self.assertEqual(expected, response)
self.m.VerifyAll()
def test_delete(self):
# Format a dummy request
@ -1113,6 +1152,7 @@ class CfnStackControllerTest(HeatTestCase):
identity = dict(identifier.HeatIdentifier('t', stack_name, '1'))
params = {'Action': 'DeleteStack', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DeleteStack')
# Stub out the RPC call to the engine with a pre-canned response
self.m.StubOutWithMock(rpc, 'call')
@ -1135,13 +1175,13 @@ class CfnStackControllerTest(HeatTestCase):
expected = {'DeleteStackResponse': {'DeleteStackResult': ''}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_delete_err_rpcerr(self):
stack_name = "wordpress"
identity = dict(identifier.HeatIdentifier('t', stack_name, '1'))
params = {'Action': 'DeleteStack', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DeleteStack')
# Stub out the RPC call to the engine with a pre-canned response
self.m.StubOutWithMock(rpc, 'call')
@ -1166,12 +1206,12 @@ class CfnStackControllerTest(HeatTestCase):
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_delete_bad_name(self):
stack_name = "wibble"
params = {'Action': 'DeleteStack', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DeleteStack')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -1188,7 +1228,6 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.delete(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_events_list_event_id_integer(self):
self._test_events_list('42')
@ -1202,6 +1241,7 @@ class CfnStackControllerTest(HeatTestCase):
identity = dict(identifier.HeatIdentifier('t', stack_name, '6'))
params = {'Action': 'DescribeStackEvents', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackEvents')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = [{u'stack_name': u'wordpress',
@ -1256,13 +1296,13 @@ class CfnStackControllerTest(HeatTestCase):
'LogicalResourceId': u'WikiDatabase'}]}}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_events_list_err_rpcerr(self):
stack_name = "wordpress"
identity = dict(identifier.HeatIdentifier('t', stack_name, '6'))
params = {'Action': 'DescribeStackEvents', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackEvents')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -1284,12 +1324,12 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.events_list(dummy_req)
self.assertEqual(type(result), exception.HeatInternalFailureError)
self.m.VerifyAll()
def test_events_list_bad_name(self):
stack_name = "wibble"
params = {'Action': 'DescribeStackEvents', 'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackEvents')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -1306,7 +1346,6 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.events_list(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_describe_stack_resource(self):
# Format a dummy request
@ -1316,6 +1355,7 @@ class CfnStackControllerTest(HeatTestCase):
'StackName': stack_name,
'LogicalResourceId': "WikiDatabase"}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackResource')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = {u'description': u'',
@ -1376,7 +1416,6 @@ class CfnStackControllerTest(HeatTestCase):
'LogicalResourceId': u'WikiDatabase'}}}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_describe_stack_resource_nonexistent_stack(self):
# Format a dummy request
@ -1386,6 +1425,7 @@ class CfnStackControllerTest(HeatTestCase):
'StackName': stack_name,
'LogicalResourceId': "WikiDatabase"}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackResource')
# Stub out the RPC call to the engine with a pre-canned response
self.m.StubOutWithMock(rpc, 'call')
@ -1401,7 +1441,6 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.describe_stack_resource(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_describe_stack_resource_nonexistent(self):
# Format a dummy request
@ -1411,6 +1450,7 @@ class CfnStackControllerTest(HeatTestCase):
'StackName': stack_name,
'LogicalResourceId': "wibble"}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackResource')
# Stub out the RPC call to the engine with a pre-canned response
self.m.StubOutWithMock(rpc, 'call')
@ -1436,7 +1476,6 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.describe_stack_resource(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_describe_stack_resources(self):
# Format a dummy request
@ -1446,6 +1485,7 @@ class CfnStackControllerTest(HeatTestCase):
'StackName': stack_name,
'LogicalResourceId': "WikiDatabase"}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackResources')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = [{u'description': u'',
@ -1505,7 +1545,6 @@ class CfnStackControllerTest(HeatTestCase):
'LogicalResourceId': u'WikiDatabase'}]}}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_describe_stack_resources_bad_name(self):
stack_name = "wibble"
@ -1513,6 +1552,7 @@ class CfnStackControllerTest(HeatTestCase):
'StackName': stack_name,
'LogicalResourceId': "WikiDatabase"}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackResources')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -1529,7 +1569,6 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.describe_stack_resources(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_describe_stack_resources_physical(self):
# Format a dummy request
@ -1539,6 +1578,7 @@ class CfnStackControllerTest(HeatTestCase):
'LogicalResourceId': "WikiDatabase",
'PhysicalResourceId': 'a3455d8c-9f88-404d-a85b-5315293e67de'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackResources')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = [{u'description': u'',
@ -1599,7 +1639,6 @@ class CfnStackControllerTest(HeatTestCase):
'LogicalResourceId': u'WikiDatabase'}]}}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_describe_stack_resources_physical_not_found(self):
# Format a dummy request
@ -1609,6 +1648,7 @@ class CfnStackControllerTest(HeatTestCase):
'LogicalResourceId': "WikiDatabase",
'PhysicalResourceId': 'aaaaaaaa-9f88-404d-cccc-ffffffffffff'}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackResources')
# Stub out the RPC call to the engine with a pre-canned response
self.m.StubOutWithMock(rpc, 'call')
@ -1627,7 +1667,6 @@ class CfnStackControllerTest(HeatTestCase):
self.assertEqual(type(response),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def test_describe_stack_resources_err_inval(self):
# Format a dummy request containing both StackName and
@ -1638,10 +1677,10 @@ class CfnStackControllerTest(HeatTestCase):
'StackName': stack_name,
'PhysicalResourceId': "123456"}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'DescribeStackResources')
ret = self.controller.describe_stack_resources(dummy_req)
self.assertEqual(type(ret),
exception.HeatInvalidParameterCombinationError)
self.m.VerifyAll()
def test_list_stack_resources(self):
# Format a dummy request
@ -1650,6 +1689,7 @@ class CfnStackControllerTest(HeatTestCase):
params = {'Action': 'ListStackResources',
'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'ListStackResources')
# Stub out the RPC call to the engine with a pre-canned response
engine_resp = [{u'resource_identity':
@ -1698,13 +1738,13 @@ class CfnStackControllerTest(HeatTestCase):
'LogicalResourceId': u'WikiDatabase'}]}}}
self.assertEqual(response, expected)
self.m.VerifyAll()
def test_list_stack_resources_bad_name(self):
stack_name = "wibble"
params = {'Action': 'ListStackResources',
'StackName': stack_name}
dummy_req = self._dummy_GET_request(params)
self._stub_enforce(dummy_req, 'ListStackResources')
# Insert an engine RPC error and ensure we map correctly to the
# heat exception type
@ -1721,25 +1761,3 @@ class CfnStackControllerTest(HeatTestCase):
result = self.controller.list_stack_resources(dummy_req)
self.assertEqual(type(result),
exception.HeatInvalidParameterValueError)
self.m.VerifyAll()
def setUp(self):
super(CfnStackControllerTest, self).setUp()
opts = [
cfg.StrOpt('config_dir', default=policy_path),
cfg.StrOpt('config_file', default='foo'),
cfg.StrOpt('project', default='heat'),
]
cfg.CONF.register_opts(opts)
cfg.CONF.set_default('host', 'host')
self.topic = rpc_api.ENGINE_TOPIC
self.api_version = '1.0'
# Create WSGI controller instance
class DummyConfig():
bind_port = 8000
cfgopts = DummyConfig()
self.controller = stacks.StackController(options=cfgopts)
self.controller.policy.enforcer.policy_path = (policy_path +
'deny_stack_user.json')