openstack
/
heat
Code Issues Proposed changes

Merge "Split OpenStack API V1 unit tests"

This commit is contained in:
Jenkins 2015-08-06 07:58:34 +00:00 committed by Gerrit Code Review
parent adbb85f001 eb6ed3c623
commit cd10238cce
12 changed files with 5009 additions and 4789 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

229
heat/tests/api/openstack_v1/test_actions.py Normal file
View File

@ -0,0 +1,229 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
import mock
import six
import webob.exc
import heat.api.middleware.fault as fault
import heat.api.openstack.v1.actions as actions
from heat.common import identifier
from heat.common import policy
from heat.rpc import client as rpc_client
from heat.tests.api.openstack_v1 import tools
from heat.tests import common
@mock.patch.object(policy.Enforcer, 'enforce')
class ActionControllerTest(tools.ControllerTest, common.HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
'''
def setUp(self):
super(ActionControllerTest, self).setUp()
# Create WSGI controller instance
class DummyConfig(object):
bind_port = 8004
cfgopts = DummyConfig()
self.controller = actions.ActionController(options=cfgopts)
def test_action_suspend(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {'suspend': None}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('stack_suspend', {'stack_identity': stack_identity})
).AndReturn(None)
self.m.ReplayAll()
result = self.controller.action(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.assertIsNone(result)
self.m.VerifyAll()
def test_action_resume(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {'resume': None}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('stack_resume', {'stack_identity': stack_identity})
).AndReturn(None)
self.m.ReplayAll()
result = self.controller.action(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.assertIsNone(result)
self.m.VerifyAll()
def test_action_cancel_update(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {'cancel_update': None}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('stack_cancel_update',
{'stack_identity': stack_identity,
'cancel_with_rollback': True}),
version='1.14'
).AndReturn(None)
self.m.ReplayAll()
result = self.controller.action(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.assertIsNone(result)
self.m.VerifyAll()
def test_action_badaction(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {'notallowed': None}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
self.m.ReplayAll()
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.action,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.m.VerifyAll()
def test_action_badaction_empty(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
self.m.ReplayAll()
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.action,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.m.VerifyAll()
def test_action_badaction_multiple(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {'one': None, 'two': None}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
self.m.ReplayAll()
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.action,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.m.VerifyAll()
def test_action_rmt_aterr(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {'suspend': None}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('stack_suspend', {'stack_identity': stack_identity})
).AndRaise(tools.to_remote_error(AttributeError()))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.action,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.assertEqual(400, resp.json['code'])
self.assertEqual('AttributeError', resp.json['error']['type'])
self.m.VerifyAll()
def test_action_err_denied_policy(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', False)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {'suspend': None}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.action,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.assertEqual(403, resp.status_int)
self.assertIn('403 Forbidden', six.text_type(resp))
def test_action_badaction_ise(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'action', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
body = {'oops': None}
req = self._post(stack_identity._tenant_path() + '/actions',
data=json.dumps(body))
self.m.ReplayAll()
self.controller.ACTIONS = (SUSPEND, NEW) = ('suspend', 'oops')
self.assertRaises(webob.exc.HTTPInternalServerError,
self.controller.action,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
body=body)
self.m.VerifyAll()

4789
heat/tests/api/openstack_v1/test_api_openstack_v1.py
View File

File diff suppressed because it is too large Load Diff

75
heat/tests/api/openstack_v1/test_build_info.py Normal file
View File

@ -0,0 +1,75 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
import six
import heat.api.middleware.fault as fault
import heat.api.openstack.v1.build_info as build_info
from heat.common import policy
from heat.tests.api.openstack_v1 import tools
from heat.tests import common
@mock.patch.object(policy.Enforcer, 'enforce')
class BuildInfoControllerTest(tools.ControllerTest, common.HeatTestCase):
def setUp(self):
super(BuildInfoControllerTest, self).setUp()
self.controller = build_info.BuildInfoController({})
def test_theres_a_default_api_build_revision(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'build_info', True)
req = self._get('/build_info')
self.controller.rpc_client = mock.Mock()
response = self.controller.build_info(req, tenant_id=self.tenant)
self.assertIn('api', response)
self.assertIn('revision', response['api'])
self.assertEqual('unknown', response['api']['revision'])
@mock.patch.object(build_info.cfg, 'CONF')
def test_response_api_build_revision_from_config_file(
self, mock_conf, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'build_info', True)
req = self._get('/build_info')
mock_engine = mock.Mock()
mock_engine.get_revision.return_value = 'engine_revision'
self.controller.rpc_client = mock_engine
mock_conf.revision = {'heat_revision': 'test'}
response = self.controller.build_info(req, tenant_id=self.tenant)
self.assertEqual('test', response['api']['revision'])
def test_retrieves_build_revision_from_the_engine(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'build_info', True)
req = self._get('/build_info')
mock_engine = mock.Mock()
mock_engine.get_revision.return_value = 'engine_revision'
self.controller.rpc_client = mock_engine
response = self.controller.build_info(req, tenant_id=self.tenant)
self.assertIn('engine', response)
self.assertIn('revision', response['engine'])
self.assertEqual('engine_revision', response['engine']['revision'])
def test_build_info_err_denied_policy(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'build_info', False)
req = self._get('/build_info')
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.build_info,
req, tenant_id=self.tenant)
self.assertEqual(403, resp.status_int)
self.assertIn('403 Forbidden', six.text_type(resp))

632
heat/tests/api/openstack_v1/test_events.py Normal file
View File

@ -0,0 +1,632 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
import six
import webob.exc
import heat.api.middleware.fault as fault
import heat.api.openstack.v1.events as events
from heat.common import exception as heat_exc
from heat.common import identifier
from heat.common import policy
from heat.rpc import client as rpc_client
from heat.tests.api.openstack_v1 import tools
from heat.tests import common
@mock.patch.object(policy.Enforcer, 'enforce')
class EventControllerTest(tools.ControllerTest, common.HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
'''
def setUp(self):
super(EventControllerTest, self).setUp()
# Create WSGI controller instance
class DummyConfig(object):
bind_port = 8004
cfgopts = DummyConfig()
self.controller = events.EventController(options=cfgopts)
def test_resource_index_event_id_integer(self, mock_enforce):
self._test_resource_index('42', mock_enforce)
def test_resource_index_event_id_uuid(self, mock_enforce):
self._test_resource_index('a3455d8c-9f88-404d-a85b-5315293e67de',
mock_enforce)
def _test_resource_index(self, event_id, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
ev_identity = identifier.EventIdentifier(event_id=event_id,
**res_identity)
req = self._get(stack_identity._tenant_path() +
'/resources/' + res_name + '/events')
kwargs = {'stack_identity': stack_identity,
'limit': None, 'sort_keys': None, 'marker': None,
'sort_dir': None, 'filters': None}
engine_resp = [
{
u'stack_name': u'wordpress',
u'event_time': u'2012-07-23T13:05:39Z',
u'stack_identity': dict(stack_identity),
u'resource_name': res_name,
u'resource_status_reason': u'state changed',
u'event_identity': dict(ev_identity),
u'resource_action': u'CREATE',
u'resource_status': u'IN_PROGRESS',
u'physical_resource_id': None,
u'resource_properties': {u'UserData': u'blah'},
u'resource_type': u'AWS::EC2::Instance',
},
{
u'stack_name': u'wordpress',
u'event_time': u'2012-07-23T13:05:39Z',
u'stack_identity': dict(stack_identity),
u'resource_name': 'SomeOtherResource',
u'logical_resource_id': 'SomeOtherResource',
u'resource_status_reason': u'state changed',
u'event_identity': dict(ev_identity),
u'resource_action': u'CREATE',
u'resource_status': u'IN_PROGRESS',
u'physical_resource_id': None,
u'resource_properties': {u'UserData': u'blah'},
u'resource_type': u'AWS::EC2::Instance',
}
]
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context, ('list_events', kwargs)
).AndReturn(engine_resp)
self.m.ReplayAll()
result = self.controller.index(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
expected = {
'events': [
{
'id': event_id,
'links': [
{'href': self._url(ev_identity), 'rel': 'self'},
{'href': self._url(res_identity), 'rel': 'resource'},
{'href': self._url(stack_identity), 'rel': 'stack'},
],
u'resource_name': res_name,
u'logical_resource_id': res_name,
u'resource_status_reason': u'state changed',
u'event_time': u'2012-07-23T13:05:39Z',
u'resource_status': u'CREATE_IN_PROGRESS',
u'physical_resource_id': None,
}
]
}
self.assertEqual(expected, result)
self.m.VerifyAll()
def test_stack_index_event_id_integer(self, mock_enforce):
self._test_stack_index('42', mock_enforce)
def test_stack_index_event_id_uuid(self, mock_enforce):
self._test_stack_index('a3455d8c-9f88-404d-a85b-5315293e67de',
mock_enforce)
def _test_stack_index(self, event_id, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
ev_identity = identifier.EventIdentifier(event_id=event_id,
**res_identity)
req = self._get(stack_identity._tenant_path() + '/events')
kwargs = {'stack_identity': stack_identity,
'limit': None, 'sort_keys': None, 'marker': None,
'sort_dir': None, 'filters': None}
engine_resp = [
{
u'stack_name': u'wordpress',
u'event_time': u'2012-07-23T13:05:39Z',
u'stack_identity': dict(stack_identity),
u'resource_name': res_name,
u'resource_status_reason': u'state changed',
u'event_identity': dict(ev_identity),
u'resource_action': u'CREATE',
u'resource_status': u'IN_PROGRESS',
u'physical_resource_id': None,
u'resource_properties': {u'UserData': u'blah'},
u'resource_type': u'AWS::EC2::Instance',
}
]
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('list_events', kwargs)
).AndReturn(engine_resp)
self.m.ReplayAll()
result = self.controller.index(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
expected = {
'events': [
{
'id': event_id,
'links': [
{'href': self._url(ev_identity), 'rel': 'self'},
{'href': self._url(res_identity), 'rel': 'resource'},
{'href': self._url(stack_identity), 'rel': 'stack'},
],
u'resource_name': res_name,
u'logical_resource_id': res_name,
u'resource_status_reason': u'state changed',
u'event_time': u'2012-07-23T13:05:39Z',
u'resource_status': u'CREATE_IN_PROGRESS',
u'physical_resource_id': None,
}
]
}
self.assertEqual(expected, result)
self.m.VerifyAll()
def test_index_stack_nonexist(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wibble', '6')
req = self._get(stack_identity._tenant_path() + '/events')
kwargs = {'stack_identity': stack_identity,
'limit': None, 'sort_keys': None, 'marker': None,
'sort_dir': None, 'filters': None}
error = heat_exc.StackNotFound(stack_name='a')
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('list_events', kwargs)
).AndRaise(tools.to_remote_error(error))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.index,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
self.assertEqual(404, resp.json['code'])
self.assertEqual('StackNotFound', resp.json['error']['type'])
self.m.VerifyAll()
def test_index_err_denied_policy(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', False)
stack_identity = identifier.HeatIdentifier(self.tenant,
'wibble', '6')
req = self._get(stack_identity._tenant_path() + '/events')
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.index,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
self.assertEqual(403, resp.status_int)
self.assertIn('403 Forbidden', six.text_type(resp))
def test_index_resource_nonexist(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
event_id = '42'
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
ev_identity = identifier.EventIdentifier(event_id=event_id,
**res_identity)
req = self._get(stack_identity._tenant_path() +
'/resources/' + res_name + '/events')
kwargs = {'stack_identity': stack_identity,
'limit': None, 'sort_keys': None, 'marker': None,
'sort_dir': None, 'filters': None}
engine_resp = [
{
u'stack_name': u'wordpress',
u'event_time': u'2012-07-23T13:05:39Z',
u'stack_identity': dict(stack_identity),
u'resource_name': 'SomeOtherResource',
u'resource_status_reason': u'state changed',
u'event_identity': dict(ev_identity),
u'resource_action': u'CREATE',
u'resource_status': u'IN_PROGRESS',
u'physical_resource_id': None,
u'resource_properties': {u'UserData': u'blah'},
u'resource_type': u'AWS::EC2::Instance',
}
]
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('list_events', kwargs)
).AndReturn(engine_resp)
self.m.ReplayAll()
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.index,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.m.VerifyAll()
@mock.patch.object(rpc_client.EngineClient, 'call')
def test_index_whitelists_pagination_params(self, mock_call, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
params = {
'limit': 10,
'sort_keys': 'fake sort keys',
'marker': 'fake marker',
'sort_dir': 'fake sort dir',
'balrog': 'you shall not pass!'
}
stack_identity = identifier.HeatIdentifier(self.tenant,
'wibble', '6')
req = self._get(stack_identity._tenant_path() + '/events',
params=params)
mock_call.return_value = []
self.controller.index(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
rpc_call_args, _ = mock_call.call_args
engine_args = rpc_call_args[1][1]
self.assertEqual(6, len(engine_args))
self.assertIn('limit', engine_args)
self.assertEqual(10, engine_args['limit'])
self.assertIn('sort_keys', engine_args)
self.assertEqual(['fake sort keys'], engine_args['sort_keys'])
self.assertIn('marker', engine_args)
self.assertEqual('fake marker', engine_args['marker'])
self.assertIn('sort_dir', engine_args)
self.assertEqual('fake sort dir', engine_args['sort_dir'])
self.assertIn('filters', engine_args)
self.assertIsNone(engine_args['filters'])
self.assertNotIn('balrog', engine_args)
@mock.patch.object(rpc_client.EngineClient, 'call')
def test_index_limit_not_int(self, mock_call, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
sid = identifier.HeatIdentifier(self.tenant, 'wibble', '6')
req = self._get(sid._tenant_path() + '/events',
params={'limit': 'not-an-int'})
ex = self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.index, req,
tenant_id=self.tenant,
stack_name=sid.stack_name,
stack_id=sid.stack_id)
self.assertEqual("Only integer is acceptable by 'limit'.",
six.text_type(ex))
self.assertFalse(mock_call.called)
@mock.patch.object(rpc_client.EngineClient, 'call')
def test_index_whitelist_filter_params(self, mock_call, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
params = {
'resource_status': 'COMPLETE',
'resource_action': 'CREATE',
'resource_name': 'my_server',
'resource_type': 'OS::Nova::Server',
'balrog': 'you shall not pass!'
}
stack_identity = identifier.HeatIdentifier(self.tenant,
'wibble', '6')
req = self._get(stack_identity._tenant_path() + '/events',
params=params)
mock_call.return_value = []
self.controller.index(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
rpc_call_args, _ = mock_call.call_args
engine_args = rpc_call_args[1][1]
self.assertIn('filters', engine_args)
filters = engine_args['filters']
self.assertEqual(4, len(filters))
self.assertIn('resource_status', filters)
self.assertEqual('COMPLETE', filters['resource_status'])
self.assertIn('resource_action', filters)
self.assertEqual('CREATE', filters['resource_action'])
self.assertIn('resource_name', filters)
self.assertEqual('my_server', filters['resource_name'])
self.assertIn('resource_type', filters)
self.assertEqual('OS::Nova::Server', filters['resource_type'])
self.assertNotIn('balrog', filters)
def test_show_event_id_integer(self, mock_enforce):
self._test_show('42', mock_enforce)
def test_show_event_id_uuid(self, mock_enforce):
self._test_show('a3455d8c-9f88-404d-a85b-5315293e67de', mock_enforce)
def _test_show(self, event_id, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
ev1_identity = identifier.EventIdentifier(event_id='41',
**res_identity)
ev_identity = identifier.EventIdentifier(event_id=event_id,
**res_identity)
req = self._get(stack_identity._tenant_path() +
'/resources/' + res_name + '/events/' + event_id)
kwargs = {'stack_identity': stack_identity,
'limit': None, 'sort_keys': None, 'marker': None,
'sort_dir': None, 'filters': None}
engine_resp = [
{
u'stack_name': u'wordpress',
u'event_time': u'2012-07-23T13:05:39Z',
u'stack_identity': dict(stack_identity),
u'resource_name': res_name,
u'resource_status_reason': u'state changed',
u'event_identity': dict(ev1_identity),
u'resource_action': u'CREATE',
u'resource_status': u'IN_PROGRESS',
u'physical_resource_id': None,
u'resource_properties': {u'UserData': u'blah'},
u'resource_type': u'AWS::EC2::Instance',
},
{
u'stack_name': u'wordpress',
u'event_time': u'2012-07-23T13:06:00Z',
u'stack_identity': dict(stack_identity),
u'resource_name': res_name,
u'resource_status_reason': u'state changed',
u'event_identity': dict(ev_identity),
u'resource_action': u'CREATE',
u'resource_status': u'COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_properties': {u'UserData': u'blah'},
u'resource_type': u'AWS::EC2::Instance',
}
]
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('list_events', kwargs)
).AndReturn(engine_resp)
self.m.ReplayAll()
result = self.controller.show(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name,
event_id=event_id)
expected = {
'event': {
'id': event_id,
'links': [
{'href': self._url(ev_identity), 'rel': 'self'},
{'href': self._url(res_identity), 'rel': 'resource'},
{'href': self._url(stack_identity), 'rel': 'stack'},
],
u'resource_name': res_name,
u'logical_resource_id': res_name,
u'resource_status_reason': u'state changed',
u'event_time': u'2012-07-23T13:06:00Z',
u'resource_status': u'CREATE_COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance',
u'resource_properties': {u'UserData': u'blah'},
}
}
self.assertEqual(expected, result)
self.m.VerifyAll()
def test_show_nonexist_event_id_integer(self, mock_enforce):
self._test_show_nonexist('42', '41', mock_enforce)
def test_show_nonexist_event_id_uuid(self, mock_enforce):
self._test_show_nonexist('a3455d8c-9f88-404d-a85b-5315293e67de',
'x3455x8x-9x88-404x-x85x-5315293x67xx',
mock_enforce)
def _test_show_nonexist(self, event_id, search_event_id, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
ev_identity = identifier.EventIdentifier(event_id=search_event_id,
**res_identity)
req = self._get(stack_identity._tenant_path() +
'/resources/' + res_name + '/events/' + event_id)
kwargs = {'stack_identity': stack_identity,
'limit': None, 'sort_keys': None, 'marker': None,
'sort_dir': None, 'filters': None}
engine_resp = [
{
u'stack_name': u'wordpress',
u'event_time': u'2012-07-23T13:05:39Z',
u'stack_identity': dict(stack_identity),
u'resource_name': res_name,
u'resource_status_reason': u'state changed',
u'event_identity': dict(ev_identity),
u'resource_action': u'CREATE',
u'resource_status': u'IN_PROGRESS',
u'physical_resource_id': None,
u'resource_properties': {u'UserData': u'blah'},
u'resource_type': u'AWS::EC2::Instance',
}
]
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context, ('list_events', kwargs)).AndReturn(engine_resp)
self.m.ReplayAll()
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name, event_id=event_id)
self.m.VerifyAll()
def test_show_bad_resource(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
event_id = '42'
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
ev_identity = identifier.EventIdentifier(event_id='41',
**res_identity)
req = self._get(stack_identity._tenant_path() +
'/resources/' + res_name + '/events/' + event_id)
kwargs = {'stack_identity': stack_identity,
'limit': None, 'sort_keys': None, 'marker': None,
'sort_dir': None, 'filters': None}
engine_resp = [
{
u'stack_name': u'wordpress',
u'event_time': u'2012-07-23T13:05:39Z',
u'stack_identity': dict(stack_identity),
u'resource_name': 'SomeOtherResourceName',
u'resource_status_reason': u'state changed',
u'event_identity': dict(ev_identity),
u'resource_action': u'CREATE',
u'resource_status': u'IN_PROGRESS',
u'physical_resource_id': None,
u'resource_properties': {u'UserData': u'blah'},
u'resource_type': u'AWS::EC2::Instance',
}
]
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context, ('list_events', kwargs)).AndReturn(engine_resp)
self.m.ReplayAll()
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name, event_id=event_id)
self.m.VerifyAll()
def test_show_stack_nonexist(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
event_id = '42'
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wibble', '6')
req = self._get(stack_identity._tenant_path() +
'/resources/' + res_name + '/events/' + event_id)
kwargs = {'stack_identity': stack_identity,
'limit': None, 'sort_keys': None, 'marker': None,
'sort_dir': None, 'filters': None}
error = heat_exc.StackNotFound(stack_name='a')
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context, ('list_events', kwargs)
).AndRaise(tools.to_remote_error(error))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.show,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name,
event_id=event_id)
self.assertEqual(404, resp.json['code'])
self.assertEqual('StackNotFound', resp.json['error']['type'])
self.m.VerifyAll()
def test_show_err_denied_policy(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', False)
event_id = '42'
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wibble', '6')
req = self._get(stack_identity._tenant_path() +
'/resources/' + res_name + '/events/' + event_id)
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.show,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name,
event_id=event_id)
self.assertEqual(403, resp.status_int)
self.assertIn('403 Forbidden', six.text_type(resp))

701
heat/tests/api/openstack_v1/test_resources.py Normal file
View File

@ -0,0 +1,701 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
import six
import webob.exc
import heat.api.middleware.fault as fault
import heat.api.openstack.v1.resources as resources
from heat.common import exception as heat_exc
from heat.common import identifier
from heat.common import policy
from heat.rpc import api as rpc_api
from heat.rpc import client as rpc_client
from heat.tests.api.openstack_v1 import tools
from heat.tests import common
@mock.patch.object(policy.Enforcer, 'enforce')
class ResourceControllerTest(tools.ControllerTest, common.HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
'''
def setUp(self):
super(ResourceControllerTest, self).setUp()
# Create WSGI controller instance
class DummyConfig(object):
bind_port = 8004
cfgopts = DummyConfig()
self.controller = resources.ResourceController(options=cfgopts)
def test_index(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(stack_identity._tenant_path() + '/resources')
engine_resp = [
{
u'resource_identity': dict(res_identity),
u'stack_name': stack_identity.stack_name,
u'resource_name': res_name,
u'resource_status_reason': None,
u'updated_time': u'2012-07-23T13:06:00Z',
u'stack_identity': stack_identity,
u'resource_action': u'CREATE',
u'resource_status': u'COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance',
}
]
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('list_stack_resources', {'stack_identity': stack_identity,
'nested_depth': 0,
'with_detail': False,
}),
version='1.12'
).AndReturn(engine_resp)
self.m.ReplayAll()
result = self.controller.index(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
expected = {
'resources': [{'links': [{'href': self._url(res_identity),
'rel': 'self'},
{'href': self._url(stack_identity),
'rel': 'stack'}],
u'resource_name': res_name,
u'logical_resource_id': res_name,
u'resource_status_reason': None,
u'updated_time': u'2012-07-23T13:06:00Z',
u'resource_status': u'CREATE_COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance'}]}
self.assertEqual(expected, result)
self.m.VerifyAll()
def test_index_nonexist(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'rubbish', '1')
req = self._get(stack_identity._tenant_path() + '/resources')
error = heat_exc.StackNotFound(stack_name='a')
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('list_stack_resources', {'stack_identity': stack_identity,
'nested_depth': 0,
'with_detail': False}),
version='1.12'
).AndRaise(tools.to_remote_error(error))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.index,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
self.assertEqual(404, resp.json['code'])
self.assertEqual('StackNotFound', resp.json['error']['type'])
self.m.VerifyAll()
def test_index_nested_depth(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'rubbish', '1')
req = self._get(stack_identity._tenant_path() + '/resources',
{'nested_depth': '99'})
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('list_stack_resources', {'stack_identity': stack_identity,
'nested_depth': 99,
'with_detail': False}),
version='1.12'
).AndReturn([])
self.m.ReplayAll()
result = self.controller.index(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
self.assertEqual([], result['resources'])
self.m.VerifyAll()
def test_index_nested_depth_not_int(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
stack_identity = identifier.HeatIdentifier(self.tenant,
'rubbish', '1')
req = self._get(stack_identity._tenant_path() + '/resources',
{'nested_depth': 'non-int'})
mock_call = self.patchobject(rpc_client.EngineClient, 'call')
ex = self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.index, req,
tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
self.assertEqual("Only integer is acceptable by 'nested_depth'.",
six.text_type(ex))
self.assertFalse(mock_call.called)
def test_index_denied_policy(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', False)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(stack_identity._tenant_path() + '/resources')
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.index,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
self.assertEqual(403, resp.status_int)
self.assertIn('403 Forbidden', six.text_type(resp))
def test_index_detail(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(stack_identity._tenant_path() + '/resources',
{'with_detail': 'true'})
resp_parameters = {
"OS::project_id": "3ab5b02fa01f4f95afa1e254afc4a435",
"network": "cf05086d-07c7-4ed6-95e5-e4af724677e6",
"OS::stack_name": "s1", "admin_pass": "******",
"key_name": "kk", "image": "fa5d387e-541f-4dfb-ae8a-83a614683f84",
"db_port": "50000",
"OS::stack_id": "723d7cee-46b3-4433-9c21-f3378eb0bfc4",
"flavor": "1"
},
engine_resp = [
{
u'resource_identity': dict(res_identity),
u'stack_name': stack_identity.stack_name,
u'resource_name': res_name,
u'resource_status_reason': None,
u'updated_time': u'2012-07-23T13:06:00Z',
u'stack_identity': stack_identity,
u'resource_action': u'CREATE',
u'resource_status': u'COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance',
u'parameters': resp_parameters,
u'description': u'Hello description',
u'stack_user_project_id': u'6f38bcfebbc4400b82d50c1a2ea3057d',
}
]
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('list_stack_resources', {'stack_identity': stack_identity,
'nested_depth': 0,
'with_detail': True}),
version='1.12'
).AndReturn(engine_resp)
self.m.ReplayAll()
result = self.controller.index(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id)
expected = {
'resources': [{'links': [{'href': self._url(res_identity),
'rel': 'self'},
{'href': self._url(stack_identity),
'rel': 'stack'}],
u'resource_name': res_name,
u'logical_resource_id': res_name,
u'resource_status_reason': None,
u'updated_time': u'2012-07-23T13:06:00Z',
u'resource_status': u'CREATE_COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance',
u'parameters': resp_parameters,
u'description': u'Hello description',
u'stack_user_project_id':
u'6f38bcfebbc4400b82d50c1a2ea3057d'}]}
self.assertEqual(expected, result)
self.m.VerifyAll()
def test_show(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(stack_identity._tenant_path())
engine_resp = {
u'description': u'',
u'resource_identity': dict(res_identity),
u'stack_name': stack_identity.stack_name,
u'resource_name': res_name,
u'resource_status_reason': None,
u'updated_time': u'2012-07-23T13:06:00Z',
u'stack_identity': dict(stack_identity),
u'resource_action': u'CREATE',
u'resource_status': u'COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance',
u'metadata': {u'ensureRunning': u'true'}
}
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('describe_stack_resource',
{'stack_identity': stack_identity, 'resource_name': res_name,
'with_attr': None}),
version='1.2'
).AndReturn(engine_resp)
self.m.ReplayAll()
result = self.controller.show(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
expected = {
'resource': {
'links': [
{'href': self._url(res_identity), 'rel': 'self'},
{'href': self._url(stack_identity), 'rel': 'stack'},
],
u'description': u'',
u'resource_name': res_name,
u'logical_resource_id': res_name,
u'resource_status_reason': None,
u'updated_time': u'2012-07-23T13:06:00Z',
u'resource_status': u'CREATE_COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance',
}
}
self.assertEqual(expected, result)
self.m.VerifyAll()
def test_show_with_nested_stack(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
nested_stack_identity = identifier.HeatIdentifier(self.tenant,
'nested', 'some_id')
req = self._get(stack_identity._tenant_path())
engine_resp = {
u'description': u'',
u'resource_identity': dict(res_identity),
u'stack_name': stack_identity.stack_name,
u'resource_name': res_name,
u'resource_status_reason': None,
u'updated_time': u'2012-07-23T13:06:00Z',
u'stack_identity': dict(stack_identity),
u'resource_action': u'CREATE',
u'resource_status': u'COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance',
u'metadata': {u'ensureRunning': u'true'},
u'nested_stack_id': dict(nested_stack_identity)
}
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('describe_stack_resource',
{'stack_identity': stack_identity, 'resource_name': res_name,
'with_attr': None}),
version='1.2'
).AndReturn(engine_resp)
self.m.ReplayAll()
result = self.controller.show(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
expected = [{'href': self._url(res_identity), 'rel': 'self'},
{'href': self._url(stack_identity), 'rel': 'stack'},
{'href': self._url(nested_stack_identity), 'rel': 'nested'}
]
self.assertEqual(expected, result['resource']['links'])
self.assertIsNone(result.get(rpc_api.RES_NESTED_STACK_ID))
self.m.VerifyAll()
def test_show_nonexist(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'rubbish', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(res_identity._tenant_path())
error = heat_exc.StackNotFound(stack_name='a')
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('describe_stack_resource',
{'stack_identity': stack_identity, 'resource_name': res_name,
'with_attr': None}),
version='1.2'
).AndRaise(tools.to_remote_error(error))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.show,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual(404, resp.json['code'])
self.assertEqual('StackNotFound', resp.json['error']['type'])
self.m.VerifyAll()
def test_show_with_single_attribute(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant, 'foo', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
mock_describe = mock.Mock(return_value={'foo': 'bar'})
self.controller.rpc_client.describe_stack_resource = mock_describe
req = self._get(res_identity._tenant_path(), {'with_attr': 'baz'})
resp = self.controller.show(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual({'resource': {'foo': 'bar'}}, resp)
args, kwargs = mock_describe.call_args
self.assertIn('baz', kwargs['with_attr'])
def test_show_with_multiple_attributes(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant, 'foo', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
mock_describe = mock.Mock(return_value={'foo': 'bar'})
self.controller.rpc_client.describe_stack_resource = mock_describe
req = self._get(res_identity._tenant_path())
req.environ['QUERY_STRING'] = 'with_attr=a1&with_attr=a2&with_attr=a3'
resp = self.controller.show(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual({'resource': {'foo': 'bar'}}, resp)
args, kwargs = mock_describe.call_args
self.assertIn('a1', kwargs['with_attr'])
self.assertIn('a2', kwargs['with_attr'])
self.assertIn('a3', kwargs['with_attr'])
def test_show_nonexist_resource(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'Wibble'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(res_identity._tenant_path())
error = heat_exc.ResourceNotFound(stack_name='a', resource_name='b')
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('describe_stack_resource',
{'stack_identity': stack_identity, 'resource_name': res_name,
'with_attr': None}),
version='1.2'
).AndRaise(tools.to_remote_error(error))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.show,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual(404, resp.json['code'])
self.assertEqual('ResourceNotFound', resp.json['error']['type'])
self.m.VerifyAll()
def test_show_uncreated_resource(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(res_identity._tenant_path())
error = heat_exc.ResourceNotAvailable(resource_name='')
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('describe_stack_resource',
{'stack_identity': stack_identity, 'resource_name': res_name,
'with_attr': None}),
version='1.2'
).AndRaise(tools.to_remote_error(error))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.show,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual(404, resp.json['code'])
self.assertEqual('ResourceNotAvailable', resp.json['error']['type'])
self.m.VerifyAll()
def test_show_err_denied_policy(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show', False)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(res_identity._tenant_path())
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.show,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual(403, resp.status_int)
self.assertIn('403 Forbidden', six.text_type(resp))
def test_metadata_show(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'metadata', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(stack_identity._tenant_path())
engine_resp = {
u'description': u'',
u'resource_identity': dict(res_identity),
u'stack_name': stack_identity.stack_name,
u'resource_name': res_name,
u'resource_status_reason': None,
u'updated_time': u'2012-07-23T13:06:00Z',
u'stack_identity': dict(stack_identity),
u'resource_action': u'CREATE',
u'resource_status': u'COMPLETE',
u'physical_resource_id':
u'a3455d8c-9f88-404d-a85b-5315293e67de',
u'resource_type': u'AWS::EC2::Instance',
u'metadata': {u'ensureRunning': u'true'}
}
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('describe_stack_resource',
{'stack_identity': stack_identity, 'resource_name': res_name,
'with_attr': None}),
version='1.2'
).AndReturn(engine_resp)
self.m.ReplayAll()
result = self.controller.metadata(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
expected = {'metadata': {u'ensureRunning': u'true'}}
self.assertEqual(expected, result)
self.m.VerifyAll()
def test_metadata_show_nonexist(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'metadata', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'rubbish', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(res_identity._tenant_path() + '/metadata')
error = heat_exc.StackNotFound(stack_name='a')
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('describe_stack_resource',
{'stack_identity': stack_identity, 'resource_name': res_name,
'with_attr': None}),
version='1.2'
).AndRaise(tools.to_remote_error(error))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.metadata,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual(404, resp.json['code'])
self.assertEqual('StackNotFound', resp.json['error']['type'])
self.m.VerifyAll()
def test_metadata_show_nonexist_resource(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'metadata', True)
res_name = 'wibble'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(res_identity._tenant_path() + '/metadata')
error = heat_exc.ResourceNotFound(stack_name='a', resource_name='b')
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('describe_stack_resource',
{'stack_identity': stack_identity, 'resource_name': res_name,
'with_attr': None}),
version='1.2'
).AndRaise(tools.to_remote_error(error))
self.m.ReplayAll()
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.metadata,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual(404, resp.json['code'])
self.assertEqual('ResourceNotFound', resp.json['error']['type'])
self.m.VerifyAll()
def test_metadata_show_err_denied_policy(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'metadata', False)
res_name = 'wibble'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '1')
res_identity = identifier.ResourceIdentifier(resource_name=res_name,
**stack_identity)
req = self._get(res_identity._tenant_path() + '/metadata')
resp = tools.request_with_middleware(
fault.FaultWrapper,
self.controller.metadata,
req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name)
self.assertEqual(403, resp.status_int)
self.assertIn('403 Forbidden', six.text_type(resp))
def test_signal(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'signal', True)
res_name = 'WikiDatabase'
stack_identity = identifier.HeatIdentifier(self.tenant,
'wordpress', '6')
req = self._get(stack_identity._tenant_path())
self.m.StubOutWithMock(rpc_client.EngineClient, 'call')
rpc_client.EngineClient.call(
req.context,
('resource_signal', {'stack_identity': stack_identity,
'resource_name': res_name,
'details': 'Signal content',
'sync_call': False}),
version='1.3')
self.m.ReplayAll()
result = self.controller.signal(req, tenant_id=self.tenant,
stack_name=stack_identity.stack_name,
stack_id=stack_identity.stack_id,
resource_name=res_name,
body="Signal content")
self.assertIsNone(result)
self.m.VerifyAll()

554
heat/tests/api/openstack_v1/test_routes.py Normal file
View File

@ -0,0 +1,554 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import heat.api.openstack.v1 as api_v1
from heat.tests import common
class RoutesTest(common.HeatTestCase):
def assertRoute(self, mapper, path, method, action, controller,
params=None):
params = params or {}
route = mapper.match(path, {'REQUEST_METHOD': method})
self.assertIsNotNone(route)
self.assertEqual(action, route['action'])
self.assertEqual(
controller, route['controller'].controller.__class__.__name__)
del(route['action'])
del(route['controller'])
self.assertEqual(params, route)
def setUp(self):
super(RoutesTest, self).setUp()
self.m = api_v1.API({}).map
def test_template_handling(self):
self.assertRoute(
self.m,
'/aaaa/resource_types',
'GET',
'list_resource_types',
'StackController',
{
'tenant_id': 'aaaa',
})
self.assertRoute(
self.m,
'/aaaa/resource_types/test_type',
'GET',
'resource_schema',
'StackController',
{
'tenant_id': 'aaaa',
'type_name': 'test_type'
})
self.assertRoute(
self.m,
'/aaaa/resource_types/test_type/template',
'GET',
'generate_template',
'StackController',
{
'tenant_id': 'aaaa',
'type_name': 'test_type'
})
self.assertRoute(
self.m,
'/aaaa/validate',
'POST',
'validate_template',
'StackController',
{
'tenant_id': 'aaaa'
})
def test_stack_collection(self):
self.assertRoute(
self.m,
'/aaaa/stacks',
'GET',
'index',
'StackController',
{
'tenant_id': 'aaaa'
})
self.assertRoute(
self.m,
'/aaaa/stacks',
'POST',
'create',
'StackController',
{
'tenant_id': 'aaaa'
})
self.assertRoute(
self.m,
'/aaaa/stacks/preview',
'POST',
'preview',
'StackController',
{
'tenant_id': 'aaaa'
})
self.assertRoute(
self.m,
'/aaaa/stacks/detail',
'GET',
'detail',
'StackController',
{
'tenant_id': 'aaaa'
})
def test_stack_data(self):
self.assertRoute(
self.m,
'/aaaa/stacks/teststack',
'GET',
'lookup',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack'
})
self.assertRoute(
self.m,
'/aaaa/stacks/arn:openstack:heat::6548ab64fbda49deb188851a3b7d8c8b'
':stacks/stack-1411-06/1c5d9bb2-3464-45e2-a728-26dfa4e1d34a',
'GET',
'lookup',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'arn:openstack:heat:'
':6548ab64fbda49deb188851a3b7d8c8b:stacks/stack-1411-06/'
'1c5d9bb2-3464-45e2-a728-26dfa4e1d34a'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/resources',
'GET',
'lookup',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'path': 'resources'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/events',
'GET',
'lookup',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'path': 'events'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb',
'GET',
'show',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
})
def test_stack_snapshot(self):
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/snapshots',
'POST',
'snapshot',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/snapshots/cccc',
'GET',
'show_snapshot',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
'snapshot_id': 'cccc'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/snapshots/cccc',
'DELETE',
'delete_snapshot',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
'snapshot_id': 'cccc'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/snapshots',
'GET',
'list_snapshots',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/snapshots/cccc/restore',
'POST',
'restore_snapshot',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
'snapshot_id': 'cccc'
})
def test_stack_data_template(self):
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/template',
'GET',
'template',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/template',
'GET',
'lookup',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'path': 'template'
})
def test_stack_post_actions(self):
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/actions',
'POST',
'action',
'ActionController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
})
def test_stack_post_actions_lookup_redirect(self):
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/actions',
'POST',
'lookup',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'path': 'actions'
})
def test_stack_update_delete(self):
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb',
'PUT',
'update',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb',
'DELETE',
'delete',
'StackController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
})
def test_resources(self):
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/resources',
'GET',
'index',
'ResourceController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/resources/cccc',
'GET',
'show',
'ResourceController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
'resource_name': 'cccc'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/resources/cccc/metadata',
'GET',
'metadata',
'ResourceController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
'resource_name': 'cccc'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/resources/cccc/signal',
'POST',
'signal',
'ResourceController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
'resource_name': 'cccc'
})
def test_events(self):
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/events',
'GET',
'index',
'EventController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/resources/cccc/events',
'GET',
'index',
'EventController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
'resource_name': 'cccc'
})
self.assertRoute(
self.m,
'/aaaa/stacks/teststack/bbbb/resources/cccc/events/dddd',
'GET',
'show',
'EventController',
{
'tenant_id': 'aaaa',
'stack_name': 'teststack',
'stack_id': 'bbbb',
'resource_name': 'cccc',
'event_id': 'dddd'
})
def test_software_configs(self):
self.assertRoute(
self.m,
'/aaaa/software_configs',
'GET',
'index',
'SoftwareConfigController',
{
'tenant_id': 'aaaa'
})
self.assertRoute(
self.m,
'/aaaa/software_configs',
'POST',
'create',
'SoftwareConfigController',
{
'tenant_id': 'aaaa'
})
self.assertRoute(
self.m,
'/aaaa/software_configs/bbbb',
'GET',
'show',
'SoftwareConfigController',
{
'tenant_id': 'aaaa',
'config_id': 'bbbb'
})
self.assertRoute(
self.m,
'/aaaa/software_configs/bbbb',
'DELETE',
'delete',
'SoftwareConfigController',
{
'tenant_id': 'aaaa',
'config_id': 'bbbb'
})
def test_software_deployments(self):
self.assertRoute(
self.m,
'/aaaa/software_deployments',
'GET',
'index',
'SoftwareDeploymentController',
{
'tenant_id': 'aaaa'
})
self.assertRoute(
self.m,
'/aaaa/software_deployments',
'POST',
'create',
'SoftwareDeploymentController',
{
'tenant_id': 'aaaa'
})
self.assertRoute(
self.m,
'/aaaa/software_deployments/bbbb',
'GET',
'show',
'SoftwareDeploymentController',
{
'tenant_id': 'aaaa',
'deployment_id': 'bbbb'
})
self.assertRoute(
self.m,
'/aaaa/software_deployments/bbbb',
'PUT',
'update',
'SoftwareDeploymentController',
{
'tenant_id': 'aaaa',
'deployment_id': 'bbbb'
})
self.assertRoute(
self.m,
'/aaaa/software_deployments/bbbb',
'DELETE',
'delete',
'SoftwareDeploymentController',
{
'tenant_id': 'aaaa',
'deployment_id': 'bbbb'
})
def test_build_info(self):
self.assertRoute(
self.m,
'/fake_tenant/build_info',
'GET',
'build_info',
'BuildInfoController',
{'tenant_id': 'fake_tenant'}
)
def test_405(self):
self.assertRoute(
self.m,
'/fake_tenant/validate',
'GET',
'reject',
'DefaultMethodController',
{'tenant_id': 'fake_tenant', 'allowed_methods': 'POST'}
)
self.assertRoute(
self.m,
'/fake_tenant/stacks',
'PUT',
'reject',
'DefaultMethodController',
{'tenant_id': 'fake_tenant', 'allowed_methods': 'GET,POST'}
)
self.assertRoute(
self.m,
'/fake_tenant/stacks/fake_stack/stack_id',
'POST',
'reject',
'DefaultMethodController',
{'tenant_id': 'fake_tenant', 'stack_name': 'fake_stack',
'stack_id': 'stack_id', 'allowed_methods': 'GET,PUT,PATCH,DELETE'}
)
def test_options(self):
self.assertRoute(
self.m,
'/fake_tenant/validate',
'OPTIONS',
'options',
'DefaultMethodController',
{'tenant_id': 'fake_tenant', 'allowed_methods': 'POST'}
)
self.assertRoute(
self.m,
'/fake_tenant/stacks/fake_stack/stack_id',
'OPTIONS',
'options',
'DefaultMethodController',
{'tenant_id': 'fake_tenant', 'stack_name': 'fake_stack',
'stack_id': 'stack_id', 'allowed_methods': 'GET,PUT,PATCH,DELETE'}
)
def test_services(self):
self.assertRoute(
self.m,
'/aaaa/services',
'GET',
'index',
'ServiceController',
{
'tenant_id': 'aaaa'
})

55
heat/tests/api/openstack_v1/test_services.py Normal file
View File

@ -0,0 +1,55 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from oslo_messaging import exceptions
import webob.exc
import heat.api.openstack.v1.services as services
from heat.common import policy
from heat.tests.api.openstack_v1 import tools
from heat.tests import common
class ServiceControllerTest(tools.ControllerTest, common.HeatTestCase):
def setUp(self):
super(ServiceControllerTest, self).setUp()
self.controller = services.ServiceController({})
@mock.patch.object(policy.Enforcer, 'enforce')
def test_index(self, mock_enforce):
self._mock_enforce_setup(
mock_enforce, 'index')
req = self._get('/services')
return_value = []
with mock.patch.object(
self.controller.rpc_client,
'list_services',
return_value=return_value):
resp = self.controller.index(req, tenant_id=self.tenant)
self.assertEqual(
{'services': []}, resp)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_index_503(self, mock_enforce):
self._mock_enforce_setup(
mock_enforce, 'index')
req = self._get('/services')
with mock.patch.object(
self.controller.rpc_client,
'list_services',
side_effect=exceptions.MessagingTimeout()):
self.assertRaises(
webob.exc.HTTPServiceUnavailable,
self.controller.index, req, tenant_id=self.tenant)

160
heat/tests/api/openstack_v1/test_software_configs.py Normal file
View File

@ -0,0 +1,160 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
import mock
import webob.exc
import heat.api.middleware.fault as fault
import heat.api.openstack.v1.software_configs as software_configs
from heat.common import exception as heat_exc
from heat.common import policy
from heat.tests.api.openstack_v1 import tools
from heat.tests import common
class SoftwareConfigControllerTest(tools.ControllerTest, common.HeatTestCase):
def setUp(self):
super(SoftwareConfigControllerTest, self).setUp()
self.controller = software_configs.SoftwareConfigController({})
def test_default(self):
self.assertRaises(
webob.exc.HTTPNotFound, self.controller.default, None)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_index(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'index')
req = self._get('/software_configs')
with mock.patch.object(
self.controller.rpc_client,
'list_software_configs',
return_value=[]):
resp = self.controller.index(req, tenant_id=self.tenant)
self.assertEqual(
{'software_configs': []}, resp)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_show(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show')
config_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._get('/software_configs/%s' % config_id)
return_value = {
'id': config_id,
'name': 'config_mysql',
'group': 'Heat::Shell',
'config': '#!/bin/bash',
'inputs': [],
'ouputs': [],
'options': []}
expected = {'software_config': return_value}
with mock.patch.object(
self.controller.rpc_client,
'show_software_config',
return_value=return_value):
resp = self.controller.show(
req, config_id=config_id, tenant_id=self.tenant)
self.assertEqual(expected, resp)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_show_not_found(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show')
config_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._get('/software_configs/%s' % config_id)
error = heat_exc.NotFound('Not found %s' % config_id)
with mock.patch.object(
self.controller.rpc_client,
'show_software_config',
side_effect=tools.to_remote_error(error)):
resp = tools.request_with_middleware(fault.FaultWrapper,
self.controller.show,
req, config_id=config_id,
tenant_id=self.tenant)
self.assertEqual(404, resp.json['code'])
self.assertEqual('NotFound', resp.json['error']['type'])
@mock.patch.object(policy.Enforcer, 'enforce')
def test_create(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'create')
body = {
'name': 'config_mysql',
'group': 'Heat::Shell',
'config': '#!/bin/bash',
'inputs': [],
'ouputs': [],
'options': []}
return_value = body.copy()
config_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
return_value['id'] = config_id
req = self._post('/software_configs', json.dumps(body))
expected = {'software_config': return_value}
with mock.patch.object(
self.controller.rpc_client,
'create_software_config',
return_value=return_value):
resp = self.controller.create(
req, body=body, tenant_id=self.tenant)
self.assertEqual(expected, resp)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_delete(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'delete')
config_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._delete('/software_configs/%s' % config_id)
return_value = None
with mock.patch.object(
self.controller.rpc_client,
'delete_software_config',
return_value=return_value):
self.assertRaises(
webob.exc.HTTPNoContent, self.controller.delete,
req, config_id=config_id, tenant_id=self.tenant)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_delete_error(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'delete')
config_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._delete('/software_configs/%s' % config_id)
error = Exception('something wrong')
with mock.patch.object(
self.controller.rpc_client,
'delete_software_config',
side_effect=tools.to_remote_error(error)):
resp = tools.request_with_middleware(
fault.FaultWrapper, self.controller.delete,
req, config_id=config_id, tenant_id=self.tenant)
self.assertEqual(500, resp.json['code'])
self.assertEqual('Exception', resp.json['error']['type'])
@mock.patch.object(policy.Enforcer, 'enforce')
def test_delete_not_found(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'delete')
config_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._delete('/software_configs/%s' % config_id)
error = heat_exc.NotFound('Not found %s' % config_id)
with mock.patch.object(
self.controller.rpc_client,
'delete_software_config',
side_effect=tools.to_remote_error(error)):
resp = tools.request_with_middleware(
fault.FaultWrapper, self.controller.delete,
req, config_id=config_id, tenant_id=self.tenant)
self.assertEqual(404, resp.json['code'])
self.assertEqual('NotFound', resp.json['error']['type'])

256
heat/tests/api/openstack_v1/test_software_deployments.py Normal file
View File

@ -0,0 +1,256 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
import mock
import webob.exc
import heat.api.middleware.fault as fault
import heat.api.openstack.v1.software_deployments as software_deployments
from heat.common import exception as heat_exc
from heat.common import policy
from heat.tests.api.openstack_v1 import tools
from heat.tests import common
class SoftwareDeploymentControllerTest(tools.ControllerTest,
common.HeatTestCase):
def setUp(self):
super(SoftwareDeploymentControllerTest, self).setUp()
self.controller = software_deployments.SoftwareDeploymentController({})
def test_default(self):
self.assertRaises(
webob.exc.HTTPNotFound, self.controller.default, None)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_index(self, mock_enforce):
self._mock_enforce_setup(
mock_enforce, 'index', expected_request_count=2)
req = self._get('/software_deployments')
return_value = []
with mock.patch.object(
self.controller.rpc_client,
'list_software_deployments',
return_value=return_value) as mock_call:
resp = self.controller.index(req, tenant_id=self.tenant)
self.assertEqual(
{'software_deployments': []}, resp)
whitelist = mock_call.call_args[1]
self.assertEqual({}, whitelist)
server_id = 'fb322564-7927-473d-8aad-68ae7fbf2abf'
req = self._get('/software_deployments', {'server_id': server_id})
with mock.patch.object(
self.controller.rpc_client,
'list_software_deployments',
return_value=return_value) as mock_call:
resp = self.controller.index(req, tenant_id=self.tenant)
self.assertEqual(
{'software_deployments': []}, resp)
whitelist = mock_call.call_args[1]
self.assertEqual({'server_id': server_id}, whitelist)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_show(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show')
deployment_id = '38eccf10-97e5-4ae8-9d37-b577c9801750'
config_id = 'd00ba4aa-db33-42e1-92f4-2a6469260107'
server_id = 'fb322564-7927-473d-8aad-68ae7fbf2abf'
req = self._get('/software_deployments/%s' % deployment_id)
return_value = {
'id': deployment_id,
'server_id': server_id,
'input_values': {},
'output_values': {},
'action': 'INIT',
'status': 'COMPLETE',
'status_reason': None,
'config_id': config_id,
'config': '#!/bin/bash',
'name': 'config_mysql',
'group': 'Heat::Shell',
'inputs': [],
'outputs': [],
'options': []}
expected = {'software_deployment': return_value}
with mock.patch.object(
self.controller.rpc_client,
'show_software_deployment',
return_value=return_value):
resp = self.controller.show(
req, deployment_id=config_id, tenant_id=self.tenant)
self.assertEqual(expected, resp)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_show_not_found(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'show')
deployment_id = '38eccf10-97e5-4ae8-9d37-b577c9801750'
req = self._get('/software_deployments/%s' % deployment_id)
error = heat_exc.NotFound('Not found %s' % deployment_id)
with mock.patch.object(
self.controller.rpc_client,
'show_software_deployment',
side_effect=tools.to_remote_error(error)):
resp = tools.request_with_middleware(
fault.FaultWrapper, self.controller.show,
req, deployment_id=deployment_id, tenant_id=self.tenant)
self.assertEqual(404, resp.json['code'])
self.assertEqual('NotFound', resp.json['error']['type'])
@mock.patch.object(policy.Enforcer, 'enforce')
def test_create(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'create')
config_id = 'd00ba4aa-db33-42e1-92f4-2a6469260107'
server_id = 'fb322564-7927-473d-8aad-68ae7fbf2abf'
body = {
'server_id': server_id,
'input_values': {},
'action': 'INIT',
'status': 'COMPLETE',
'status_reason': None,
'config_id': config_id}
return_value = body.copy()
deployment_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
return_value['id'] = deployment_id
req = self._post('/software_deployments', json.dumps(body))
expected = {'software_deployment': return_value}
with mock.patch.object(
self.controller.rpc_client,
'create_software_deployment',
return_value=return_value):
resp = self.controller.create(
req, body=body, tenant_id=self.tenant)
self.assertEqual(expected, resp)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_update(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'update')
config_id = 'd00ba4aa-db33-42e1-92f4-2a6469260107'
server_id = 'fb322564-7927-473d-8aad-68ae7fbf2abf'
body = {
'input_values': {},
'action': 'INIT',
'status': 'COMPLETE',
'status_reason': None,
'config_id': config_id}
return_value = body.copy()
deployment_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
return_value['id'] = deployment_id
req = self._put('/software_deployments/%s' % deployment_id,
json.dumps(body))
return_value['server_id'] = server_id
expected = {'software_deployment': return_value}
with mock.patch.object(
self.controller.rpc_client,
'update_software_deployment',
return_value=return_value):
resp = self.controller.update(
req, deployment_id=deployment_id,
body=body, tenant_id=self.tenant)
self.assertEqual(expected, resp)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_update_no_input_values(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'update')
config_id = 'd00ba4aa-db33-42e1-92f4-2a6469260107'
server_id = 'fb322564-7927-473d-8aad-68ae7fbf2abf'
body = {
'action': 'INIT',
'status': 'COMPLETE',
'status_reason': None,
'config_id': config_id}
return_value = body.copy()
deployment_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
return_value['id'] = deployment_id
req = self._put('/software_deployments/%s' % deployment_id,
json.dumps(body))
return_value['server_id'] = server_id
expected = {'software_deployment': return_value}
with mock.patch.object(
self.controller.rpc_client,
'update_software_deployment',
return_value=return_value):
resp = self.controller.update(
req, deployment_id=deployment_id,
body=body, tenant_id=self.tenant)
self.assertEqual(expected, resp)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_update_not_found(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'update')
deployment_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._put('/software_deployments/%s' % deployment_id,
'{}')
error = heat_exc.NotFound('Not found %s' % deployment_id)
with mock.patch.object(
self.controller.rpc_client,
'update_software_deployment',
side_effect=tools.to_remote_error(error)):
resp = tools.request_with_middleware(
fault.FaultWrapper, self.controller.update,
req, deployment_id=deployment_id,
body={}, tenant_id=self.tenant)
self.assertEqual(404, resp.json['code'])
self.assertEqual('NotFound', resp.json['error']['type'])
@mock.patch.object(policy.Enforcer, 'enforce')
def test_delete(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'delete')
deployment_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._delete('/software_deployments/%s' % deployment_id)
return_value = None
with mock.patch.object(
self.controller.rpc_client,
'delete_software_deployment',
return_value=return_value):
self.assertRaises(
webob.exc.HTTPNoContent, self.controller.delete,
req, deployment_id=deployment_id, tenant_id=self.tenant)
@mock.patch.object(policy.Enforcer, 'enforce')
def test_delete_error(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'delete')
deployment_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._delete('/software_deployments/%s' % deployment_id)
error = Exception('something wrong')
with mock.patch.object(
self.controller.rpc_client,
'delete_software_deployment',
side_effect=tools.to_remote_error(error)):
resp = tools.request_with_middleware(
fault.FaultWrapper, self.controller.delete,
req, deployment_id=deployment_id, tenant_id=self.tenant)
self.assertEqual(500, resp.json['code'])
self.assertEqual('Exception', resp.json['error']['type'])
@mock.patch.object(policy.Enforcer, 'enforce')
def test_delete_not_found(self, mock_enforce):
self._mock_enforce_setup(mock_enforce, 'delete')
deployment_id = 'a45559cd-8736-4375-bc39-d6a7bb62ade2'
req = self._delete('/software_deployments/%s' % deployment_id)
error = heat_exc.NotFound('Not Found %s' % deployment_id)
with mock.patch.object(
self.controller.rpc_client,
'delete_software_deployment',
side_effect=tools.to_remote_error(error)):
resp = tools.request_with_middleware(
fault.FaultWrapper, self.controller.delete,
req, deployment_id=deployment_id, tenant_id=self.tenant)
self.assertEqual(404, resp.json['code'])
self.assertEqual('NotFound', resp.json['error']['type'])

2215
heat/tests/api/openstack_v1/test_stacks.py Normal file
View File

File diff suppressed because it is too large Load Diff

0
heat/tests/api/openstack_v1/test_stacks_view.py → heat/tests/api/openstack_v1/test_views_stacks_view.py
View File

132
heat/tests/api/openstack_v1/tools.py Normal file
View File

@ -0,0 +1,132 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_config import cfg
from oslo_log import log
from oslo_messaging._drivers import common as rpc_common
import webob.exc
from heat.common import wsgi
from heat.rpc import api as rpc_api
from heat.tests import utils
def request_with_middleware(middleware, func, req, *args, **kwargs):
@webob.dec.wsgify
def _app(req):
return func(req, *args, **kwargs)
resp = middleware(_app).process_request(req)
return resp
def to_remote_error(error):
"""Converts the given exception to the one with the _Remote suffix.
"""
exc_info = (type(error), error, None)
serialized = rpc_common.serialize_remote_exception(exc_info)
remote_error = rpc_common.deserialize_remote_exception(
serialized, ["heat.common.exception"])
return remote_error
class ControllerTest(object):
"""
Common utilities for testing API Controllers.
"""
def __init__(self, *args, **kwargs):
super(ControllerTest, self).__init__(*args, **kwargs)
cfg.CONF.set_default('host', 'server.test')
self.topic = rpc_api.ENGINE_TOPIC
self.api_version = '1.0'
self.tenant = 't'
self.mock_enforce = None
log.register_options(cfg.CONF)
def _environ(self, path):
return {
'SERVER_NAME': 'server.test',
'SERVER_PORT': 8004,
'SCRIPT_NAME': '/v1',
'PATH_INFO': '/%s' % self.tenant + path,
'wsgi.url_scheme': 'http',
}
def _simple_request(self, path, params=None, method='GET'):
environ = self._environ(path)
environ['REQUEST_METHOD'] = method
if params:
qs = "&".join(["=".join([k, str(params[k])]) for k in params])
environ['QUERY_STRING'] = qs
req = wsgi.Request(environ)
req.context = utils.dummy_context('api_test_user', self.tenant)
self.context = req.context
return req
def _get(self, path, params=None):
return self._simple_request(path, params=params)
def _delete(self, path):
return self._simple_request(path, method='DELETE')
def _abandon(self, path):
return self._simple_request(path, method='DELETE')
def _data_request(self, path, data, content_type='application/json',
method='POST'):
environ = self._environ(path)
environ['REQUEST_METHOD'] = method
req = wsgi.Request(environ)
req.context = utils.dummy_context('api_test_user', self.tenant)
self.context = req.context
req.body = data
return req
def _post(self, path, data, content_type='application/json'):
return self._data_request(path, data, content_type)
def _put(self, path, data, content_type='application/json'):
return self._data_request(path, data, content_type, method='PUT')
def _patch(self, path, data, content_type='application/json'):
return self._data_request(path, data, content_type, method='PATCH')
def _url(self, id):
host = 'server.test:8004'
path = '/v1/%(tenant)s/stacks/%(stack_name)s/%(stack_id)s%(path)s' % id
return 'http://%s%s' % (host, path)
def tearDown(self):
# Common tearDown to assert that policy enforcement happens for all
# controller actions
if self.mock_enforce:
self.mock_enforce.assert_called_with(
action=self.action,
context=self.context,
scope=self.controller.REQUEST_SCOPE)
self.assertEqual(self.expected_request_count,
len(self.mock_enforce.call_args_list))
super(ControllerTest, self).tearDown()
def _mock_enforce_setup(self, mocker, action, allowed=True,
expected_request_count=1):
self.mock_enforce = mocker
self.action = action
self.mock_enforce.return_value = allowed
self.expected_request_count = expected_request_count