1 # Licensed under the Apache License, Version 2.0 (the "License"); you may
2 # not use this file except in compliance with the License. You may obtain
3 # a copy of the License at
5 # http://www.apache.org/licenses/LICENSE-2.0
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10 # License for the specific language governing permissions and limitations
15 from toscaparser.common import exception
16 from toscaparser.substitution_mappings import SubstitutionMappings
17 from toscaparser.tests.base import TestCase
18 from toscaparser.topology_template import TopologyTemplate
19 from toscaparser.tosca_template import ToscaTemplate
20 from toscaparser.utils.gettextutils import _
21 import toscaparser.utils.yamlparser
23 YAML_LOADER = toscaparser.utils.yamlparser.load_yaml
26 class TopologyTemplateTest(TestCase):
31 self.tosca_tpl_path = os.path.join(
32 os.path.dirname(os.path.abspath(__file__)),
33 "data/topology_template/transactionsubsystem.yaml")
34 self.tpl = YAML_LOADER(self.tosca_tpl_path)
35 self.topo_tpl = self.tpl.get('topology_template')
36 self.imports = self.tpl.get('imports')
37 self.topo = TopologyTemplate(self.topo_tpl,
38 self._get_all_custom_def())
40 def _get_custom_def(self, type_definition):
42 for definition in self.imports:
43 if os.path.isabs(definition):
46 tpl_dir = os.path.dirname(os.path.abspath(self.tosca_tpl_path))
47 def_file = os.path.join(tpl_dir, definition)
48 custom_type = YAML_LOADER(def_file)
49 custom_defs.update(custom_type.get(type_definition))
52 def _get_all_custom_def(self):
54 custom_defs.update(self._get_custom_def('node_types'))
55 custom_defs.update(self._get_custom_def('capability_types'))
58 def _get_custom_types(self):
60 def_file = os.path.join(
61 os.path.dirname(os.path.abspath(__file__)),
62 "data/topology_template/definitions.yaml")
63 custom_type = YAML_LOADER(def_file)
64 node_types = custom_type['node_types']
65 for name in node_types:
66 defintion = node_types[name]
67 custom_types[name] = defintion
70 def test_description(self):
71 expected_desc = 'Template of a database including its hosting stack.'
72 self.assertEqual(expected_desc, self.topo.description)
74 def test_inputs(self):
76 ['mq_server_ip', 'my_cpus', 'receiver_port'],
77 sorted([input.name for input in self.topo.inputs]))
79 input_name = "receiver_port"
80 expected_description = "Port to be used for receiving messages."
81 for input in self.topo.inputs:
82 if input.name == input_name:
83 self.assertEqual(expected_description, input.description)
85 def test_node_tpls(self):
86 '''Test nodetemplate names.'''
88 ['app', 'server', 'websrv'],
89 sorted([tpl.name for tpl in self.topo.nodetemplates]))
92 expected_type = "example.SomeApp"
93 expected_properties = ['admin_user', 'pool_size']
94 expected_capabilities = ['feature', 'message_receiver']
95 expected_requirements = [{'host': {'node': 'websrv'}}]
96 expected_relationshp = ['tosca.relationships.HostedOn']
97 expected_host = ['websrv']
98 for tpl in self.topo.nodetemplates:
99 if tpl_name == tpl.name:
100 '''Test node type.'''
101 self.assertEqual(tpl.type, expected_type)
103 '''Test properties.'''
106 sorted(tpl.get_properties().keys()))
108 '''Test capabilities.'''
110 expected_capabilities,
111 sorted(tpl.get_capabilities().keys()))
113 '''Test requirements.'''
115 expected_requirements, tpl.requirements)
117 '''Test relationship.'''
118 ''' TODO : skip tempororily. need to fix it
121 expected_relationshp,
122 [x.type for x in tpl.relationships.keys()])
125 [y.name for y in tpl.relationships.values()])
126 '''Test interfaces.'''
127 # TODO(hurf) add interface test when new template is available
129 if tpl.name == 'server':
130 '''Test property value'''
131 props = tpl.get_properties()
132 if props and 'mem_size' in props.keys():
133 self.assertEqual(props['mem_size'].value, '4096 MB')
134 '''Test capability'''
135 caps = tpl.get_capabilities()
136 self.assertIn('os', caps.keys())
140 if caps and 'os' in caps.keys():
141 capability = caps['os']
142 os_props_objs = capability.get_properties_objects()
143 os_props = capability.get_properties()
144 os_type_prop = capability.get_property_value('type')
148 [p.value for p in os_props_objs if p.name == 'type'])
151 os_props['type'].value if 'type' in os_props else '')
152 self.assertEqual('Linux', os_props['type'].value)
153 self.assertEqual('Linux', os_type_prop)
155 def test_outputs(self):
158 sorted([output.name for output in self.topo.outputs]))
160 def test_groups(self):
161 group = self.topo.groups[0]
162 self.assertEqual('webserver_group', group.name)
163 self.assertEqual(['websrv', 'server'], group.members)
164 for node in group.get_member_nodes():
165 if node.name == 'server':
166 '''Test property value'''
167 props = node.get_properties()
168 if props and 'mem_size' in props.keys():
169 self.assertEqual(props['mem_size'].value, '4096 MB')
171 def test_system_template(self):
172 tpl_path = os.path.join(
173 os.path.dirname(os.path.abspath(__file__)),
174 "data/topology_template/system.yaml")
175 system_tosca_template = ToscaTemplate(tpl_path)
176 self.assertIsNotNone(system_tosca_template)
178 len(system_tosca_template.
179 nested_tosca_templates_with_topology), 4)
180 self.assertTrue(system_tosca_template.has_nested_templates())
182 def test_invalid_keyname(self):
184 substitution_mappings:
185 node_type: example.DatabaseSubsystem
187 database_endpoint: [ db_app, database_endpoint ]
189 receiver1: [ tran_app, receiver1 ]
192 sub_mappings = (toscaparser.utils.yamlparser.
193 simple_parse(tpl_snippet))['substitution_mappings']
194 expected_message = _(
195 'SubstitutionMappings contains unknown field '
196 '"invalid_key". Refer to the definition '
197 'to verify valid values.')
198 err = self.assertRaises(
199 exception.UnknownFieldError,
200 lambda: SubstitutionMappings(sub_mappings, None, None,
202 self.assertEqual(expected_message, err.__str__())
204 def test_missing_required_keyname(self):
206 substitution_mappings:
208 database_endpoint: [ db_app, database_endpoint ]
210 receiver1: [ tran_app, receiver1 ]
212 sub_mappings = (toscaparser.utils.yamlparser.
213 simple_parse(tpl_snippet))['substitution_mappings']
214 expected_message = _('SubstitutionMappings used in topology_template '
215 'is missing required field "node_type".')
216 err = self.assertRaises(
217 exception.MissingRequiredFieldError,
218 lambda: SubstitutionMappings(sub_mappings, None, None,
220 self.assertEqual(expected_message, err.__str__())
222 def test_invalid_nodetype(self):
224 substitution_mappings:
225 node_type: example.DatabaseSubsystem1
227 database_endpoint: [ db_app, database_endpoint ]
229 receiver1: [ tran_app, receiver1 ]
231 sub_mappings = (toscaparser.utils.yamlparser.
232 simple_parse(tpl_snippet))['substitution_mappings']
233 custom_defs = self._get_custom_types()
234 expected_message = _('Node type "example.DatabaseSubsystem1" '
235 'is not a valid type.')
236 err = self.assertRaises(
237 exception.InvalidNodeTypeError,
238 lambda: SubstitutionMappings(sub_mappings, None, None,
239 None, None, custom_defs))
240 self.assertEqual(expected_message, err.__str__())
242 def test_system_with_input_validation(self):
243 tpl_path0 = os.path.join(
244 os.path.dirname(os.path.abspath(__file__)),
245 "data/topology_template/validate/system_invalid_input.yaml")
246 tpl_path1 = os.path.join(
247 os.path.dirname(os.path.abspath(__file__)),
248 "data/topology_template/validate/"
249 "queuingsubsystem_invalid_input.yaml")
250 errormsg = _('SubstitutionMappings with node_type '
251 'example.QueuingSubsystem is missing '
252 'required input definition of input "server_port".')
254 # It's invalid in nested template.
255 self.assertRaises(exception.ValidationError,
256 lambda: ToscaTemplate(tpl_path0))
257 exception.ExceptionCollector.assertExceptionMessage(
258 exception.MissingRequiredInputError, errormsg)
260 # Subtemplate deploy standaolone is also invalid.
261 self.assertRaises(exception.ValidationError,
262 lambda: ToscaTemplate(tpl_path1))
263 exception.ExceptionCollector.assertExceptionMessage(
264 exception.MissingRequiredInputError, errormsg)