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 test_description(self):
59 expected_desc = 'Template of a database including its hosting stack.'
60 self.assertEqual(expected_desc, self.topo.description)
62 def test_inputs(self):
64 ['mq_server_ip', 'my_cpus', 'receiver_port'],
65 sorted([input.name for input in self.topo.inputs]))
67 input_name = "receiver_port"
68 expected_description = "Port to be used for receiving messages."
69 for input in self.topo.inputs:
70 if input.name == input_name:
71 self.assertEqual(expected_description, input.description)
73 def test_node_tpls(self):
74 '''Test nodetemplate names.'''
76 ['app', 'server', 'websrv'],
77 sorted([tpl.name for tpl in self.topo.nodetemplates]))
80 expected_type = "example.SomeApp"
81 expected_properties = ['admin_user', 'pool_size']
82 expected_capabilities = ['feature', 'message_receiver']
83 expected_requirements = [{'host': {'node': 'websrv'}}]
84 expected_relationshp = ['tosca.relationships.HostedOn']
85 expected_host = ['websrv']
86 for tpl in self.topo.nodetemplates:
87 if tpl_name == tpl.name:
89 self.assertEqual(tpl.type, expected_type)
91 '''Test properties.'''
94 sorted(tpl.get_properties().keys()))
96 '''Test capabilities.'''
98 expected_capabilities,
99 sorted(tpl.get_capabilities().keys()))
101 '''Test requirements.'''
103 expected_requirements, tpl.requirements)
105 '''Test relationship.'''
106 ''' TODO : skip tempororily. need to fix it
109 expected_relationshp,
110 [x.type for x in tpl.relationships.keys()])
113 [y.name for y in tpl.relationships.values()])
114 '''Test interfaces.'''
115 # TODO(hurf) add interface test when new template is available
117 if tpl.name == 'server':
118 '''Test property value'''
119 props = tpl.get_properties()
120 if props and 'mem_size' in props.keys():
121 self.assertEqual(props['mem_size'].value, '4096 MB')
122 '''Test capability'''
123 caps = tpl.get_capabilities()
124 self.assertIn('os', caps.keys())
128 if caps and 'os' in caps.keys():
129 capability = caps['os']
130 os_props_objs = capability.get_properties_objects()
131 os_props = capability.get_properties()
132 os_type_prop = capability.get_property_value('type')
136 [p.value for p in os_props_objs if p.name == 'type'])
139 os_props['type'].value if 'type' in os_props else '')
140 self.assertEqual('Linux', os_props['type'].value)
141 self.assertEqual('Linux', os_type_prop)
143 def test_outputs(self):
146 sorted([output.name for output in self.topo.outputs]))
148 def test_groups(self):
149 group = self.topo.groups[0]
150 self.assertEqual('webserver_group', group.name)
151 self.assertEqual(['websrv', 'server'], group.members)
152 for node in group.get_member_nodes():
153 if node.name == 'server':
154 '''Test property value'''
155 props = node.get_properties()
156 if props and 'mem_size' in props.keys():
157 self.assertEqual(props['mem_size'].value, '4096 MB')
159 def test_system_template(self):
160 tpl_path = os.path.join(
161 os.path.dirname(os.path.abspath(__file__)),
162 "data/topology_template/system.yaml")
163 system_tosca_template = ToscaTemplate(tpl_path)
164 self.assertIsNotNone(system_tosca_template)
166 len(system_tosca_template.
167 nested_tosca_templates_with_topology), 4)
169 def test_invalid_keyname(self):
171 substitution_mappings:
172 node_type: example.DatabaseSubsystem
174 database_endpoint: [ db_app, database_endpoint ]
176 receiver1: [ tran_app, receiver1 ]
179 sub_mappings = (toscaparser.utils.yamlparser.
180 simple_parse(tpl_snippet))['substitution_mappings']
181 expected_message = _(
182 'SubstitutionMappings contains unknown field '
183 '"invalid_key". Refer to the definition '
184 'to verify valid values.')
185 err = self.assertRaises(
186 exception.UnknownFieldError,
187 lambda: SubstitutionMappings(sub_mappings, None, None,
189 self.assertEqual(expected_message, err.__str__())
191 def test_missing_required_keyname(self):
193 substitution_mappings:
195 database_endpoint: [ db_app, database_endpoint ]
197 receiver1: [ tran_app, receiver1 ]
199 sub_mappings = (toscaparser.utils.yamlparser.
200 simple_parse(tpl_snippet))['substitution_mappings']
201 expected_message = _('SubstitutionMappings used in topology_template '
202 'is missing required field "node_type".')
203 err = self.assertRaises(
204 exception.MissingRequiredFieldError,
205 lambda: SubstitutionMappings(sub_mappings, None, None,
207 self.assertEqual(expected_message, err.__str__())