openstack/shade
Code Issues Proposed changes

Replace heatclient testing with requests_mock

Browse Source 
Change-Id: Ib5aa24911e5ffc7426ed5d3e651367cc86e03269
This commit is contained in:
Monty Taylor 2017-03-26 12:18:56 -05:00
parent 12523389a8
commit 510075d494
No known key found for this signature in database
GPG Key ID: 7BAE94BC7141A594
5 changed files with 517 additions and 180 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
shade/tests/fakes.py
View File

@ -17,6 +17,9 @@ fakes
Fakes used for testing Fakes used for testing
""" """
import uuid
from shade._heat import template_format
PROJECT_ID = '1c36b64c840a42cd9e9b931a369337f0' PROJECT_ID = '1c36b64c840a42cd9e9b931a369337f0'
FLAVOR_ID = u'0c1d9008-f546-4608-9e8f-f8bdaec8dddd' FLAVOR_ID = u'0c1d9008-f546-4608-9e8f-f8bdaec8dddd'
@ -24,6 +27,8 @@ CHOCOLATE_FLAVOR_ID = u'0c1d9008-f546-4608-9e8f-f8bdaec8ddde'
STRAWBERRY_FLAVOR_ID = u'0c1d9008-f546-4608-9e8f-f8bdaec8dddf' STRAWBERRY_FLAVOR_ID = u'0c1d9008-f546-4608-9e8f-f8bdaec8dddf'
COMPUTE_ENDPOINT = 'https://compute.example.com/v2.1/{project_id}'.format( COMPUTE_ENDPOINT = 'https://compute.example.com/v2.1/{project_id}'.format(
project_id=PROJECT_ID) project_id=PROJECT_ID)
ORCHESTRATION_ENDPOINT = 'https://orchestration.example.com/v1/{p}'.format(
p=PROJECT_ID)
def make_fake_flavor(flavor_id, name, ram=100, disk=1600, vcpus=24): def make_fake_flavor(flavor_id, name, ram=100, disk=1600, vcpus=24):
@ -54,6 +59,24 @@ FAKE_CHOCOLATE_FLAVOR = make_fake_flavor(
FAKE_STRAWBERRY_FLAVOR = make_fake_flavor( FAKE_STRAWBERRY_FLAVOR = make_fake_flavor(
STRAWBERRY_FLAVOR_ID, 'strawberry', ram=300) STRAWBERRY_FLAVOR_ID, 'strawberry', ram=300)
FAKE_FLAVOR_LIST = [FAKE_FLAVOR, FAKE_CHOCOLATE_FLAVOR, FAKE_STRAWBERRY_FLAVOR] FAKE_FLAVOR_LIST = [FAKE_FLAVOR, FAKE_CHOCOLATE_FLAVOR, FAKE_STRAWBERRY_FLAVOR]
FAKE_TEMPLATE = '''heat_template_version: 2014-10-16
parameters:
length:
type: number
default: 10
resources:
my_rand:
type: OS::Heat::RandomString
properties:
length: {get_param: length}
outputs:
rand:
value:
get_attr: [my_rand, value]
'''
FAKE_TEMPLATE_CONTENT = template_format.parse(FAKE_TEMPLATE)
def make_fake_server(server_id, name, status='ACTIVE'): def make_fake_server(server_id, name, status='ACTIVE'):
@ -105,6 +128,56 @@ def make_fake_server(server_id, name, status='ACTIVE'):
"config_drive": "True"} "config_drive": "True"}
def make_fake_stack(id, name, description=None, status='CREATE_COMPLETE'):
return {
'creation_time': '2017-03-23T23:57:12Z',
'deletion_time': '2017-03-23T23:57:12Z',
'description': description,
'id': id,
'links': [],
'parent': None,
'stack_name': name,
'stack_owner': None,
'stack_status': status,
'stack_user_project_id': PROJECT_ID,
'tags': None,
'updated_time': '2017-03-23T23:57:12Z',
}
def make_fake_stack_event(
id, name, status='CREATE_COMPLETED', resource_name='id'):
event_id = uuid.uuid4().hex
self_url = "{endpoint}/stacks/{name}/{id}/resources/{name}/events/{event}"
resource_url = "{endpoint}/stacks/{name}/{id}/resources/{name}"
return {
"resource_name": id if resource_name == 'id' else name,
"event_time": "2017-03-26T19:38:18",
"links": [
{
"href": self_url.format(
endpoint=ORCHESTRATION_ENDPOINT,
name=name, id=id, event=event_id),
"rel": "self"
}, {
"href": resource_url.format(
endpoint=ORCHESTRATION_ENDPOINT,
name=name, id=id),
"rel": "resource"
}, {
"href": "{endpoint}/stacks/{name}/{id}".format(
endpoint=ORCHESTRATION_ENDPOINT,
name=name, id=id),
"rel": "stack"
}],
"logical_resource_id": name,
"resource_status": status,
"resource_status_reason": "",
"physical_resource_id": id,
"id": event_id,
}
class FakeEndpoint(object): class FakeEndpoint(object):
def __init__(self, id, service_id, region, publicurl, internalurl=None, def __init__(self, id, service_id, region, publicurl, internalurl=None,
adminurl=None): adminurl=None):

