tosca2heat/tosca-parser/toscaparser/tests/test_topology_template.py

   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
   4 #
   5 #         http://www.apache.org/licenses/LICENSE-2.0
   6 #
   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
  11 #    under the License.
  12
  13 import os
  14
  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
  19
  20 YAML_LOADER = toscaparser.utils.yamlparser.load_yaml
  21
  22
  23 class TopologyTemplateTest(TestCase):
  24
  25     def setUp(self):
  26         TestCase.setUp(self)
  27         '''TOSCA template.'''
  28         self.tosca_tpl_path = os.path.join(
  29             os.path.dirname(os.path.abspath(__file__)),
  30             "data/topology_template/subsystem.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())
  36
  37     def _get_custom_def(self, type_definition):
  38         custom_defs = {}
  39         for definition in self.imports:
  40             if os.path.isabs(definition):
  41                 def_file = definition
  42             else:
  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))
  47         return custom_defs
  48
  49     def _get_all_custom_def(self):
  50         custom_defs = {}
  51         custom_defs.update(self._get_custom_def('node_types'))
  52         custom_defs.update(self._get_custom_def('capability_types'))
  53         return custom_defs
  54
  55     def test_description(self):
  56         expected_desc = 'Template of a database including its hosting stack.'
  57         self.assertEqual(expected_desc, self.topo.description)
  58
  59     def test_inputs(self):
  60         self.assertEqual(
  61             ['mq_server_ip', 'my_cpus', 'receiver_port'],
  62             sorted([input.name for input in self.topo.inputs]))
  63
  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)
  69
  70     def test_node_tpls(self):
  71         '''Test nodetemplate names.'''
  72         self.assertEqual(
  73             ['app', 'server', 'websrv'],
  74             sorted([tpl.name for tpl in self.topo.nodetemplates]))
  75
  76         tpl_name = "app"
  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:
  85                 '''Test node type.'''
  86                 self.assertEqual(tpl.type, expected_type)
  87
  88                 '''Test properties.'''
  89                 self.assertEqual(
  90                     expected_properties,
  91                     sorted(tpl.get_properties().keys()))
  92
  93                 '''Test capabilities.'''
  94                 self.assertEqual(
  95                     expected_capabilities,
  96                     sorted(tpl.get_capabilities().keys()))
  97
  98                 '''Test requirements.'''
  99                 self.assertEqual(
 100                     expected_requirements, tpl.requirements)
 101
 102                 '''Test relationship.'''
 103                 ''' TODO : skip tempororily. need to fix it
 104                 '''
 105                 self.assertEqual(
 106                     expected_relationshp,
 107                     [x.type for x in tpl.relationships.keys()])
 108                 self.assertEqual(
 109                     expected_host,
 110                     [y.name for y in tpl.relationships.values()])
 111                 '''Test interfaces.'''
 112                 # TODO(hurf) add interface test when new template is available
 113
 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())
 122                 os_props_objs = None
 123                 os_props = None
 124                 os_type_prop = None
 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')
 130                     break
 131                 self.assertEqual(
 132                     ['Linux'],
 133                     [p.value for p in os_props_objs if p.name == 'type'])
 134                 self.assertEqual(
 135                     'Linux',
 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)
 139
 140     def test_outputs(self):
 141         self.assertEqual(
 142             ['receiver_ip'],
 143             sorted([output.name for output in self.topo.outputs]))
 144
 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')
 155
 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         self.assertIsNotNone(ToscaTemplate(tpl_path))