Browse Source

Move process template functionality to utils

Change-Id: I6c48d8dd3c5c480eef0001746ce5e243d6820be9
changes/14/706814/4
Rabi Mishra 2 years ago
parent
commit
4d969f0456
  1. 1
      setup.cfg
  2. 13
      tripleo_common/actions/deployment.py
  3. 14
      tripleo_common/actions/package_update.py
  4. 38
      tripleo_common/actions/parameters.py
  5. 15
      tripleo_common/actions/scale.py
  6. 364
      tripleo_common/actions/templates.py
  7. 4
      tripleo_common/tests/actions/test_package_update.py
  8. 420
      tripleo_common/tests/actions/test_templates.py
  9. 460
      tripleo_common/tests/utils/test_template.py
  10. 355
      tripleo_common/utils/template.py
  11. 64
      workbooks/plan_management.yaml

1
setup.cfg

@ -126,7 +126,6 @@ mistral.actions =
tripleo.scale.delete_node = tripleo_common.actions.scale:ScaleDownAction
tripleo.swift.tempurl = tripleo_common.actions.swifthelper:SwiftTempUrlAction
tripleo.swift.swift_information = tripleo_common.actions.swifthelper:SwiftInformationAction
tripleo.templates.process = tripleo_common.actions.templates:ProcessTemplatesAction
tripleo.templates.upload = tripleo_common.actions.templates:UploadTemplatesAction
tripleo.templates.upload_plan_environment = tripleo_common.actions.templates:UploadPlanEnvironmentAction
tripleo.validations.get_pubkey = tripleo_common.actions.validations:GetPubkeyAction

13
tripleo_common/actions/deployment.py