5
shade/tests/functional/test_stack.py
View File

@ -20,6 +20,7 @@ Functional tests for `shade` stack methods.
import tempfile import tempfile
from shade import exc from shade import exc
from shade.tests import fakes
from shade.tests.functional import base from shade.tests.functional import base
simple_template = '''heat_template_version: 2014-10-16 simple_template = '''heat_template_version: 2014-10-16
@ -94,7 +95,7 @@ class TestStack(base.BaseFunctionalTestCase):
def test_stack_simple(self): def test_stack_simple(self):
test_template = tempfile.NamedTemporaryFile(delete=False) test_template = tempfile.NamedTemporaryFile(delete=False)
test_template.write(simple_template) test_template.write(fakes.FAKE_TEMPLATE)
test_template.close() test_template.close()
self.stack_name = self.getUniqueString('simple_stack') self.stack_name = self.getUniqueString('simple_stack')
self.addCleanup(self._cleanup_stack) self.addCleanup(self._cleanup_stack)
@ -151,7 +152,7 @@ class TestStack(base.BaseFunctionalTestCase):
test_template.close() test_template.close()
simple_tmpl = tempfile.NamedTemporaryFile(suffix='.yaml', delete=False) simple_tmpl = tempfile.NamedTemporaryFile(suffix='.yaml', delete=False)
simple_tmpl.write(simple_template) simple_tmpl.write(fakes.FAKE_TEMPLATE)
simple_tmpl.close() simple_tmpl.close()
env = tempfile.NamedTemporaryFile(suffix='.yaml', delete=False) env = tempfile.NamedTemporaryFile(suffix='.yaml', delete=False)

7
shade/tests/unit/base.py
View File

@ -556,11 +556,8 @@ class RequestsMockTestCase(BaseTestCase):
if stop_after and x > stop_after: if stop_after and x > stop_after:
break break
self.assertEqual( self.assertEqual(
call['method'], history.method, (call['method'], call['url']), (history.method, history.url),
'Method mismatch on call {index}'.format(index=x)) 'REST mismatch on call {index}'.format(index=x))
self.assertEqual(
call['url'], history.url,
'URL mismatch on call {index}'.format(index=x))
if 'json' in call: if 'json' in call:
self.assertEqual( self.assertEqual(
call['json'], history.json(), call['json'], history.json(),

13
shade/tests/unit/fixtures/catalog-v3.json
View File

@ -109,6 +109,19 @@
"endpoints_links": [], "endpoints_links": [],
"name": "swift", "name": "swift",
"type": "object-store" "type": "object-store"
},
{
"endpoints": [
{
"id": "4deb4d0504a044a395d4480741ba628c",
"interface": "public",
"region": "RegionOne",
"url": "https://orchestration.example.com/v1/1c36b64c840a42cd9e9b931a369337f0"
}
],
"endpoints_links": [],
"name": "heat",
"type": "orchestration"
} }
], ],
"expires_at": "9999-12-31T23:59:59Z", "expires_at": "9999-12-31T23:59:59Z",

599
shade/tests/unit/test_stack.py
View File

