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.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
  22
  23 YAML_LOADER = toscaparser.utils.yamlparser.load_yaml
  24
  25
  26 class TopologyTemplateTest(TestCase):
  27
  28     def setUp(self):
  29         TestCase.setUp(self)
  30         '''TOSCA template.'''
  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())
  39
  40     def _get_custom_def(self, type_definition):
  41         custom_defs = {}
  42         for definition in self.imports:
  43             if os.path.isabs(definition):
  44                 def_file = definition
  45             else:
  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))
  50         return custom_defs
  51
  52     def _get_all_custom_def(self):
  53         custom_defs = {}
  54         custom_defs.update(self._get_custom_def('node_types'))
  55         custom_defs.update(self._get_custom_def('capability_types'))
  56         return custom_defs
  57
  58     def test_description(self):
  59         expected_desc = 'Template of a database including its hosting stack.'
  60         self.assertEqual(expected_desc, self.topo.description)
  61
  62     def test_inputs(self):
  63         self.assertEqual(
  64             ['mq_server_ip', 'my_cpus', 'receiver_port'],
  65             sorted([input.name for input in self.topo.inputs]))
  66
  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)
  72
  73     def test_node_tpls(self):
  74         '''Test nodetemplate names.'''
  75         self.assertEqual(
  76             ['app', 'server', 'websrv'],
  77             sorted([tpl.name for tpl in self.topo.nodetemplates]))
  78
  79         tpl_name = "app"
  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:
  88                 '''Test node type.'''
  89                 self.assertEqual(tpl.type, expected_type)
  90
  91                 '''Test properties.'''
  92                 self.assertEqual(
  93                     expected_properties,
  94                     sorted(tpl.get_properties().keys()))
  95
  96                 '''Test capabilities.'''
  97                 self.assertEqual(
  98                     expected_capabilities,
  99                     sorted(tpl.get_capabilities().keys()))
 100
 101                 '''Test requirements.'''
 102                 self.assertEqual(
 103                     expected_requirements, tpl.requirements)
 104
 105                 '''Test relationship.'''
 106                 ''' TODO : skip tempororily. need to fix it
 107                 '''
 108                 self.assertEqual(
 109                     expected_relationshp,
 110                     [x.type for x in tpl.relationships.keys()])
 111                 self.assertEqual(
 112                     expected_host,
 113                     [y.name for y in tpl.relationships.values()])
 114                 '''Test interfaces.'''
 115                 # TODO(hurf) add interface test when new template is available
 116
 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())
 125                 os_props_objs = None
 126                 os_props = None
 127                 os_type_prop = None
 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')
 133                     break
 134                 self.assertEqual(
 135                     ['Linux'],
 136                     [p.value for p in os_props_objs if p.name == 'type'])
 137                 self.assertEqual(
 138                     'Linux',
 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)
 142
 143     def test_outputs(self):
 144         self.assertEqual(
 145             ['receiver_ip'],
 146             sorted([output.name for output in self.topo.outputs]))
 147
 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')
 158
 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)
 165         self.assertEqual(
 166             len(system_tosca_template.
 167                 nested_tosca_templates_with_topology), 4)
 168
 169     def test_invalid_keyname(self):
 170         tpl_snippet = '''
 171         substitution_mappings:
 172           node_type: example.DatabaseSubsystem
 173           capabilities:
 174             database_endpoint: [ db_app, database_endpoint ]
 175           requirements:
 176             receiver1: [ tran_app, receiver1 ]
 177           invalid_key: 123
 178         '''
 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,
 188                                          None, None, None))
 189         self.assertEqual(expected_message, err.__str__())
 190
 191     def test_missing_required_keyname(self):
 192         tpl_snippet = '''
 193         substitution_mappings:
 194           capabilities:
 195             database_endpoint: [ db_app, database_endpoint ]
 196           requirements:
 197             receiver1: [ tran_app, receiver1 ]
 198         '''
 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,
 206                                          None, None, None))
 207         self.assertEqual(expected_message, err.__str__())