Source code for heat_integrationtests.functional.test_remote_stack

#    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 heatclient import exc
import six

from heat_integrationtests.functional import functional_base


[docs]class RemoteStackTest(functional_base.FunctionalTestsBase):
    template = '''
heat_template_version: 2013-05-23
resources:
  my_stack:
    type: OS::Heat::Stack
    properties:
      context:
        region_name: RegionOne
      template:
        get_file: remote_stack.yaml
outputs:
  key:
    value: {get_attr: [my_stack, outputs]}
'''

    remote_template = '''
heat_template_version: 2013-05-23
resources:
  random1:
    type: OS::Heat::RandomString
outputs:
  remote_key:
    value: {get_attr: [random1, value]}
'''

[docs]    def setUp(self):
        super(RemoteStackTest, self).setUp()
        # replacing the template region with the one from the config
        self.template = self.template.replace('RegionOne',
                                              self.conf.region)

[docs]    def test_remote_stack_alone(self):
        stack_id = self.stack_create(template=self.remote_template)
        expected_resources = {'random1': 'OS::Heat::RandomString'}
        self.assertEqual(expected_resources, self.list_resources(stack_id))
        stack = self.client.stacks.get(stack_id)
        output_value = self._stack_output(stack, 'remote_key')
        self.assertEqual(32, len(output_value))

[docs]    def test_stack_create(self):
        files = {'remote_stack.yaml': self.remote_template}
        stack_id = self.stack_create(files=files)

        expected_resources = {'my_stack': 'OS::Heat::Stack'}
        self.assertEqual(expected_resources, self.list_resources(stack_id))

        stack = self.client.stacks.get(stack_id)
        output = self._stack_output(stack, 'key')
        parent_output_value = output['remote_key']
        self.assertEqual(32, len(parent_output_value))

        rsrc = self.client.resources.get(stack_id, 'my_stack')
        remote_id = rsrc.physical_resource_id
        rstack = self.client.stacks.get(remote_id)
        self.assertEqual(remote_id, rstack.id)
        remote_output_value = self._stack_output(rstack, 'remote_key')
        self.assertEqual(32, len(remote_output_value))
        self.assertEqual(parent_output_value, remote_output_value)

        remote_resources = {'random1': 'OS::Heat::RandomString'}
        self.assertEqual(remote_resources, self.list_resources(remote_id))

[docs]    def test_stack_create_bad_region(self):
        tmpl_bad_region = self.template.replace(self.conf.region, 'DARKHOLE')
        files = {'remote_stack.yaml': self.remote_template}
        kwargs = {
            'template': tmpl_bad_region,
            'files': files
        }
        ex = self.assertRaises(exc.HTTPBadRequest, self.stack_create, **kwargs)

        error_msg = ('ERROR: Cannot establish connection to Heat endpoint '
                     'at region "DARKHOLE" due to "publicURL endpoint for '
                     'orchestration service in DARKHOLE region not found"')
        self.assertEqual(error_msg, six.text_type(ex))

[docs]    def test_stack_resource_validation_fail(self):
        tmpl_bad_format = self.remote_template.replace('resources', 'resource')
        files = {'remote_stack.yaml': tmpl_bad_format}
        kwargs = {'files': files}
        ex = self.assertRaises(exc.HTTPBadRequest, self.stack_create, **kwargs)

        error_msg = ('ERROR: Failed validating stack template using Heat '
                     'endpoint at region "%s" due to '
                     '"ERROR: The template section is '
                     'invalid: resource"') % self.conf.region
        self.assertEqual(error_msg, six.text_type(ex))

[docs]    def test_stack_update(self):
        files = {'remote_stack.yaml': self.remote_template}
        stack_id = self.stack_create(files=files)

        expected_resources = {'my_stack': 'OS::Heat::Stack'}
        self.assertEqual(expected_resources, self.list_resources(stack_id))

        rsrc = self.client.resources.get(stack_id, 'my_stack')
        physical_resource_id = rsrc.physical_resource_id
        rstack = self.client.stacks.get(physical_resource_id)
        self.assertEqual(physical_resource_id, rstack.id)

        remote_resources = {'random1': 'OS::Heat::RandomString'}
        self.assertEqual(remote_resources,
                         self.list_resources(rstack.id))
        # do an update
        update_template = self.remote_template.replace('random1', 'random2')
        files = {'remote_stack.yaml': update_template}
        self.update_stack(stack_id, self.template, files=files)

        # check if the remote stack is still there with the same ID
        self.assertEqual(expected_resources, self.list_resources(stack_id))
        rsrc = self.client.resources.get(stack_id, 'my_stack')
        physical_resource_id = rsrc.physical_resource_id
        rstack = self.client.stacks.get(physical_resource_id)
        self.assertEqual(physical_resource_id, rstack.id)

        remote_resources = {'random2': 'OS::Heat::RandomString'}
        self.assertEqual(remote_resources,
                         self.list_resources(rstack.id))

[docs]    def test_stack_suspend_resume(self):
        files = {'remote_stack.yaml': self.remote_template}
        stack_id = self.stack_create(files=files)
        self.stack_suspend(stack_id)
        self.stack_resume(stack_id)
OpenStack

Source code for heat_integrationtests.functional.test_remote_stack

Quick search