@ -11,228 +11,481 @@
# under the License. # under the License.
import mock import tempfile
import testtools import testtools
import shade import shade
from shade._heat import event_utils
from shade._heat import template_utils
from shade import meta from shade import meta
from shade.tests import fakes from shade.tests import fakes
from shade.tests.unit import base from shade.tests.unit import base
class TestStack(base.TestCase): class TestStack(base.RequestsMockTestCase):
@mock.patch.object(shade.OpenStackCloud, 'heat_client') def setUp(self):
def test_list_stacks(self, mock_heat): super(TestStack, self).setUp()
self.stack_id = self.getUniqueString('id')
self.stack_name = self.getUniqueString('name')
self.stack = fakes.make_fake_stack(self.stack_id, self.stack_name)
def test_list_stacks(self):
fake_stacks = [ fake_stacks = [
fakes.FakeStack('001', 'stack1'), self.stack,
fakes.FakeStack('002', 'stack2'), fakes.make_fake_stack(
self.getUniqueString('id'),
self.getUniqueString('name'))
] ]
mock_heat.stacks.list.return_value = fake_stacks self.register_uris([
dict(method='GET',
uri='{endpoint}/stacks'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT),
json={"stacks": fake_stacks}),
])
stacks = self.cloud.list_stacks() stacks = self.cloud.list_stacks()
mock_heat.stacks.list.assert_called_once_with()
self.assertEqual( self.assertEqual(
self.cloud._normalize_stacks(meta.obj_list_to_dict(fake_stacks)), [f.toDict() for f in self.cloud._normalize_stacks(fake_stacks)],
stacks) [f.toDict() for f in stacks])
@mock.patch.object(shade.OpenStackCloud, 'heat_client') self.assert_calls()
def test_list_stacks_exception(self, mock_heat):
mock_heat.stacks.list.side_effect = Exception() def test_list_stacks_exception(self):
self.register_uris([
dict(method='GET',
uri='{endpoint}/stacks'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT),
status_code=404)
])
with testtools.ExpectedException( with testtools.ExpectedException(
shade.OpenStackCloudException, shade.OpenStackCloudException,
"Error fetching stack list" "Error fetching stack list"
): ):
self.cloud.list_stacks() self.cloud.list_stacks()
self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'heat_client') def test_search_stacks(self):
def test_search_stacks(self, mock_heat):
fake_stacks = [ fake_stacks = [
fakes.FakeStack('001', 'stack1'), self.stack,
fakes.FakeStack('002', 'stack2'), fakes.make_fake_stack(
self.getUniqueString('id'),
self.getUniqueString('name'))
] ]
mock_heat.stacks.list.return_value = fake_stacks self.register_uris([
dict(method='GET',
uri='{endpoint}/stacks'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT),
json={"stacks": fake_stacks}),
])
stacks = self.cloud.search_stacks() stacks = self.cloud.search_stacks()
mock_heat.stacks.list.assert_called_once_with()
self.assertEqual( self.assertEqual(
self.cloud._normalize_stacks(meta.obj_list_to_dict(fake_stacks)), self.cloud._normalize_stacks(meta.obj_list_to_dict(fake_stacks)),
stacks) stacks)
self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'heat_client') def test_search_stacks_filters(self):
def test_search_stacks_filters(self, mock_heat):
fake_stacks = [ fake_stacks = [
fakes.FakeStack('001', 'stack1', status='CREATE_COMPLETE'), self.stack,
fakes.FakeStack('002', 'stack2', status='CREATE_FAILED'), fakes.make_fake_stack(
self.getUniqueString('id'),
self.getUniqueString('name'),
status='CREATE_FAILED')
] ]
mock_heat.stacks.list.return_value = fake_stacks self.register_uris([
filters = {'status': 'COMPLETE'} dict(method='GET',
uri='{endpoint}/stacks'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT),
json={"stacks": fake_stacks}),
])
filters = {'status': 'FAILED'}
stacks = self.cloud.search_stacks(filters=filters) stacks = self.cloud.search_stacks(filters=filters)
mock_heat.stacks.list.assert_called_once_with()
self.assertEqual( self.assertEqual(
self.cloud._normalize_stacks( self.cloud._normalize_stacks(
meta.obj_list_to_dict(fake_stacks[:1])), meta.obj_list_to_dict(fake_stacks[1:])),
stacks) stacks)
self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'heat_client') def test_search_stacks_exception(self):
def test_search_stacks_exception(self, mock_heat): self.register_uris([
mock_heat.stacks.list.side_effect = Exception() dict(method='GET',
uri='{endpoint}/stacks'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT),
status_code=404)
])
with testtools.ExpectedException( with testtools.ExpectedException(
shade.OpenStackCloudException, shade.OpenStackCloudException,
"Error fetching stack list" "Error fetching stack list"
): ):
self.cloud.search_stacks() self.cloud.search_stacks()
@mock.patch.object(shade.OpenStackCloud, 'get_stack') def test_delete_stack(self):
@mock.patch.object(shade.OpenStackCloud, 'heat_client') self.register_uris([
def test_delete_stack(self, mock_heat, mock_get): dict(method='GET',
stack = {'id': 'stack_id', 'name': 'stack_name'} uri='{endpoint}/stacks/{name}'.format(
mock_get.return_value = stack endpoint=fakes.ORCHESTRATION_ENDPOINT,
self.assertTrue(self.cloud.delete_stack('stack_name')) name=self.stack_name),
mock_get.assert_called_once_with('stack_name') json={"stack": self.stack}),
mock_heat.stacks.delete.assert_called_once_with(stack['id']) dict(method='GET',
uri='{endpoint}/stacks/{name}/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
dict(method='DELETE',
uri='{endpoint}/stacks/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id)),
])
self.assertTrue(self.cloud.delete_stack(self.stack_name))
self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'get_stack') def test_delete_stack_not_found(self):
@mock.patch.object(shade.OpenStackCloud, 'heat_client') self.register_uris([
def test_delete_stack_not_found(self, mock_heat, mock_get): dict(method='GET',
mock_get.return_value = None uri='{endpoint}/stacks/stack_name'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT),
status_code=404),
])
self.assertFalse(self.cloud.delete_stack('stack_name')) self.assertFalse(self.cloud.delete_stack('stack_name'))
mock_get.assert_called_once_with('stack_name') self.assert_calls()
self.assertFalse(mock_heat.stacks.delete.called)
@mock.patch.object(shade.OpenStackCloud, 'get_stack') def test_delete_stack_exception(self):
@mock.patch.object(shade.OpenStackCloud, 'heat_client') self.register_uris([
def test_delete_stack_exception(self, mock_heat, mock_get): dict(method='GET',
stack = {'id': 'stack_id', 'name': 'stack_name'} uri='{endpoint}/stacks/{id}'.format(
mock_get.return_value = stack endpoint=fakes.ORCHESTRATION_ENDPOINT,
mock_heat.stacks.delete.side_effect = Exception('ouch') id=self.stack_id),
with testtools.ExpectedException( json={"stack": self.stack}),
shade.OpenStackCloudException, dict(method='GET',
"Failed to delete stack stack_name: ouch" uri='{endpoint}/stacks/{name}/{id}'.format(
): endpoint=fakes.ORCHESTRATION_ENDPOINT,
self.cloud.delete_stack('stack_name') id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
dict(method='DELETE',
uri='{endpoint}/stacks/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id),
status_code=400,
reason="ouch"),
])
with testtools.ExpectedException(shade.OpenStackCloudException):
self.cloud.delete_stack(self.stack_id)
self.assert_calls()
@mock.patch.object(event_utils, 'poll_for_events') def test_delete_stack_wait(self):
@mock.patch.object(shade.OpenStackCloud, 'get_stack') marker_event = fakes.make_fake_stack_event(
@mock.patch.object(shade.OpenStackCloud, 'heat_client') self.stack_id, self.stack_name, status='CREATE_COMPLETE')
def test_delete_stack_wait(self, mock_heat, mock_get, mock_poll): marker_qs = 'marker={e_id}&sort_dir=asc'.format(
stack = {'id': 'stack_id', 'name': 'stack_name'} e_id=marker_event['id'])
mock_get.side_effect = (stack, None) self.register_uris([
self.assertTrue(self.cloud.delete_stack('stack_name', wait=True)) dict(method='GET',
mock_heat.stacks.delete.assert_called_once_with(stack['id']) uri='{endpoint}/stacks/{id}'.format(
self.assertEqual(2, mock_get.call_count) endpoint=fakes.ORCHESTRATION_ENDPOINT,
self.assertEqual(1, mock_poll.call_count) id=self.stack_id),
json={"stack": self.stack}),
dict(method='GET',
uri='{endpoint}/stacks/{name}/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
dict(method='GET',
uri='{endpoint}/stacks/{id}/events?{qs}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id,
qs='limit=1&sort_dir=desc'),
complete_qs=True,
json={"events": [marker_event]}),
dict(method='DELETE',
uri='{endpoint}/stacks/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id)),
dict(method='GET',
uri='{endpoint}/stacks/{id}/events?{qs}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id,
qs=marker_qs),
complete_qs=True,
json={"events": [
fakes.make_fake_stack_event(
self.stack_id, self.stack_name,
status='DELETE_COMPLETE'),
]}),
dict(method='GET',
uri='{endpoint}/stacks/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
status_code=404),
])
@mock.patch.object(event_utils, 'poll_for_events') self.assertTrue(self.cloud.delete_stack(self.stack_id, wait=True))
@mock.patch.object(shade.OpenStackCloud, 'get_stack') self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'heat_client')
def test_delete_stack_wait_failed(self, mock_heat, mock_get, mock_poll):
stack = {'id': 'stack_id', 'name': 'stack_name'}
stack_failed = {'id': 'stack_id', 'name': 'stack_name',
'stack_status': 'DELETE_FAILED',
'stack_status_reason': 'ouch'}
mock_get.side_effect = (stack, stack_failed)
with testtools.ExpectedException(
shade.OpenStackCloudException,
"Failed to delete stack stack_name: ouch"
):
self.cloud.delete_stack('stack_name', wait=True)
mock_heat.stacks.delete.assert_called_once_with(stack['id'])
self.assertEqual(2, mock_get.call_count)
self.assertEqual(1, mock_poll.call_count)
@mock.patch.object(template_utils, 'get_template_contents') def test_delete_stack_wait_failed(self):
@mock.patch.object(shade.OpenStackCloud, 'heat_client') failed_stack = self.stack.copy()
def test_create_stack(self, mock_heat, mock_template): failed_stack['stack_status'] = 'DELETE_FAILED'
mock_template.return_value = ({}, {}) marker_event = fakes.make_fake_stack_event(
mock_heat.stacks.create.return_value = fakes.FakeStack('001', 'stack1') self.stack_id, self.stack_name, status='CREATE_COMPLETE')
mock_heat.stacks.get.return_value = fakes.FakeStack('001', 'stack1') marker_qs = 'marker={e_id}&sort_dir=asc'.format(
self.cloud.create_stack('stack_name') e_id=marker_event['id'])
self.assertTrue(mock_template.called) self.register_uris([
mock_heat.stacks.create.assert_called_once_with( dict(method='GET',
stack_name='stack_name', uri='{endpoint}/stacks/{id}'.format(
disable_rollback=False, endpoint=fakes.ORCHESTRATION_ENDPOINT,
environment={}, id=self.stack_id),
parameters={}, json={"stack": self.stack}),
template={}, dict(method='GET',
files={}, uri='{endpoint}/stacks/{name}/{id}'.format(
timeout_mins=60, endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
dict(method='GET',
uri='{endpoint}/stacks/{id}/events?{qs}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id,
qs='limit=1&sort_dir=desc'),
complete_qs=True,
json={"events": [marker_event]}),
dict(method='DELETE',
uri='{endpoint}/stacks/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id)),
dict(method='GET',
uri='{endpoint}/stacks/{id}/events?{qs}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id,
qs=marker_qs),
complete_qs=True,
json={"events": [
fakes.make_fake_stack_event(
self.stack_id, self.stack_name,
status='DELETE_COMPLETE'),
]}),
dict(method='GET',
uri='{endpoint}/stacks/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={'stack': failed_stack}),
dict(method='GET',
uri='{endpoint}/stacks/{name}/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": failed_stack}),
])
with testtools.ExpectedException(shade.OpenStackCloudException):
self.cloud.delete_stack(self.stack_id, wait=True)
self.assert_calls()
def test_create_stack(self):
test_template = tempfile.NamedTemporaryFile(delete=False)
test_template.write(fakes.FAKE_TEMPLATE.encode('utf-8'))
test_template.close()
self.register_uris([
dict(
method='POST', uri='{endpoint}/stacks'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT),
json={"stack": self.stack},
validate=dict(
json={
'disable_rollback': False,
'environment': {},
'files': {},
'parameters': {},
'stack_name': self.stack_name,
'template': fakes.FAKE_TEMPLATE_CONTENT,
'timeout_mins': 60}
)),
dict(
method='GET',
uri='{endpoint}/stacks/{name}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name),
json={"stack": self.stack}),
dict(
method='GET',
uri='{endpoint}/stacks/{name}/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
])
self.cloud.create_stack(
self.stack_name,
template_file=test_template.name
) )
@mock.patch.object(event_utils, 'poll_for_events') self.assert_calls()
@mock.patch.object(template_utils, 'get_template_contents')
@mock.patch.object(shade.OpenStackCloud, 'get_stack')
@mock.patch.object(shade.OpenStackCloud, 'heat_client')
def test_create_stack_wait(self, mock_heat, mock_get, mock_template,
mock_poll):
stack = {'id': 'stack_id', 'name': 'stack_name'}
mock_template.return_value = ({}, {})
mock_get.return_value = stack
mock_heat.stacks.create.return_value = fakes.FakeStack('001', 'stack1')
mock_heat.stacks.get.return_value = fakes.FakeStack('001', 'stack1')
ret = self.cloud.create_stack('stack_name', wait=True)
self.assertTrue(mock_template.called)
mock_heat.stacks.create.assert_called_once_with(
stack_name='stack_name',
disable_rollback=False,
environment={},
parameters={},
template={},
files={},
timeout_mins=60,
)
self.assertEqual(1, mock_get.call_count)
self.assertEqual(1, mock_poll.call_count)
self.assertEqual(stack, ret)
@mock.patch.object(template_utils, 'get_template_contents') def test_create_stack_wait(self):
@mock.patch.object(shade.OpenStackCloud, 'heat_client')
def test_update_stack(self, mock_heat, mock_template):
mock_template.return_value = ({}, {})
mock_heat.stacks.update.return_value = fakes.FakeStack('001', 'stack1')
mock_heat.stacks.get.return_value = fakes.FakeStack('001', 'stack1')
self.cloud.update_stack('stack_name')
self.assertTrue(mock_template.called)
mock_heat.stacks.update.assert_called_once_with(
stack_id='stack_name',
disable_rollback=False,
environment={},
parameters={},
template={},
files={},
timeout_mins=60,
)
@mock.patch.object(event_utils, 'poll_for_events') test_template = tempfile.NamedTemporaryFile(delete=False)
@mock.patch.object(template_utils, 'get_template_contents') test_template.write(fakes.FAKE_TEMPLATE.encode('utf-8'))
@mock.patch.object(shade.OpenStackCloud, 'get_stack') test_template.close()
@mock.patch.object(shade.OpenStackCloud, 'heat_client')
def test_update_stack_wait(self, mock_heat, mock_get, mock_template,
mock_poll):
stack = {'id': 'stack_id', 'name': 'stack_name'}
mock_template.return_value = ({}, {})
mock_get.return_value = stack
ret = self.cloud.update_stack('stack_name', wait=True)
self.assertTrue(mock_template.called)
mock_heat.stacks.update.assert_called_once_with(
stack_id='stack_name',
disable_rollback=False,
environment={},
parameters={},
template={},
files={},
timeout_mins=60,
)
self.assertEqual(1, mock_get.call_count)
self.assertEqual(1, mock_poll.call_count)
self.assertEqual(stack, ret)
@mock.patch.object(shade.OpenStackCloud, 'heat_client') self.register_uris([
def test_get_stack(self, mock_heat): dict(
stack = fakes.FakeStack('azerty', 'stack',) method='POST', uri='{endpoint}/stacks'.format(
mock_heat.stacks.get.return_value = stack endpoint=fakes.ORCHESTRATION_ENDPOINT),
res = self.cloud.get_stack('stack') json={"stack": self.stack},
validate=dict(
json={
'disable_rollback': False,
'environment': {},
'files': {},
'parameters': {},
'stack_name': self.stack_name,
'template': fakes.FAKE_TEMPLATE_CONTENT,
'timeout_mins': 60}
)),
dict(
method='GET',
uri='{endpoint}/stacks/{name}/events?sort_dir=asc'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name),
json={"events": [
fakes.make_fake_stack_event(
self.stack_id, self.stack_name,
status='CREATE_COMPLETE',
resource_name='name'),
]}),
dict(
method='GET',
uri='{endpoint}/stacks/{name}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name),
json={"stack": self.stack}),
dict(
method='GET',
uri='{endpoint}/stacks/{name}/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
])
self.cloud.create_stack(
self.stack_name,
template_file=test_template.name,
wait=True)
self.assert_calls()
def test_update_stack(self):
test_template = tempfile.NamedTemporaryFile(delete=False)
test_template.write(fakes.FAKE_TEMPLATE.encode('utf-8'))
test_template.close()
self.register_uris([
dict(
method='PUT',
uri='{endpoint}/stacks/{name}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name),
validate=dict(
json={
'disable_rollback': False,
'environment': {},
'files': {},
'parameters': {},
'template': fakes.FAKE_TEMPLATE_CONTENT,
'timeout_mins': 60})),
dict(
method='GET',
uri='{endpoint}/stacks/{name}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name),
json={"stack": self.stack}),
dict(
method='GET',
uri='{endpoint}/stacks/{name}/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
])
self.cloud.update_stack(
self.stack_name,
template_file=test_template.name)
self.assert_calls()
def test_update_stack_wait(self):
marker_event = fakes.make_fake_stack_event(
self.stack_id, self.stack_name, status='CREATE_COMPLETE',
resource_name='name')
marker_qs = 'marker={e_id}&sort_dir=asc'.format(
e_id=marker_event['id'])
test_template = tempfile.NamedTemporaryFile(delete=False)
test_template.write(fakes.FAKE_TEMPLATE.encode('utf-8'))
test_template.close()
self.register_uris([
dict(
method='GET',
uri='{endpoint}/stacks/{name}/events?{qs}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name,
qs='limit=1&sort_dir=desc'),
json={"events": [marker_event]}),
dict(
method='PUT',
uri='{endpoint}/stacks/{name}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name),
validate=dict(
json={
'disable_rollback': False,
'environment': {},
'files': {},
'parameters': {},
'template': fakes.FAKE_TEMPLATE_CONTENT,
'timeout_mins': 60})),
dict(
method='GET',
uri='{endpoint}/stacks/{name}/events?{qs}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name,
qs=marker_qs),
json={"events": [
fakes.make_fake_stack_event(
self.stack_id, self.stack_name,
status='UPDATE_COMPLETE',
resource_name='name'),
]}),
dict(
method='GET',
uri='{endpoint}/stacks/{name}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name),
json={"stack": self.stack}),
dict(
method='GET',
uri='{endpoint}/stacks/{name}/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
])
self.cloud.update_stack(
self.stack_name,
template_file=test_template.name,
wait=True)
self.assert_calls()
def test_get_stack(self):
self.register_uris([
dict(method='GET',
uri='{endpoint}/stacks/{name}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
name=self.stack_name),
json={"stack": self.stack}),
dict(method='GET',
uri='{endpoint}/stacks/{name}/{id}'.format(
endpoint=fakes.ORCHESTRATION_ENDPOINT,
id=self.stack_id, name=self.stack_name),
json={"stack": self.stack}),
])
res = self.cloud.get_stack(self.stack_name)
self.assertIsNotNone(res) self.assertIsNotNone(res)
self.assertEqual(stack.stack_name, res['stack_name']) self.assertEqual(self.stack['stack_name'], res['stack_name'])
self.assertEqual(stack.stack_name, res['name']) self.assertEqual(self.stack['stack_name'], res['name'])
self.assertEqual(stack.stack_status, res['stack_status']) self.assertEqual(self.stack['stack_status'], res['stack_status'])
self.assertEqual('COMPLETE', res['status'])
self.assert_calls()