@ -24,12 +24,12 @@ from mistral_lib import actions
from swiftclient import exceptions as swiftexceptions
from tripleo_common.actions import base
from tripleo_common.actions import templates
from tripleo_common import constants
from tripleo_common import update
from tripleo_common.utils import overcloudrc
from tripleo_common.utils import plan as plan_utils
from tripleo_common.utils import swift as swiftutils
from tripleo_common.utils import template as template_utils
LOG = logging.getLogger(__name__)
@ -192,16 +192,9 @@ class DeployStackAction(base.TripleOAction):
return actions.Result(error=err_msg)
# process all plan files and create or update a stack
process_templates_action = templates.ProcessTemplatesAction(
container=self.container
processed_data = template_utils.process_templates(
swift, heat, container=self.container
)
processed_data = process_templates_action.run(context)
# If we receive a 'Result' instance it is because the parent action
# had an error.
if isinstance(processed_data, actions.Result):
return processed_data
stack_args = processed_data.copy()
stack_args['timeout_mins'] = self.timeout_mins

14
tripleo_common/actions/package_update.py

@ -21,9 +21,9 @@ from mistral_lib import actions
from swiftclient import exceptions as swiftexceptions
from tripleo_common.actions import base
from tripleo_common.actions import templates
from tripleo_common import constants
from tripleo_common.utils import plan as plan_utils
from tripleo_common.utils import template as templates
LOG = logging.getLogger(__name__)
@ -38,6 +38,7 @@ class UpdateStackAction(base.TripleOAction):
def run(self, context):
# get the stack. Error if doesn't exist
heat = self.get_orchestration_client(context)
swift = self.get_object_client(context)
try:
stack = heat.stacks.get(self.container)
except heat_exc.HTTPNotFound:
@ -88,16 +89,9 @@ class UpdateStackAction(base.TripleOAction):
return actions.Result(error=err_msg)
# process all plan files and create or update a stack
process_templates_action = templates.ProcessTemplatesAction(
container=self.container
processed_data = templates.process_templates(
swift, heat, container=self.container
)
processed_data = process_templates_action.run(context)
# If we receive a 'Result' instance it is because the parent action
# had an error.
if isinstance(processed_data, actions.Result):
return processed_data
stack_args = processed_data.copy()
LOG.info("Performing Heat stack update")

38
tripleo_common/actions/parameters.py

@ -23,13 +23,13 @@ from mistral_lib import actions
from swiftclient import exceptions as swiftexceptions
from tripleo_common.actions import base
from tripleo_common.actions import templates
from tripleo_common import constants
from tripleo_common import exception
from tripleo_common.utils import nodes
from tripleo_common.utils import parameters as parameter_utils
from tripleo_common.utils import passwords as password_utils
from tripleo_common.utils import plan as plan_utils
from tripleo_common.utils import template as template_utils
LOG = logging.getLogger(__name__)
@ -51,16 +51,9 @@ class GetParametersAction(base.TripleOAction):
if cached is not None:
return cached
process_templates_action = templates.ProcessTemplatesAction(
container=self.container
processed_data = template_utils.process_templates(
swift, heat, container=self.container
)
processed_data = process_templates_action.run(context)
# If we receive a 'Result' instance it is because the parent action
# had an error.
if isinstance(processed_data, actions.Result):
return processed_data
processed_data['show_nested'] = True
# respect previously user set param values
@ -163,15 +156,9 @@ class UpdateParametersAction(base.TripleOAction):
LOG.exception(err_msg)
return actions.Result(error=err_msg)
process_templates_action = templates.ProcessTemplatesAction(
container=self.container
processed_data = template_utils.process_templates(
swift, heat, container=self.container
)
processed_data = process_templates_action.run(context)
# If we receive a 'Result' instance it is because the parent action
# had an error.
if isinstance(processed_data, actions.Result):
return processed_data
env = plan_utils.get_env(swift, self.container)
if not self.validate:
@ -208,7 +195,7 @@ class UpdateParametersAction(base.TripleOAction):
# There has been an error validating we must reprocess the
# templates with the saved working env
plan_utils.put_env(swift, saved_env)
process_templates_action._process_custom_roles(context)
template_utils.process_custom_roles(swift, heat, self.container)
err_msg = ("Error validating environment for plan %s: %s" % (
self.container, err))
@ -668,15 +655,12 @@ class GetNetworkConfigAction(base.TripleOAction):
self.role_name = role_name
def run(self, context):
process_templates_action = templates.ProcessTemplatesAction(
container=self.container
)
processed_data = process_templates_action.run(context)
swift = self.get_object_client(context)
heat = self.get_orchestration_client(context)
# If we receive a 'Result' instance it is because the parent action
# had an error.
if isinstance(processed_data, actions.Result):
return processed_data
processed_data = template_utils.process_templates(
swift, heat, container=self.container
)
# Default temporary value is used when no user input for any
# interface routes for the role networks to find network config.

15
tripleo_common/actions/scale.py

@ -19,9 +19,9 @@ from mistral_lib import actions
from tripleo_common.actions import base
from tripleo_common.actions import parameters as parameters_actions
from tripleo_common.actions import templates
from tripleo_common import constants
from tripleo_common import update
from tripleo_common.utils import template as template_utils
LOG = logging.getLogger(__name__)
@ -66,6 +66,8 @@ class ScaleDownAction(base.TripleOAction):
def _update_stack(self, parameters={},
timeout_mins=constants.STACK_TIMEOUT_DEFAULT,
context=None):
heat = self.get_orchestration_client(context)
swift = self.get_object_client(context)
# TODO(rbrady): migrate _update_stack to it's own action and update
# the workflow for scale down
@ -76,13 +78,9 @@ class ScaleDownAction(base.TripleOAction):
if isinstance(updated_plan, actions.Result):
return updated_plan
process_templates_action = templates.ProcessTemplatesAction(
container=self.container
processed_data = template_utils.process_templates(
swift, heat, container=self.container
)
processed_data = process_templates_action.run(context)
if isinstance(processed_data, actions.Result):
return processed_data
update.add_breakpoints_cleanup_into_env(processed_data['environment'])
fields = processed_data.copy()
@ -95,8 +93,7 @@ class ScaleDownAction(base.TripleOAction):
fields['clear_parameters'] = list(parameters.keys())
LOG.debug('stack update params: %s', fields)
self.get_orchestration_client(context).stacks.update(self.container,
**fields)
heat.stacks.update(self.container, **fields)
def _get_removal_params_from_heat(self, resources_by_role, resources):
stack_params = {}

364
tripleo_common/actions/templates.py

@ -12,59 +12,10 @@
# 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 jinja2
import logging
import os
import six
import yaml
from heatclient import exc as heat_exc
from mistral_lib import actions
from swiftclient import exceptions as swiftexceptions
from tripleo_common.actions import base
from tripleo_common import constants
from tripleo_common.utils import parameters
from tripleo_common.utils import plan as plan_utils
from tripleo_common.utils import swift as swiftutils
LOG = logging.getLogger(__name__)
class J2SwiftLoader(jinja2.BaseLoader):
"""Jinja2 loader to fetch included files from swift
This attempts to fetch a template include file from the given container.
An optional search path or list of search paths can be provided. By default
only the absolute path relative to the container root is searched.
"""
def __init__(self, swift, container, searchpath=None):
self.swift = swift
self.container = container
if searchpath is not None:
if isinstance(searchpath, six.string_types):
self.searchpath = [searchpath]
else:
self.searchpath = list(searchpath)
else:
self.searchpath = []
# Always search the absolute path from the root of the swift container
if '' not in self.searchpath:
self.searchpath.append('')
def get_source(self, environment, template):
pieces = jinja2.loaders.split_template_path(template)
for searchpath in self.searchpath:
template_path = os.path.join(searchpath, *pieces)
try:
source = swiftutils.get_object_string(self.swift,
self.container,
template_path)
return source, None, False
except swiftexceptions.ClientException:
pass
raise jinja2.exceptions.TemplateNotFound(template)
class UploadTemplatesAction(base.UploadDirectoryAction):
@ -87,318 +38,3 @@ class UploadPlanEnvironmentAction(base.TripleOAction):
swift = self.get_object_client(context)
# Push plan environment to the swift container
plan_utils.put_env(swift, self.plan_environment)
class ProcessTemplatesAction(base.TripleOAction):
"""Process Templates and Environments
This method processes the templates and files in a given deployment
plan into a format that can be passed to Heat.
"""
def __init__(self, container=constants.DEFAULT_CONTAINER_NAME):
super(ProcessTemplatesAction, self).__init__()
self.container = container
def _j2_render_and_put(self,
j2_template,
j2_data,
outfile_name=None,
context=None):
swift = self.get_object_client(context)
yaml_f = outfile_name or j2_template.replace('.j2.yaml', '.yaml')
# Search for templates relative to the current template path first
template_base = os.path.dirname(yaml_f)
j2_loader = J2SwiftLoader(swift, self.container, template_base)
try:
# Render the j2 template
template = jinja2.Environment(loader=j2_loader).from_string(
j2_template)
r_template = template.render(**j2_data)
except jinja2.exceptions.TemplateError as ex:
error_msg = ("Error rendering template %s : %s"
% (yaml_f, six.text_type(ex)))
LOG.error(error_msg)
raise Exception(error_msg)
try:
# write the template back to the plan container
LOG.info("Writing rendered template %s" % yaml_f)
swiftutils.put_object_string(swift, self.container, yaml_f,
r_template)
except swiftexceptions.ClientException:
error_msg = ("Error storing file %s in container %s"
% (yaml_f, self.container))
LOG.error(error_msg)
raise Exception(error_msg)
def _get_j2_excludes_file(self, context):
swift = self.get_object_client(context)
try:
j2_excl_file = swiftutils.get_object_string(
swift, self.container, constants.OVERCLOUD_J2_EXCLUDES)
j2_excl_data = yaml.safe_load(j2_excl_file)
if (j2_excl_data is None or j2_excl_data.get('name') is None):
j2_excl_data = {"name": []}
LOG.info("j2_excludes.yaml is either empty or there are "
"no templates to exclude, defaulting the J2 "
"excludes list to: %s" % j2_excl_data)
except swiftexceptions.ClientException:
j2_excl_data = {"name": []}
LOG.info("No J2 exclude file found, defaulting "
"the J2 excludes list to: %s" % j2_excl_data)
return j2_excl_data
def _heat_resource_exists(self, heatclient, stack, nested_stack_name,
resource_name, context):
if stack is None:
LOG.debug("Resource does not exist because stack does not exist")
return False
try:
nested_stack = heatclient.resources.get(stack.id,
nested_stack_name)
except heat_exc.HTTPNotFound:
LOG.debug(
"Resource does not exist because {} stack does "
"not exist".format(nested_stack_name))
return False
try:
heatclient.resources.get(nested_stack.physical_resource_id,
resource_name)
except heat_exc.HTTPNotFound:
LOG.debug("Resource does not exist: {}".format(resource_name))
return False
else:
LOG.debug("Resource exists: {}".format(resource_name))
return True
def _process_custom_roles(self, context):
swift = self.get_object_client(context)
try:
j2_role_file = swiftutils.get_object_string(
swift, self.container, constants.OVERCLOUD_J2_ROLES_NAME)
role_data = yaml.safe_load(j2_role_file)
except swiftexceptions.ClientException:
LOG.info("No %s file found, skipping jinja templating"
% constants.OVERCLOUD_J2_ROLES_NAME)
return
try:
j2_network_file = swiftutils.get_object_string(
swift, self.container, constants.OVERCLOUD_J2_NETWORKS_NAME)
network_data = yaml.safe_load(j2_network_file)
# Allow no networks defined in network_data
if network_data is None:
network_data = []
except swiftexceptions.ClientException:
# Until t-h-t contains network_data.yaml we tolerate a missing file
LOG.warning("No %s file found, ignoring"
% constants.OVERCLOUD_J2_ROLES_NAME)
network_data = []
j2_excl_data = self._get_j2_excludes_file(context)
try:
# Iterate over all files in the plan container
# we j2 render any with the .j2.yaml suffix
container_files = swift.get_container(self.container)
except swiftexceptions.ClientException as ex:
error_msg = ("Error listing contents of container %s : %s"
% (self.container, six.text_type(ex)))
LOG.error(error_msg)
raise Exception(error_msg)
role_names = [r.get('name') for r in role_data]
r_map = {}
for r in role_data:
r_map[r.get('name')] = r
excl_templates = j2_excl_data.get('name')
heatclient = self.get_orchestration_client(context)
stack = None
try:
stack = heatclient.stacks.get(self.container,
resolve_outputs=False)
except heat_exc.HTTPNotFound:
LOG.debug("Stack does not exist")
n_map = {}
for n in network_data:
if n.get('enabled') is not False:
n_map[n.get('name')] = n
if not n.get('name_lower'):
n_map[n.get('name')]['name_lower'] = n.get('name').lower()
if n.get('name') == constants.API_NETWORK and 'compat_name' \
not in n.keys():
# Check to see if legacy named API network exists
# and if so we need to set compat_name
api_net = "{}Network".format(constants.LEGACY_API_NETWORK)
if self._heat_resource_exists(heatclient, stack, 'Networks',
api_net, context):
n['compat_name'] = 'Internal'
LOG.info("Upgrade compatibility enabled for legacy "
"network resource Internal.")
else:
LOG.info("skipping %s network: network is disabled." %
n.get('name'))
plan_utils.cache_delete(swift, self.container,
"tripleo.parameters.get")
for f in [f.get('name') for f in container_files[1]]:
# We do three templating passes here:
# 1. *.role.j2.yaml - we template just the role name
# and create multiple files (one per role)
# 2 *.network.j2.yaml - we template the network name and
# data and create multiple files for networks and
# network ports (one per network)
# 3. *.j2.yaml - we template with all roles_data,
# and create one file common to all roles
if f.endswith('.role.j2.yaml'):
LOG.info("jinja2 rendering role template %s" % f)
j2_template = swiftutils.get_object_string(swift,
self.container,
f)
LOG.info("jinja2 rendering roles %s" % ","
.join(role_names))
for role in role_names:
LOG.info("jinja2 rendering role %s" % role)
out_f = "-".join(
[role.lower(),
os.path.basename(f).replace('.role.j2.yaml',
'.yaml')])
out_f_path = os.path.join(os.path.dirname(f), out_f)
if ('network/config' in os.path.dirname(f) and
r_map[role].get('deprecated_nic_config_name')):
d_name = r_map[role].get('deprecated_nic_config_name')
out_f_path = os.path.join(os.path.dirname(f), d_name)
elif ('network/config' in os.path.dirname(f)):
d_name = "%s.yaml" % role.lower()
out_f_path = os.path.join(os.path.dirname(f), d_name)
if not (out_f_path in excl_templates):
if '{{role.name}}' in j2_template:
j2_data = {'role': r_map[role],
'networks': network_data}
self._j2_render_and_put(j2_template,
j2_data,
out_f_path,
context=context)
else:
# Backwards compatibility with templates
# that specify {{role}} vs {{role.name}}
j2_data = {'role': role, 'networks': network_data}
LOG.debug("role legacy path for role %s" % role)
self._j2_render_and_put(j2_template,
j2_data,
out_f_path,
context=context)
else:
LOG.info("Skipping rendering of %s, defined in %s" %
(out_f_path, j2_excl_data))
elif (f.endswith('.network.j2.yaml')):
LOG.info("jinja2 rendering network template %s" % f)
j2_template = swiftutils.get_object_string(swift,
self.container,
f)
LOG.info("jinja2 rendering networks %s" % ",".join(n_map))
for network in n_map:
j2_data = {'network': n_map[network]}
# Output file names in "<name>.yaml" format
out_f = os.path.basename(f).replace('.network.j2.yaml',
'.yaml')
if os.path.dirname(f).endswith('ports'):
out_f = out_f.replace('port',
n_map[network]['name_lower'])
else:
out_f = out_f.replace('network',
n_map[network]['name_lower'])
out_f_path = os.path.join(os.path.dirname(f), out_f)
if not (out_f_path in excl_templates):
self._j2_render_and_put(j2_template,
j2_data,
out_f_path,
context=context)
else:
LOG.info("Skipping rendering of %s, defined in %s" %
(out_f_path, j2_excl_data))
elif f.endswith('.j2.yaml'):
LOG.info("jinja2 rendering %s" % f)
j2_template = swiftutils.get_object_string(swift,
self.container,
f)
j2_data = {'roles': role_data, 'networks': network_data}
out_f = f.replace('.j2.yaml', '.yaml')
self._j2_render_and_put(j2_template,
j2_data,
out_f,
context=context)
def run(self, context):
error_text = None
self.context = context
swift = self.get_object_client(context)
try:
plan_env = plan_utils.get_env(swift, self.container)
except swiftexceptions.ClientException as err:
err_msg = ("Error retrieving environment for plan %s: %s" % (
self.container, err))
LOG.exception(err_msg)
return actions.Result(error=error_text)
try:
# if the jinja overcloud template exists, process it and write it
# back to the swift container before continuing processing. The
# method called below should handle the case where the files are
# not found in swift, but if they are found and an exception
# occurs during processing, that exception will cause the
# ProcessTemplatesAction to return an error result.
self._process_custom_roles(context)
except Exception as err:
LOG.exception("Error occurred while processing custom roles.")
return actions.Result(error=six.text_type(err))
template_name = plan_env.get('template', "")
template_object = os.path.join(swift.url, self.container,
template_name)
LOG.debug('Template: %s' % template_name)
try:
template_files, template = plan_utils.get_template_contents(
swift, template_object)
except Exception as err:
error_text = six.text_type(err)
LOG.exception("Error occurred while fetching %s" % template_object)
temp_env_paths = []
try:
env_paths, temp_env_paths = plan_utils.build_env_paths(
swift, self.container, plan_env)
env_files, env = plan_utils.process_environments_and_files(
swift, env_paths)
parameters.convert_docker_params(env)
except Exception as err:
error_text = six.text_type(err)
LOG.exception("Error occurred while processing plan files.")
finally:
# cleanup any local temp files
for f in temp_env_paths:
os.remove(f)
if error_text:
return actions.Result(error=error_text)
files = dict(list(template_files.items()) + list(env_files.items()))
return {
'stack_name': self.container,
'template': template,
'environment': env,
'files': files
}

4
tripleo_common/tests/actions/test_package_update.py

@ -25,7 +25,7 @@ class UpdateStackActionTest(base.TestCase):
self.timeout = 1
self.container = 'container'
@mock.patch('tripleo_common.actions.templates.ProcessTemplatesAction.run')
@mock.patch('tripleo_common.utils.template.process_templates')
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
@mock.patch('tripleo_common.actions.base.TripleOAction.'
'get_orchestration_client')
@ -40,7 +40,7 @@ class UpdateStackActionTest(base.TestCase):
mock_template_contents,
mock_get_orchestration_client,
mock_get_object_client,
mock_templates_run):
mock_process_templates):
mock_ctx = mock.MagicMock()
heat = mock.MagicMock()

420
tripleo_common/tests/actions/test_templates.py

@ -12,12 +12,7 @@
# 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 jinja2
import mock
import yaml
from heatclient import exc as heat_exc
from swiftclient import exceptions as swiftexceptions
from tripleo_common.actions import templates
from tripleo_common import constants
@ -119,418 +114,3 @@ class UploadTemplatesActionTest(base.TestCase):
constants.DEFAULT_TEMPLATES_PATH, 'test')
mock_extract_tar.assert_called_once_with(
mock_get_swift.return_value, 'test', 'tar-container')
class J2SwiftLoaderTest(base.TestCase):
@staticmethod
def _setup_swift():
def return_multiple_files(*args):
if args[1] == 'bar/foo.yaml':
return ['', 'I am foo']
else:
raise swiftexceptions.ClientException('not found')
swift = mock.MagicMock()
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
return swift
def test_include_absolute_path(self):
j2_loader = templates.J2SwiftLoader(self._setup_swift(), None)
template = jinja2.Environment(loader=j2_loader).from_string(
r'''
Included this:
{% include 'bar/foo.yaml' %}
''')
self.assertEqual(
template.render(),
'''
Included this:
I am foo
''')
def test_include_search_path(self):
j2_loader = templates.J2SwiftLoader(self._setup_swift(), None, 'bar')
template = jinja2.Environment(loader=j2_loader).from_string(
r'''
Included this:
{% include 'foo.yaml' %}
''')
self.assertEqual(
template.render(),
'''
Included this:
I am foo
''')
def test_include_not_found(self):
j2_loader = templates.J2SwiftLoader(self._setup_swift(), None)
template = jinja2.Environment(loader=j2_loader).from_string(
r'''
Included this:
{% include 'bar.yaml' %}
''')
self.assertRaises(
jinja2.exceptions.TemplateNotFound,
template.render)
def test_include_invalid_path(self):
j2_loader = templates.J2SwiftLoader(self._setup_swift(), 'bar')
template = jinja2.Environment(loader=j2_loader).from_string(
r'''
Included this:
{% include '../foo.yaml' %}
''')
self.assertRaises(
jinja2.exceptions.TemplateNotFound,
template.render)
class ProcessTemplatesActionTest(base.TestCase):
@mock.patch('heatclient.common.template_utils.'
'process_multiple_environments_and_files')
@mock.patch('heatclient.common.template_utils.get_template_contents')
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
@mock.patch('tripleo_common.actions.base.TripleOAction.'
'get_orchestration_client')
def test_run(self, mock_get_heat_client, mock_get_object_client,
mock_get_template_contents,
mock_process_multiple_environments_and_files):
mock_ctx = mock.MagicMock()
swift = mock.MagicMock(url="http://test.com")
mock_env = yaml.safe_dump({
'temp_environment': 'temp_environment',
'template': 'template',
'environments': [{u'path': u'environments/test.yaml'}]
}, default_flow_style=False)
swift.get_object.side_effect = (
({}, mock_env),
swiftexceptions.ClientException('atest2')
)
mock_get_object_client.return_value = swift
mock_get_template_contents.return_value = ({}, {
'heat_template_version': '2016-04-30'
})
mock_process_multiple_environments_and_files.return_value = ({}, {})
# Test
action = templates.ProcessTemplatesAction()
result = action.run(mock_ctx)
# Verify the values we get out
self.assertEqual(result, {
'environment': {},
'files': {},
'stack_name': constants.DEFAULT_CONTAINER_NAME,
'template': {
'heat_template_version': '2016-04-30'
}
})
def _custom_roles_mock_objclient(self, snippet_name, snippet_content,
role_data=None):
def return_multiple_files(*args):
if args[1] == constants.OVERCLOUD_J2_NAME:
return ['', JINJA_SNIPPET]
if args[1] == snippet_name:
return ['', snippet_content]
if args[1] == constants.OVERCLOUD_J2_EXCLUDES:
return ['', J2_EXCLUDES]
elif args[1] == constants.OVERCLOUD_J2_ROLES_NAME:
return ['', role_data or ROLE_DATA_YAML]
elif args[1] == constants.OVERCLOUD_J2_NETWORKS_NAME:
return ['', NETWORK_DATA_YAML]
def return_container_files(*args):
return ('headers', [
{'name': constants.OVERCLOUD_J2_NAME},
{'name': snippet_name},
{'name': constants.OVERCLOUD_J2_ROLES_NAME},
{'name': constants.OVERCLOUD_J2_NETWORKS_NAME}])
# setup swift
swift = mock.MagicMock()
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
swift.get_container = mock.MagicMock(
side_effect=return_container_files)
return swift
@mock.patch('tripleo_common.actions.templates.ProcessTemplatesAction'
'._heat_resource_exists')
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
@mock.patch('tripleo_common.actions.base.TripleOAction.'
'get_orchestration_client')
def test_process_custom_roles(self, get_heat_client_mock,
get_obj_client_mock, resource_exists_mock):
resource_exists_mock.return_value = False
swift = self._custom_roles_mock_objclient(
'foo.j2.yaml', JINJA_SNIPPET)
get_obj_client_mock.return_value = swift
# Test
action = templates.ProcessTemplatesAction()
mock_ctx = mock.MagicMock()
action._process_custom_roles(mock_ctx)
get_object_mock_calls = [
mock.call('overcloud', constants.OVERCLOUD_J2_NAME),
mock.call('overcloud', constants.OVERCLOUD_J2_ROLES_NAME),
mock.call('overcloud', 'foo.j2.yaml'),
]
swift.get_object.assert_has_calls(
get_object_mock_calls, any_order=True)
put_object_mock_calls = [
mock.call(constants.DEFAULT_CONTAINER_NAME,
constants.OVERCLOUD_YAML_NAME,
EXPECTED_JINJA_RESULT),
mock.call(constants.DEFAULT_CONTAINER_NAME,
'foo.yaml',
EXPECTED_JINJA_RESULT),
]
swift.put_object.assert_has_calls(
put_object_mock_calls, any_order=True)
@mock.patch('tripleo_common.actions.templates.ProcessTemplatesAction'
'._heat_resource_exists')
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
@mock.patch('tripleo_common.actions.base.TripleOAction'
'.get_orchestration_client')
def test_custom_roles_networks(self, get_heat_client_mock,
get_obj_client_mock, resource_exists_mock):
resource_exists_mock.return_value = False
swift = self._custom_roles_mock_objclient(
'role-networks.role.j2.yaml', JINJA_SNIPPET_ROLE_NETWORKS,
ROLE_DATA_ENABLE_NETWORKS)
get_obj_client_mock.return_value = swift
# Test
action = templates.ProcessTemplatesAction()
mock_ctx = mock.MagicMock()
action._process_custom_roles(mock_ctx)
expected = EXPECTED_JINJA_RESULT.replace(
'CustomRole', 'RoleWithNetworks')
put_object_mock_call = mock.call(
constants.DEFAULT_CONTAINER_NAME,
'overcloud.yaml',
expected)
self.assertEqual(swift.put_object.call_args_list[0],
put_object_mock_call)
put_object_mock_call = mock.call(
constants.DEFAULT_CONTAINER_NAME,
"rolewithnetworks-role-networks.yaml",
EXPECTED_JINJA_RESULT_ROLE_NETWORKS)
self.assertEqual(put_object_mock_call,
swift.put_object.call_args_list[1])
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
def test_j2_render_and_put(self, get_obj_client_mock):
# setup swift
swift = mock.MagicMock()
swift.get_object = mock.MagicMock()
swift.get_container = mock.MagicMock()
get_obj_client_mock.return_value = swift
mock_ctx = mock.MagicMock()
# Test
action = templates.ProcessTemplatesAction()
action._j2_render_and_put(JINJA_SNIPPET_CONFIG,
{'role': 'CustomRole'},
'customrole-config.yaml', context=mock_ctx)
action_result = swift.put_object._mock_mock_calls[0]
self.assertTrue("CustomRole" in str(action_result))
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
def test_j2_render_and_put_include(self, get_obj_client_mock):
def return_multiple_files(*args):
if args[1] == 'foo.yaml':
return ['', JINJA_SNIPPET_CONFIG]
def return_container_files(*args):
return ('headers', [{'name': 'foo.yaml'}])
# setup swift
swift = mock.MagicMock()
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
swift.get_container = mock.MagicMock(
side_effect=return_container_files)
get_obj_client_mock.return_value = swift
mock_ctx = mock.MagicMock()
# Test
action = templates.ProcessTemplatesAction()
action._j2_render_and_put(r"{% include 'foo.yaml' %}",
{'role': 'CustomRole'},
'customrole-config.yaml',
context=mock_ctx)
action_result = swift.put_object._mock_mock_calls[0]
self.assertTrue("CustomRole" in str(action_result))
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
def test_j2_render_and_put_include_relative(
self,
get_obj_client_mock):
def return_multiple_files(*args):
if args[1] == 'bar/foo.yaml':
return ['', JINJA_SNIPPET_CONFIG]
def return_container_files(*args):
return ('headers', [{'name': 'bar/foo.yaml'}])
# setup swift
swift = mock.MagicMock()
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
swift.get_container = mock.MagicMock(
side_effect=return_container_files)
get_obj_client_mock.return_value = swift
mock_ctx = mock.MagicMock()
# Test
action = templates.ProcessTemplatesAction()
action._j2_render_and_put(r"{% include 'foo.yaml' %}",
{'role': 'CustomRole'},
'bar/customrole-config.yaml',
context=mock_ctx)
action_result = swift.put_object._mock_mock_calls[0]
self.assertTrue("CustomRole" in str(action_result))
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
def test_get_j2_excludes_file(self, get_obj_client_mock):
mock_ctx = mock.MagicMock()
swift = mock.MagicMock()
get_obj_client_mock.return_value = swift
def return_multiple_files(*args):
if args[1] == constants.OVERCLOUD_J2_EXCLUDES:
return ['', J2_EXCLUDES]
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
# Test - J2 exclude file with valid templates
action = templates.ProcessTemplatesAction()
self.assertTrue({'name': ['puppet/controller-role.yaml']} ==
action._get_j2_excludes_file(mock_ctx))
def return_multiple_files(*args):
if args[1] == constants.OVERCLOUD_J2_EXCLUDES:
return ['', J2_EXCLUDES_EMPTY_LIST]
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
# Test - J2 exclude file with no template to exlude
action = templates.ProcessTemplatesAction()
self.assertTrue({'name': []} == action._get_j2_excludes_file(mock_ctx))
def return_multiple_files(*args):
if args[1] == constants.OVERCLOUD_J2_EXCLUDES:
return ['', J2_EXCLUDES_EMPTY_FILE]
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
# Test - J2 exclude file empty
action = templates.ProcessTemplatesAction()
self.assertTrue({'name': []} == action._get_j2_excludes_file(mock_ctx))
@mock.patch('tripleo_common.actions.base.TripleOAction.'
'get_orchestration_client')
def test_heat_resource_exists(self, client_mock):
mock_ctx = mock.MagicMock()
heat_client = mock.MagicMock()
stack = mock.MagicMock(stack_name='overcloud')
heat_client.resources.get.return_value = mock.MagicMock(
links=[{'rel': 'stack',
'href': 'http://192.0.2.1:8004/v1/'
'a959ac7d6a4a475daf2428df315c41ef/'
'stacks/overcloud/123'}],
logical_resource_id='logical_id',
physical_resource_id='resource_id',
resource_type='OS::Heat::ResourceGroup',
resource_name='InternalApiNetwork'
)
client_mock.return_value = heat_client
action = templates.ProcessTemplatesAction()
self.assertTrue(
action._heat_resource_exists(
heat_client, stack, 'Networks', 'InternalNetwork', mock_ctx))
@mock.patch('tripleo_common.actions.base.TripleOAction.'
'get_orchestration_client')
def test_no_heat_resource_exists(self, client_mock):
mock_ctx = mock.MagicMock()
heat_client = mock.MagicMock()
stack = mock.MagicMock(stack_name='overcloud')
def return_not_found(*args):
raise heat_exc.HTTPNotFound()
heat_client.resources.get.side_effect = return_not_found
client_mock.return_value = heat_client
action = templates.ProcessTemplatesAction()
self.assertFalse(
action._heat_resource_exists(
heat_client, stack, 'Networks', 'InternalNetwork', mock_ctx))
@mock.patch('tripleo_common.actions.templates.ProcessTemplatesAction'
'._heat_resource_exists')
@mock.patch('tripleo_common.actions.templates.ProcessTemplatesAction'
'._j2_render_and_put')
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
@mock.patch('tripleo_common.actions.base.TripleOAction.'
'get_orchestration_client')
def test_legacy_api_network_exists(self, get_heat_client_mock,
get_obj_client_mock, j2_mock,
resource_exists_mock):
resource_exists_mock.return_value = True
swift = self._custom_roles_mock_objclient(
'role-networks.role.j2.yaml', JINJA_SNIPPET_ROLE_NETWORKS,
ROLE_DATA_ENABLE_NETWORKS)
get_obj_client_mock.return_value = swift
# Test
action = templates.ProcessTemplatesAction()
mock_ctx = mock.MagicMock()
action._process_custom_roles(mock_ctx)
expected_j2_template = get_obj_client_mock.get_object(
action.container, 'foo.j2.yaml')[1]
expected_j2_data = {'roles': [{'name': 'CustomRole'}],
'networks': [{'name': 'InternalApi',
'compat_name': 'Internal'}]
}
assert j2_mock.called_with(expected_j2_template, expected_j2_data,
'foo.yaml', mock_ctx)
@mock.patch('tripleo_common.actions.templates.ProcessTemplatesAction'
'._heat_resource_exists')
@mock.patch('tripleo_common.actions.templates.ProcessTemplatesAction'
'._j2_render_and_put')
@mock.patch('tripleo_common.actions.base.TripleOAction.get_object_client')
@mock.patch('tripleo_common.actions.base.TripleOAction.'
'get_orchestration_client')
def test_no_legacy_api_network_exists(self, get_heat_client_mock,
get_obj_client_mock, j2_mock,
resource_exists_mock):
resource_exists_mock.return_value = False
swift = self._custom_roles_mock_objclient(
'role-networks.role.j2.yaml', JINJA_SNIPPET_ROLE_NETWORKS,
ROLE_DATA_ENABLE_NETWORKS)
get_obj_client_mock.return_value = swift
# Test
action = templates.ProcessTemplatesAction()
mock_ctx = mock.MagicMock()
action._process_custom_roles(mock_ctx)
expected_j2_template = get_obj_client_mock.get_object(
action.container, 'foo.j2.yaml')[1]
expected_j2_data = {'roles': [{'name': 'CustomRole'}],
'networks': [{'name': 'InternalApi'}]
}
assert j2_mock.called_with(expected_j2_template, expected_j2_data,
'foo.yaml', mock_ctx)

460
tripleo_common/tests/utils/test_template.py

@ -0,0 +1,460 @@
# Copyright 2016 Red Hat, Inc.
# All Rights Reserved.
#
# 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 jinja2
import mock
import yaml
from heatclient import exc as heat_exc
from swiftclient import exceptions as swiftexceptions
from tripleo_common import constants
from tripleo_common.tests import base
from tripleo_common.utils import template as template_utils
JINJA_SNIPPET = r"""
# Jinja loop for Role in role_data.yaml
{% for role in roles %}
# Resources generated for {{role.name}} Role
{{role.name}}ServiceChain:
type: OS::TripleO::Services
properties:
Services:
get_param: {{role.name}}Services
ServiceNetMap: {get_attr: [ServiceNetMap, service_net_map]}
EndpointMap: {get_attr: [EndpointMap, endpoint_map]}
DefaultPasswords: {get_attr: [DefaultPasswords, passwords]}
{% endfor %}"""
ROLE_DATA_YAML = r"""
-
name: CustomRole
"""
NETWORK_DATA_YAML = r"""
-
name: InternalApi
"""
EXPECTED_JINJA_RESULT = r"""
# Jinja loop for Role in role_data.yaml
# Resources generated for CustomRole Role
CustomRoleServiceChain:
type: OS::TripleO::Services
properties:
Services:
get_param: CustomRoleServices
ServiceNetMap: {get_attr: [ServiceNetMap, service_net_map]}
EndpointMap: {get_attr: [EndpointMap, endpoint_map]}
DefaultPasswords: {get_attr: [DefaultPasswords, passwords]}
"""
JINJA_SNIPPET_CONFIG = r"""
outputs:
OS::stack_id:
description: The software config which runs puppet on the {{role}} role
value: {get_resource: {{role}}PuppetConfigImpl}"""
J2_EXCLUDES = r"""
name:
- puppet/controller-role.yaml
"""
J2_EXCLUDES_EMPTY_LIST = r"""
name:
"""
J2_EXCLUDES_EMPTY_FILE = r"""
"""
ROLE_DATA_ENABLE_NETWORKS = r"""
- name: RoleWithNetworks
networks:
- InternalApi
"""
JINJA_SNIPPET_ROLE_NETWORKS = r"""
{%- for network in networks %}
{%- if network.name in role.networks%}
{{network.name}}Port:
type: {{role.name}}::{{network.name}}::Port
{%- endif %}
{% endfor %}
"""
EXPECTED_JINJA_RESULT_ROLE_NETWORKS = r"""
InternalApiPort:
type: RoleWithNetworks::InternalApi::Port
"""
class J2SwiftLoaderTest(base.TestCase):
@staticmethod
def _setup_swift():
def return_multiple_files(*args):
if args[1] == 'bar/foo.yaml':
return ['', 'I am foo']
else:
raise swiftexceptions.ClientException('not found')
swift = mock.MagicMock()
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
return swift
def test_include_absolute_path(self):
j2_loader = template_utils.J2SwiftLoader(self._setup_swift(), None)
template = jinja2.Environment(loader=j2_loader).from_string(
r'''
Included this:
{% include 'bar/foo.yaml' %}
''')
self.assertEqual(
template.render(),
'''
Included this:
I am foo
''')
def test_include_search_path(self):
j2_loader = template_utils.J2SwiftLoader(self._setup_swift(),
None, 'bar')
template = jinja2.Environment(loader=j2_loader).from_string(
r'''
Included this:
{% include 'foo.yaml' %}
''')
self.assertEqual(
template.render(),
'''
Included this:
I am foo
''')
def test_include_not_found(self):
j2_loader = template_utils.J2SwiftLoader(self._setup_swift(), None)
template = jinja2.Environment(loader=j2_loader).from_string(
r'''
Included this:
{% include 'bar.yaml' %}
''')
self.assertRaises(
jinja2.exceptions.TemplateNotFound,
template.render)
def test_include_invalid_path(self):
j2_loader = template_utils.J2SwiftLoader(self._setup_swift(), 'bar')
template = jinja2.Environment(loader=j2_loader).from_string(
r'''
Included this:
{% include '../foo.yaml' %}
''')
self.assertRaises(
jinja2.exceptions.TemplateNotFound,
template.render)
class ProcessTemplatesTest(base.TestCase):
@mock.patch('heatclient.common.template_utils.'
'process_multiple_environments_and_files')
@mock.patch('heatclient.common.template_utils.get_template_contents')
def test_process_template(self, mock_get_template_contents,
mock_process_multiple_environments_and_files):
swift = mock.MagicMock(url="http://test.com")
heat = mock.MagicMock()
mock_env = yaml.safe_dump({
'temp_environment': 'temp_environment',
'template': 'template',
'environments': [{u'path': u'environments/test.yaml'}]
}, default_flow_style=False)
swift.get_object.side_effect = (
({}, mock_env),
swiftexceptions.ClientException('atest2')
)
mock_get_template_contents.return_value = ({}, {
'heat_template_version': '2016-04-30'
})
mock_process_multiple_environments_and_files.return_value = ({}, {})
# Test
processed_data = template_utils.process_templates(
swift, heat, constants.DEFAULT_CONTAINER_NAME)
# Verify the values we get out
self.assertEqual(processed_data, {
'environment': {},
'files': {},
'stack_name': constants.DEFAULT_CONTAINER_NAME,
'template': {
'heat_template_version': '2016-04-30'
}
})
def _custom_roles_mock_objclient(self, snippet_name, snippet_content,
role_data=None):
def return_multiple_files(*args):
if args[1] == constants.OVERCLOUD_J2_NAME:
return ['', JINJA_SNIPPET]
if args[1] == snippet_name:
return ['', snippet_content]
if args[1] == constants.OVERCLOUD_J2_EXCLUDES:
return ['', J2_EXCLUDES]
elif args[1] == constants.OVERCLOUD_J2_ROLES_NAME:
return ['', role_data or ROLE_DATA_YAML]
elif args[1] == constants.OVERCLOUD_J2_NETWORKS_NAME:
return ['', NETWORK_DATA_YAML]
def return_container_files(*args):
return ('headers', [
{'name': constants.OVERCLOUD_J2_NAME},
{'name': snippet_name},
{'name': constants.OVERCLOUD_J2_ROLES_NAME},
{'name': constants.OVERCLOUD_J2_NETWORKS_NAME}])
# setup swift
swift = mock.MagicMock()
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
swift.get_container = mock.MagicMock(
side_effect=return_container_files)
return swift
@mock.patch('tripleo_common.utils.template.heat_resource_exists')
def test_process_custom_roles(self, resource_exists_mock):
resource_exists_mock.return_value = False
swift = self._custom_roles_mock_objclient(
'foo.j2.yaml', JINJA_SNIPPET)
heat = mock.MagicMock()
# Test
template_utils.process_custom_roles(
swift, heat, constants.DEFAULT_CONTAINER_NAME)
get_object_mock_calls = [
mock.call('overcloud', constants.OVERCLOUD_J2_NAME),
mock.call('overcloud', constants.OVERCLOUD_J2_ROLES_NAME),
mock.call('overcloud', 'foo.j2.yaml'),
]
swift.get_object.assert_has_calls(
get_object_mock_calls, any_order=True)
put_object_mock_calls = [
mock.call(constants.DEFAULT_CONTAINER_NAME,
constants.OVERCLOUD_YAML_NAME,
EXPECTED_JINJA_RESULT),
mock.call(constants.DEFAULT_CONTAINER_NAME,
'foo.yaml',
EXPECTED_JINJA_RESULT),
]
swift.put_object.assert_has_calls(
put_object_mock_calls, any_order=True)
@mock.patch('tripleo_common.utils.template.heat_resource_exists')
def test_custom_roles_networks(self, resource_exists_mock):
heat = mock.MagicMock()
resource_exists_mock.return_value = False
swift = self._custom_roles_mock_objclient(
'role-networks.role.j2.yaml', JINJA_SNIPPET_ROLE_NETWORKS,
ROLE_DATA_ENABLE_NETWORKS)
# Test
template_utils.process_custom_roles(
swift, heat, constants.DEFAULT_CONTAINER_NAME)
expected = EXPECTED_JINJA_RESULT.replace(
'CustomRole', 'RoleWithNetworks')
put_object_mock_call = mock.call(
constants.DEFAULT_CONTAINER_NAME,
'overcloud.yaml',
expected)
self.assertEqual(swift.put_object.call_args_list[0],
put_object_mock_call)
put_object_mock_call = mock.call(
constants.DEFAULT_CONTAINER_NAME,
"rolewithnetworks-role-networks.yaml",
EXPECTED_JINJA_RESULT_ROLE_NETWORKS)
self.assertEqual(put_object_mock_call,
swift.put_object.call_args_list[1])
def test_j2_render_and_put(self):
# setup swift
swift = mock.MagicMock()
swift.get_object = mock.MagicMock()
swift.get_container = mock.MagicMock()
# Test
template_utils.j2_render_and_put(
swift,
JINJA_SNIPPET_CONFIG, {'role': 'CustomRole'},
'customrole-config.yaml',
constants.DEFAULT_CONTAINER_NAME)
result = swift.put_object._mock_mock_calls[0]
self.assertTrue("CustomRole" in str(result))
def test_j2_render_and_put_include(self):
def return_multiple_files(*args):
if args[1] == 'foo.yaml':
return ['', JINJA_SNIPPET_CONFIG]
def return_container_files(*args):
return ('headers', [{'name': 'foo.yaml'}])
# setup swift
swift = mock.MagicMock()
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
swift.get_container = mock.MagicMock(
side_effect=return_container_files)
# Test
template_utils.j2_render_and_put(
swift,
r"{% include 'foo.yaml' %}",
{'role': 'CustomRole'},
'customrole-config.yaml',
constants.DEFAULT_CONTAINER_NAME)
result = swift.put_object._mock_mock_calls[0]
self.assertTrue("CustomRole" in str(result))
def test_j2_render_and_put_include_relative(self):
def return_multiple_files(*args):
if args[1] == 'bar/foo.yaml':
return ['', JINJA_SNIPPET_CONFIG]
def return_container_files(*args):
return ('headers', [{'name': 'bar/foo.yaml'}])
# setup swift
swift = mock.MagicMock()
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
swift.get_container = mock.MagicMock(
side_effect=return_container_files)
# Test
template_utils.j2_render_and_put(
swift,
r"{% include 'bar/foo.yaml' %}",
{'role': 'CustomRole'},
'bar/customrole-config.yaml',
constants.DEFAULT_CONTAINER_NAME)
result = swift.put_object._mock_mock_calls[0]
self.assertTrue("CustomRole" in str(result))
def test_get_j2_excludes_file(self):
swift = mock.MagicMock()
def return_multiple_files(*args):
if args[1] == constants.OVERCLOUD_J2_EXCLUDES:
return ['', J2_EXCLUDES]
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
# Test - J2 exclude file with valid templates
self.assertTrue({'name': ['puppet/controller-role.yaml']} ==
template_utils.get_j2_excludes_file(swift))
def return_multiple_files(*args):
if args[1] == constants.OVERCLOUD_J2_EXCLUDES:
return ['', J2_EXCLUDES_EMPTY_LIST]
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
# Test - J2 exclude file with no template to exlude
self.assertTrue(
{'name': []} == template_utils.get_j2_excludes_file(swift))
def return_multiple_files(*args):
if args[1] == constants.OVERCLOUD_J2_EXCLUDES:
return ['', J2_EXCLUDES_EMPTY_FILE]
swift.get_object = mock.MagicMock(side_effect=return_multiple_files)
# Test - J2 exclude file empty
self.assertTrue(
{'name': []} == template_utils.get_j2_excludes_file(swift))
def test_heat_resource_exists(self):
heat_client = mock.MagicMock()
stack = mock.MagicMock(stack_name='overcloud')
heat_client.resources.get.return_value = mock.MagicMock(
links=[{'rel': 'stack',
'href': 'http://192.0.2.1:8004/v1/'
'a959ac7d6a4a475daf2428df315c41ef/'
'stacks/overcloud/123'}],
logical_resource_id='logical_id',
physical_resource_id='resource_id',
resource_type='OS::Heat::ResourceGroup',
resource_name='InternalApiNetwork'
)
self.assertTrue(
template_utils.heat_resource_exists(
heat_client, stack, 'Networks', 'InternalNetwork'))
def test_no_heat_resource_exists(self):
heat_client = mock.MagicMock()
stack = mock.MagicMock(stack_name='overcloud')
def return_not_found(*args):
raise heat_exc.HTTPNotFound()
heat_client.resources.get.side_effect = return_not_found
self.assertFalse(
template_utils.heat_resource_exists(
heat_client, stack, 'Networks', 'InternalNetwork'))
@mock.patch('tripleo_common.utils.template.heat_resource_exists')
@mock.patch('tripleo_common.utils.template.j2_render_and_put')
def test_legacy_api_network_exists(self, j2_mock,
resource_exists_mock):
resource_exists_mock.return_value = True
swift = self._custom_roles_mock_objclient(
'role-networks.role.j2.yaml', JINJA_SNIPPET_ROLE_NETWORKS,
ROLE_DATA_ENABLE_NETWORKS)
heat = mock.MagicMock()