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.tests.base import TestCase
16 from toscaparser.topology_template import TopologyTemplate
17 from toscaparser.tosca_template import ToscaTemplate
18 import toscaparser.utils.yamlparser
20 YAML_LOADER = toscaparser.utils.yamlparser.load_yaml
23 class TopologyTemplateTest(TestCase):
28 self.tosca_tpl_path = os.path.join(
29 os.path.dirname(os.path.abspath(__file__)),
30 "data/topology_template/transactionsubsystem.yaml")
31 self.tpl = YAML_LOADER(self.tosca_tpl_path)
32 self.topo_tpl = self.tpl.get('topology_template')
33 self.imports = self.tpl.get('imports')
34 self.topo = TopologyTemplate(self.topo_tpl,
35 self._get_all_custom_def())
37 def _get_custom_def(self, type_definition):
39 for definition in self.imports:
40 if os.path.isabs(definition):
43 tpl_dir = os.path.dirname(os.path.abspath(self.tosca_tpl_path))
44 def_file = os.path.join(tpl_dir, definition)
45 custom_type = YAML_LOADER(def_file)
46 custom_defs.update(custom_type.get(type_definition))
49 def _get_all_custom_def(self):
51 custom_defs.update(self._get_custom_def('node_types'))
52 custom_defs.update(self._get_custom_def('capability_types'))
55 def test_description(self):
56 expected_desc = 'Template of a database including its hosting stack.'
57 self.assertEqual(expected_desc, self.topo.description)
59 def test_inputs(self):
61 ['mq_server_ip', 'my_cpus', 'receiver_port'],
62 sorted([input.name for input in self.topo.inputs]))
64 input_name = "receiver_port"
65 expected_description = "Port to be used for receiving messages."
66 for input in self.topo.inputs:
67 if input.name == input_name:
68 self.assertEqual(expected_description, input.description)
70 def test_node_tpls(self):
71 '''Test nodetemplate names.'''
73 ['app', 'server', 'websrv'],
74 sorted([tpl.name for tpl in self.topo.nodetemplates]))
77 expected_type = "example.SomeApp"
78 expected_properties = ['admin_user', 'pool_size']
79 expected_capabilities = ['feature', 'message_receiver']
80 expected_requirements = [{'host': {'node': 'websrv'}}]
81 expected_relationshp = ['tosca.relationships.HostedOn']
82 expected_host = ['websrv']
83 for tpl in self.topo.nodetemplates:
84 if tpl_name == tpl.name:
86 self.assertEqual(tpl.type, expected_type)
88 '''Test properties.'''
91 sorted(tpl.get_properties().keys()))
93 '''Test capabilities.'''
95 expected_capabilities,
96 sorted(tpl.get_capabilities().keys()))
98 '''Test requirements.'''
100 expected_requirements, tpl.requirements)
102 '''Test relationship.'''
103 ''' TODO : skip tempororily. need to fix it
106 expected_relationshp,
107 [x.type for x in tpl.relationships.keys()])
110 [y.name for y in tpl.relationships.values()])
111 '''Test interfaces.'''
112 # TODO(hurf) add interface test when new template is available
114 if tpl.name == 'server':
115 '''Test property value'''
116 props = tpl.get_properties()
117 if props and 'mem_size' in props.keys():
118 self.assertEqual(props['mem_size'].value, '4096 MB')
119 '''Test capability'''
120 caps = tpl.get_capabilities()
121 self.assertIn('os', caps.keys())
125 if caps and 'os' in caps.keys():
126 capability = caps['os']
127 os_props_objs = capability.get_properties_objects()
128 os_props = capability.get_properties()
129 os_type_prop = capability.get_property_value('type')
133 [p.value for p in os_props_objs if p.name == 'type'])
136 os_props['type'].value if 'type' in os_props else '')
137 self.assertEqual('Linux', os_props['type'].value)
138 self.assertEqual('Linux', os_type_prop)
140 def test_outputs(self):
143 sorted([output.name for output in self.topo.outputs]))
145 def test_groups(self):
146 group = self.topo.groups[0]
147 self.assertEqual('webserver_group', group.name)
148 self.assertEqual(['websrv', 'server'], group.members)
149 for node in group.get_member_nodes():
150 if node.name == 'server':
151 '''Test property value'''
152 props = node.get_properties()
153 if props and 'mem_size' in props.keys():
154 self.assertEqual(props['mem_size'].value, '4096 MB')
156 def test_system_template(self):
157 tpl_path = os.path.join(
158 os.path.dirname(os.path.abspath(__file__)),
159 "data/topology_template/system.yaml")
160 system_tosca_template = ToscaTemplate(tpl_path)
161 self.assertIsNotNone(system_tosca_template)
162 self.assertEqual(len(system_tosca_template.nested_tosca_templates), 3)