From e2cc2530fc6d54ebc975c01a4ff887ce12f0a736 Mon Sep 17 00:00:00 2001 From: Pavel Aharoni Date: Wed, 29 Mar 2017 13:35:45 +0300 Subject: [SDC-6] sdc-distribution-client 1707 rebasing Change-Id: I322a05fd79beb6ba4fee4d32afffecf531b86e98 Signed-off-by: Pavel Aharoni --- .../toscaparser/tests/test_topology_template.py | 283 +++++++++++++++++++++ 1 file changed, 283 insertions(+) create mode 100644 jython-tosca-parser/src/main/resources/Lib/site-packages/tosca_parser-0.7.0-py2.7.egg/toscaparser/tests/test_topology_template.py (limited to 'jython-tosca-parser/src/main/resources/Lib/site-packages/tosca_parser-0.7.0-py2.7.egg/toscaparser/tests/test_topology_template.py') diff --git a/jython-tosca-parser/src/main/resources/Lib/site-packages/tosca_parser-0.7.0-py2.7.egg/toscaparser/tests/test_topology_template.py b/jython-tosca-parser/src/main/resources/Lib/site-packages/tosca_parser-0.7.0-py2.7.egg/toscaparser/tests/test_topology_template.py new file mode 100644 index 0000000..f9e2122 --- /dev/null +++ b/jython-tosca-parser/src/main/resources/Lib/site-packages/tosca_parser-0.7.0-py2.7.egg/toscaparser/tests/test_topology_template.py @@ -0,0 +1,283 @@ +# 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 os + +from toscaparser.common import exception +from toscaparser.substitution_mappings import SubstitutionMappings +from toscaparser.tests.base import TestCase +from toscaparser.topology_template import TopologyTemplate +from toscaparser.tosca_template import ToscaTemplate +from toscaparser.utils.gettextutils import _ +import toscaparser.utils.yamlparser + +YAML_LOADER = toscaparser.utils.yamlparser.load_yaml + + +class TopologyTemplateTest(TestCase): + + def setUp(self): + TestCase.setUp(self) + '''TOSCA template.''' + self.tosca_tpl_path = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/topology_template/transactionsubsystem.yaml") + self.tpl = YAML_LOADER(self.tosca_tpl_path) + self.topo_tpl = self.tpl.get('topology_template') + self.imports = self.tpl.get('imports') + self.topo = TopologyTemplate(self.topo_tpl, + self._get_all_custom_def()) + + def _get_custom_def(self, type_definition): + custom_defs = {} + for definition in self.imports: + if os.path.isabs(definition): + def_file = definition + else: + tpl_dir = os.path.dirname(os.path.abspath(self.tosca_tpl_path)) + def_file = os.path.join(tpl_dir, definition) + custom_type = YAML_LOADER(def_file) + custom_defs.update(custom_type.get(type_definition)) + return custom_defs + + def _get_all_custom_def(self): + custom_defs = {} + custom_defs.update(self._get_custom_def('node_types')) + custom_defs.update(self._get_custom_def('capability_types')) + return custom_defs + + def _get_custom_types(self): + custom_types = {} + def_file = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/topology_template/definitions.yaml") + custom_type = YAML_LOADER(def_file) + node_types = custom_type['node_types'] + for name in node_types: + defintion = node_types[name] + custom_types[name] = defintion + return custom_types + + def test_description(self): + expected_desc = 'Template of a database including its hosting stack.' + self.assertEqual(expected_desc, self.topo.description) + + def test_inputs(self): + self.assertEqual( + ['mq_server_ip', 'my_cpus', 'receiver_port'], + sorted([input.name for input in self.topo.inputs])) + + input_name = "receiver_port" + expected_description = "Port to be used for receiving messages." + for input in self.topo.inputs: + if input.name == input_name: + self.assertEqual(expected_description, input.description) + + def test_node_tpls(self): + '''Test nodetemplate names.''' + self.assertEqual( + ['app', 'server', 'websrv'], + sorted([tpl.name for tpl in self.topo.nodetemplates])) + + tpl_name = "app" + expected_type = "example.SomeApp" + expected_properties = ['admin_user', 'pool_size'] + expected_capabilities = ['feature', 'message_receiver'] + expected_requirements = [{'host': {'node': 'websrv'}}] + expected_relationshp = ['tosca.relationships.HostedOn'] + expected_host = ['websrv'] + for tpl in self.topo.nodetemplates: + if tpl_name == tpl.name: + '''Test node type.''' + self.assertEqual(tpl.type, expected_type) + + '''Test properties.''' + self.assertEqual( + expected_properties, + sorted(tpl.get_properties().keys())) + + '''Test capabilities.''' + self.assertEqual( + expected_capabilities, + sorted(tpl.get_capabilities().keys())) + + '''Test requirements.''' + self.assertEqual( + expected_requirements, tpl.requirements) + + '''Test relationship.''' + ''' TODO : skip tempororily. need to fix it + ''' + self.assertEqual( + expected_relationshp, + [x.type for x in tpl.relationships.keys()]) + self.assertEqual( + expected_host, + [y.name for y in tpl.relationships.values()]) + '''Test interfaces.''' + # TODO(hurf) add interface test when new template is available + + if tpl.name == 'server': + '''Test property value''' + props = tpl.get_properties() + if props and 'mem_size' in props.keys(): + self.assertEqual(props['mem_size'].value, '4096 MB') + '''Test capability''' + caps = tpl.get_capabilities() + self.assertIn('os', caps.keys()) + os_props_objs = None + os_props = None + os_type_prop = None + if caps and 'os' in caps.keys(): + capability = caps['os'] + os_props_objs = capability.get_properties_objects() + os_props = capability.get_properties() + os_type_prop = capability.get_property_value('type') + break + self.assertEqual( + ['Linux'], + [p.value for p in os_props_objs if p.name == 'type']) + self.assertEqual( + 'Linux', + os_props['type'].value if 'type' in os_props else '') + self.assertEqual('Linux', os_props['type'].value) + self.assertEqual('Linux', os_type_prop) + + def test_outputs(self): + self.assertEqual( + sorted(['receiver_ip', 'receiver_port']), + sorted([output.name for output in self.topo.outputs])) + + def test_groups(self): + group = self.topo.groups[0] + self.assertEqual('webserver_group', group.name) + self.assertEqual(['websrv', 'server'], group.members) + for node in group.get_member_nodes(): + if node.name == 'server': + '''Test property value''' + props = node.get_properties() + if props and 'mem_size' in props.keys(): + self.assertEqual(props['mem_size'].value, '4096 MB') + + def test_system_template(self): + tpl_path = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/topology_template/system.yaml") + system_tosca_template = ToscaTemplate(tpl_path) + self.assertIsNotNone(system_tosca_template) + self.assertEqual( + len(system_tosca_template. + nested_tosca_templates_with_topology), 4) + self.assertTrue(system_tosca_template.has_nested_templates()) + + def test_invalid_keyname(self): + tpl_snippet = ''' + substitution_mappings: + node_type: example.DatabaseSubsystem + capabilities: + database_endpoint: [ db_app, database_endpoint ] + requirements: + receiver1: [ tran_app, receiver1 ] + invalid_key: 123 + ''' + sub_mappings = (toscaparser.utils.yamlparser. + simple_parse(tpl_snippet))['substitution_mappings'] + expected_message = _( + 'SubstitutionMappings contains unknown field ' + '"invalid_key". Refer to the definition ' + 'to verify valid values.') + err = self.assertRaises( + exception.UnknownFieldError, + lambda: SubstitutionMappings(sub_mappings, None, None, + None, None, None)) + self.assertEqual(expected_message, err.__str__()) + + def test_missing_required_keyname(self): + tpl_snippet = ''' + substitution_mappings: + capabilities: + database_endpoint: [ db_app, database_endpoint ] + requirements: + receiver1: [ tran_app, receiver1 ] + ''' + sub_mappings = (toscaparser.utils.yamlparser. + simple_parse(tpl_snippet))['substitution_mappings'] + expected_message = _('SubstitutionMappings used in topology_template ' + 'is missing required field "node_type".') + err = self.assertRaises( + exception.MissingRequiredFieldError, + lambda: SubstitutionMappings(sub_mappings, None, None, + None, None, None)) + self.assertEqual(expected_message, err.__str__()) + + def test_invalid_nodetype(self): + tpl_snippet = ''' + substitution_mappings: + node_type: example.DatabaseSubsystem1 + capabilities: + database_endpoint: [ db_app, database_endpoint ] + requirements: + receiver1: [ tran_app, receiver1 ] + ''' + sub_mappings = (toscaparser.utils.yamlparser. + simple_parse(tpl_snippet))['substitution_mappings'] + custom_defs = self._get_custom_types() + expected_message = _('Node type "example.DatabaseSubsystem1" ' + 'is not a valid type.') + err = self.assertRaises( + exception.InvalidNodeTypeError, + lambda: SubstitutionMappings(sub_mappings, None, None, + None, None, custom_defs)) + self.assertEqual(expected_message, err.__str__()) + + def test_system_with_input_validation(self): + tpl_path0 = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/topology_template/validate/system_invalid_input.yaml") + tpl_path1 = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/topology_template/validate/" + "queuingsubsystem_invalid_input.yaml") + errormsg = _('SubstitutionMappings with node_type ' + 'example.QueuingSubsystem is missing ' + 'required input definition of input "server_port".') + + # It's invalid in nested template. + self.assertRaises(exception.ValidationError, + lambda: ToscaTemplate(tpl_path0)) + exception.ExceptionCollector.assertExceptionMessage( + exception.MissingRequiredInputError, errormsg) + + # Subtemplate deploy standaolone is also invalid. + self.assertRaises(exception.ValidationError, + lambda: ToscaTemplate(tpl_path1)) + exception.ExceptionCollector.assertExceptionMessage( + exception.MissingRequiredInputError, errormsg) + + def test_substitution_mappings_valid_output(self): + tpl_path = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/topology_template/validate/" + "test_substitution_mappings_valid_output.yaml") + self.assertIsNotNone(ToscaTemplate(tpl_path)) + + def test_system_with_unknown_output_validation(self): + tpl_path = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/topology_template/validate/" + "test_substitution_mappings_invalid_output.yaml") + errormsg = _('\'Attribute "my_cpu_output" was not found in node ' + 'template "substitute_app".\'') + self.assertRaises(exception.ValidationError, + lambda: ToscaTemplate(tpl_path)) + exception.ExceptionCollector.assertExceptionMessage( + KeyError, errormsg) -- cgit 1.2.3-korg