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         self.tosca_system_tpl_path = os.path.join(
  38             os.path.dirname(os.path.abspath(__file__)),
  39             "data/topology_template/system.yaml")
  40         self.system_template = ToscaTemplate(self.tosca_system_tpl_path)
  41
  42     def _get_custom_def(self, type_definition):
  43         custom_defs = {}
  44         for definition in self.imports:
  45             if os.path.isabs(definition):
  46                 def_file = definition
  47             else:
  48                 tpl_dir = os.path.dirname(os.path.abspath(self.tosca_tpl_path))
  49                 def_file = os.path.join(tpl_dir, definition)
  50             custom_type = YAML_LOADER(def_file)
  51             custom_defs.update(custom_type.get(type_definition))
  52         return custom_defs
  53
  54     def _get_all_custom_def(self):
  55         custom_defs = {}
  56         custom_defs.update(self._get_custom_def('node_types'))
  57         custom_defs.update(self._get_custom_def('capability_types'))
  58         return custom_defs
  59
  60     def test_description(self):
  61         expected_desc = 'Template of a database including its hosting stack.'
  62         self.assertEqual(expected_desc, self.topo.description)
  63
  64     def test_inputs(self):
  65         self.assertEqual(
  66             ['mq_server_ip', 'my_cpus', 'receiver_port'],
  67             sorted([input.name for input in self.topo.inputs]))
  68
  69         input_name = "receiver_port"
  70         expected_description = "Port to be used for receiving messages."
  71         for input in self.topo.inputs:
  72             if input.name == input_name:
  73                 self.assertEqual(expected_description, input.description)
  74
  75     def test_node_tpls(self):
  76         '''Test nodetemplate names.'''
  77         self.assertEqual(
  78             ['app', 'server', 'websrv'],
  79             sorted([tpl.name for tpl in self.topo.nodetemplates]))
  80
  81         tpl_name = "app"
  82         expected_type = "example.SomeApp"
  83         expected_properties = ['admin_user', 'pool_size']
  84         expected_capabilities = ['feature', 'message_receiver']
  85         expected_requirements = [{'host': {'node': 'websrv'}}]
  86         expected_relationshp = ['tosca.relationships.HostedOn']
  87         expected_host = ['websrv']
  88         for tpl in self.topo.nodetemplates:
  89             if tpl_name == tpl.name:
  90                 '''Test node type.'''
  91                 self.assertEqual(tpl.type, expected_type)
  92
  93                 '''Test properties.'''
  94                 self.assertEqual(
  95                     expected_properties,
  96                     sorted(tpl.get_properties().keys()))
  97
  98                 '''Test capabilities.'''
  99                 self.assertEqual(
 100                     expected_capabilities,
 101                     sorted(tpl.get_capabilities().keys()))
 102
 103                 '''Test requirements.'''
 104                 self.assertEqual(
 105                     expected_requirements, tpl.requirements)
 106
 107                 '''Test relationship.'''
 108                 ''' TODO : skip tempororily. need to fix it
 109                 '''
 110                 self.assertEqual(
 111                     expected_relationshp,
 112                     [x.type for x in tpl.relationships.keys()])
 113                 self.assertEqual(
 114                     expected_host,
 115                     [y.name for y in tpl.relationships.values()])
 116                 '''Test interfaces.'''
 117                 # TODO(hurf) add interface test when new template is available
 118
 119             if tpl.name == 'server':
 120                 '''Test property value'''
 121                 props = tpl.get_properties()
 122                 if props and 'mem_size' in props.keys():
 123                     self.assertEqual(props['mem_size'].value, '4096 MB')
 124                 '''Test capability'''
 125                 caps = tpl.get_capabilities()
 126                 self.assertIn('os', caps.keys())
 127                 os_props_objs = None
 128                 os_props = None
 129                 os_type_prop = None
 130                 if caps and 'os' in caps.keys():
 131                     capability = caps['os']
 132                     os_props_objs = capability.get_properties_objects()
 133                     os_props = capability.get_properties()
 134                     os_type_prop = capability.get_property_value('type')
 135                     break
 136                 self.assertEqual(
 137                     ['Linux'],
 138                     [p.value for p in os_props_objs if p.name == 'type'])
 139                 self.assertEqual(
 140                     'Linux',
 141                     os_props['type'].value if 'type' in os_props else '')
 142                 self.assertEqual('Linux', os_props['type'].value)
 143                 self.assertEqual('Linux', os_type_prop)
 144
 145     def test_outputs(self):
 146         self.assertEqual(
 147             ['receiver_ip'],
 148             sorted([output.name for output in self.topo.outputs]))
 149
 150     def test_groups(self):
 151         group = self.topo.groups[0]
 152         self.assertEqual('webserver_group', group.name)
 153         self.assertEqual(['websrv', 'server'], group.members)
 154         for node in group.get_member_nodes():
 155             if node.name == 'server':
 156                 '''Test property value'''
 157                 props = node.get_properties()
 158                 if props and 'mem_size' in props.keys():
 159                     self.assertEqual(props['mem_size'].value, '4096 MB')
 160
 161     def test_system_template(self):
 162         tpl_path = os.path.join(
 163             os.path.dirname(os.path.abspath(__file__)),
 164             "data/topology_template/system.yaml")
 165         self.assertIsNotNone(ToscaTemplate(tpl_path))