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 _get_custom_types(self):
  59         custom_types = {}
  60         def_file = os.path.join(
  61             os.path.dirname(os.path.abspath(__file__)),
  62             "data/topology_template/definitions.yaml")
  63         custom_type = YAML_LOADER(def_file)
  64         node_types = custom_type['node_types']
  65         for name in node_types:
  66             defintion = node_types[name]
  67             custom_types[name] = defintion
  68         return custom_types
  69
  70     def test_description(self):
  71         expected_desc = 'Template of a database including its hosting stack.'
  72         self.assertEqual(expected_desc, self.topo.description)
  73
  74     def test_inputs(self):
  75         self.assertEqual(
  76             ['mq_server_ip', 'my_cpus', 'receiver_port'],
  77             sorted([input.name for input in self.topo.inputs]))
  78
  79         input_name = "receiver_port"
  80         expected_description = "Port to be used for receiving messages."
  81         for input in self.topo.inputs:
  82             if input.name == input_name:
  83                 self.assertEqual(expected_description, input.description)
  84
  85     def test_node_tpls(self):
  86         '''Test nodetemplate names.'''
  87         self.assertEqual(
  88             ['app', 'server', 'websrv'],
  89             sorted([tpl.name for tpl in self.topo.nodetemplates]))
  90
  91         tpl_name = "app"
  92         expected_type = "example.SomeApp"
  93         expected_properties = ['admin_user', 'pool_size']
  94         expected_capabilities = ['feature', 'message_receiver']
  95         expected_requirements = [{'host': {'node': 'websrv'}}]
  96         expected_relationshp = ['tosca.relationships.HostedOn']
  97         expected_host = ['websrv']
  98         for tpl in self.topo.nodetemplates:
  99             if tpl_name == tpl.name:
 100                 '''Test node type.'''
 101                 self.assertEqual(tpl.type, expected_type)
 102
 103                 '''Test properties.'''
 104                 self.assertEqual(
 105                     expected_properties,
 106                     sorted(tpl.get_properties().keys()))
 107
 108                 '''Test capabilities.'''
 109                 self.assertEqual(
 110                     expected_capabilities,
 111                     sorted(tpl.get_capabilities().keys()))
 112
 113                 '''Test requirements.'''
 114                 self.assertEqual(
 115                     expected_requirements, tpl.requirements)
 116
 117                 '''Test relationship.'''
 118                 ''' TODO : skip tempororily. need to fix it
 119                 '''
 120                 self.assertEqual(
 121                     expected_relationshp,
 122                     [x.type for x in tpl.relationships.keys()])
 123                 self.assertEqual(
 124                     expected_host,
 125                     [y.name for y in tpl.relationships.values()])
 126                 '''Test interfaces.'''
 127                 # TODO(hurf) add interface test when new template is available
 128
 129             if tpl.name == 'server':
 130                 '''Test property value'''
 131                 props = tpl.get_properties()
 132                 if props and 'mem_size' in props.keys():
 133                     self.assertEqual(props['mem_size'].value, '4096 MB')
 134                 '''Test capability'''
 135                 caps = tpl.get_capabilities()
 136                 self.assertIn('os', caps.keys())
 137                 os_props_objs = None
 138                 os_props = None
 139                 os_type_prop = None
 140                 if caps and 'os' in caps.keys():
 141                     capability = caps['os']
 142                     os_props_objs = capability.get_properties_objects()
 143                     os_props = capability.get_properties()
 144                     os_type_prop = capability.get_property_value('type')
 145                     break
 146                 self.assertEqual(
 147                     ['Linux'],
 148                     [p.value for p in os_props_objs if p.name == 'type'])
 149                 self.assertEqual(
 150                     'Linux',
 151                     os_props['type'].value if 'type' in os_props else '')
 152                 self.assertEqual('Linux', os_props['type'].value)
 153                 self.assertEqual('Linux', os_type_prop)
 154
 155     def test_outputs(self):
 156         self.assertEqual(
 157             sorted(['receiver_ip', 'receiver_port']),
 158             sorted([output.name for output in self.topo.outputs]))
 159
 160     def test_groups(self):
 161         group = self.topo.groups[0]
 162         self.assertEqual('webserver_group', group.name)
 163         self.assertEqual(['websrv', 'server'], group.members)
 164         for node in group.get_member_nodes():
 165             if node.name == 'server':
 166                 '''Test property value'''
 167                 props = node.get_properties()
 168                 if props and 'mem_size' in props.keys():
 169                     self.assertEqual(props['mem_size'].value, '4096 MB')
 170
 171     def test_system_template(self):
 172         tpl_path = os.path.join(
 173             os.path.dirname(os.path.abspath(__file__)),
 174             "data/topology_template/system.yaml")
 175         system_tosca_template = ToscaTemplate(tpl_path)
 176         self.assertIsNotNone(system_tosca_template)
 177         self.assertEqual(
 178             len(system_tosca_template.
 179                 nested_tosca_templates_with_topology), 4)
 180         self.assertTrue(system_tosca_template.has_nested_templates())
 181
 182     def test_invalid_keyname(self):
 183         tpl_snippet = '''
 184         substitution_mappings:
 185           node_type: example.DatabaseSubsystem
 186           capabilities:
 187             database_endpoint: [ db_app, database_endpoint ]
 188           requirements:
 189             receiver1: [ tran_app, receiver1 ]
 190           invalid_key: 123
 191         '''
 192         sub_mappings = (toscaparser.utils.yamlparser.
 193                         simple_parse(tpl_snippet))['substitution_mappings']
 194         expected_message = _(
 195             'SubstitutionMappings contains unknown field '
 196             '"invalid_key". Refer to the definition '
 197             'to verify valid values.')
 198         err = self.assertRaises(
 199             exception.UnknownFieldError,
 200             lambda: SubstitutionMappings(sub_mappings, None, None,
 201                                          None, None, None))
 202         self.assertEqual(expected_message, err.__str__())
 203
 204     def test_missing_required_keyname(self):
 205         tpl_snippet = '''
 206         substitution_mappings:
 207           capabilities:
 208             database_endpoint: [ db_app, database_endpoint ]
 209           requirements:
 210             receiver1: [ tran_app, receiver1 ]
 211         '''
 212         sub_mappings = (toscaparser.utils.yamlparser.
 213                         simple_parse(tpl_snippet))['substitution_mappings']
 214         expected_message = _('SubstitutionMappings used in topology_template '
 215                              'is missing required field "node_type".')
 216         err = self.assertRaises(
 217             exception.MissingRequiredFieldError,
 218             lambda: SubstitutionMappings(sub_mappings, None, None,
 219                                          None, None, None))
 220         self.assertEqual(expected_message, err.__str__())
 221
 222     def test_invalid_nodetype(self):
 223             tpl_snippet = '''
 224             substitution_mappings:
 225               node_type: example.DatabaseSubsystem1
 226               capabilities:
 227                 database_endpoint: [ db_app, database_endpoint ]
 228               requirements:
 229                 receiver1: [ tran_app, receiver1 ]
 230             '''
 231             sub_mappings = (toscaparser.utils.yamlparser.
 232                             simple_parse(tpl_snippet))['substitution_mappings']
 233             custom_defs = self._get_custom_types()
 234             expected_message = _('Node type "example.DatabaseSubsystem1" '
 235                                  'is not a valid type.')
 236             err = self.assertRaises(
 237                 exception.InvalidNodeTypeError,
 238                 lambda: SubstitutionMappings(sub_mappings, None, None,
 239                                              None, None, custom_defs))
 240             self.assertEqual(expected_message, err.__str__())
 241
 242     def test_system_with_input_validation(self):
 243             tpl_path0 = os.path.join(
 244                 os.path.dirname(os.path.abspath(__file__)),
 245                 "data/topology_template/validate/system_invalid_input.yaml")
 246             tpl_path1 = os.path.join(
 247                 os.path.dirname(os.path.abspath(__file__)),
 248                 "data/topology_template/validate/"
 249                 "queuingsubsystem_invalid_input.yaml")
 250             errormsg = _('SubstitutionMappings with node_type '
 251                          'example.QueuingSubsystem is missing '
 252                          'required input definition of input "server_port".')
 253
 254             # It's invalid in nested template.
 255             self.assertRaises(exception.ValidationError,
 256                               lambda: ToscaTemplate(tpl_path0))
 257             exception.ExceptionCollector.assertExceptionMessage(
 258                 exception.MissingRequiredInputError, errormsg)
 259
 260             # Subtemplate deploy standaolone is also invalid.
 261             self.assertRaises(exception.ValidationError,
 262                               lambda: ToscaTemplate(tpl_path1))
 263             exception.ExceptionCollector.assertExceptionMessage(
 264                 exception.MissingRequiredInputError, errormsg)