tosca2heat/tosca-parser/toscaparser/tests/test_functions.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 import six
  15 from toscaparser.common import exception
  16 from toscaparser import functions
  17 from toscaparser.tests.base import TestCase
  18 from toscaparser.tosca_template import ToscaTemplate
  19 from toscaparser.utils.gettextutils import _
  20
  21
  22 class IntrinsicFunctionsTest(TestCase):
  23
  24     tosca_tpl = os.path.join(
  25         os.path.dirname(os.path.abspath(__file__)),
  26         "data/tosca_single_instance_wordpress.yaml")
  27     params = {'db_name': 'my_wordpress', 'db_user': 'my_db_user',
  28               'db_root_pwd': '12345678'}
  29     tosca = ToscaTemplate(tosca_tpl, parsed_params=params)
  30
  31     def _get_node(self, node_name, tosca=None):
  32         if tosca is None:
  33             tosca = self.tosca
  34         return [
  35             node for node in tosca.nodetemplates
  36             if node.name == node_name][0]
  37
  38     def _get_operation(self, interfaces, operation):
  39         return [
  40             interface for interface in interfaces
  41             if interface.name == operation][0]
  42
  43     def _get_property(self, node_template, property_name):
  44         return [prop.value for prop in node_template.get_properties_objects()
  45                 if prop.name == property_name][0]
  46
  47     def _get_inputs_dict(self):
  48         inputs = {}
  49         for input in self.tosca.inputs:
  50             inputs[input.name] = input.default
  51         return inputs
  52
  53     def _get_input(self, name):
  54         self._get_inputs_dict()[name]
  55
  56     def test_get_property(self):
  57         wordpress = self._get_node('wordpress')
  58         operation = self._get_operation(wordpress.interfaces, 'configure')
  59         wp_db_password = operation.inputs['wp_db_password']
  60         self.assertIsInstance(wp_db_password, functions.GetProperty)
  61         result = wp_db_password.result()
  62         self.assertEqual('wp_pass', result)
  63
  64     def test_get_property_with_input_param(self):
  65         wordpress = self._get_node('wordpress')
  66         operation = self._get_operation(wordpress.interfaces, 'configure')
  67         wp_db_user = operation.inputs['wp_db_user']
  68         self.assertIsInstance(wp_db_user, functions.GetProperty)
  69         result = wp_db_user.result()
  70         self.assertEqual('my_db_user', result)
  71
  72     def test_unknown_capability_property(self):
  73         self.assertRaises(exception.ValidationError, self._load_template,
  74                           'functions/test_unknown_capability_property.yaml')
  75         exception.ExceptionCollector.assertExceptionMessage(
  76             KeyError,
  77             _('\'Property "unknown" was not found in capability '
  78               '"database_endpoint" of node template "database" referenced '
  79               'from node template "database".\''))
  80
  81     def test_get_input_in_properties(self):
  82         mysql_dbms = self._get_node('mysql_dbms')
  83         expected_inputs = ['db_root_pwd', 'db_port']
  84         props = mysql_dbms.get_properties()
  85         for key in props.keys():
  86             prop = props[key]
  87             self.assertIsInstance(prop.value, functions.GetInput)
  88             expected_inputs.remove(prop.value.input_name)
  89         self.assertListEqual(expected_inputs, [])
  90
  91     def test_get_input_validation(self):
  92         self.assertRaises(
  93             exception.ValidationError, self._load_template,
  94             'functions/test_unknown_input_in_property.yaml')
  95         exception.ExceptionCollector.assertExceptionMessage(
  96             exception.UnknownInputError,
  97             _('Unknown input "objectstore_name".'))
  98
  99         self.assertRaises(
 100             exception.ValidationError, self._load_template,
 101             'functions/test_unknown_input_in_interface.yaml')
 102         exception.ExceptionCollector.assertExceptionMessage(
 103             exception.UnknownInputError,
 104             _('Unknown input "image_id".'))
 105
 106         self.assertRaises(
 107             exception.ValidationError, self._load_template,
 108             'functions/test_invalid_function_signature.yaml')
 109         exception.ExceptionCollector.assertExceptionMessage(
 110             ValueError,
 111             _('Expected one argument for function "get_input" but received '
 112               '"[\'cpus\', \'cpus\']".'))
 113
 114     def test_get_input_default_value_result(self):
 115         mysql_dbms = self._get_node('mysql_dbms')
 116         dbms_port = self._get_property(mysql_dbms, 'port')
 117         self.assertEqual(3306, dbms_port.result())
 118         dbms_root_password = self._get_property(mysql_dbms,
 119                                                 'root_password')
 120         self.assertEqual(dbms_root_password.result(), '12345678')
 121
 122     def test_get_property_with_host(self):
 123         tosca_tpl = os.path.join(
 124             os.path.dirname(os.path.abspath(__file__)),
 125             "data/functions/test_get_property_with_host.yaml")
 126         mysql_database = self._get_node('mysql_database',
 127                                         ToscaTemplate(tosca_tpl,
 128                                                       parsed_params={
 129                                                           'db_root_pwd': '123'
 130                                                       }))
 131         operation = self._get_operation(mysql_database.interfaces, 'configure')
 132         db_port = operation.inputs['db_port']
 133         self.assertIsInstance(db_port, functions.GetProperty)
 134         result = db_port.result()
 135         self.assertEqual(3306, result)
 136         test = operation.inputs['test']
 137         self.assertIsInstance(test, functions.GetProperty)
 138         result = test.result()
 139         self.assertEqual(1, result)
 140
 141     def test_get_property_with_nested_params(self):
 142         tosca_tpl = os.path.join(
 143             os.path.dirname(os.path.abspath(__file__)),
 144             "data/functions/tosca_nested_property_names_indexes.yaml")
 145         webserver = self._get_node('wordpress',
 146                                    ToscaTemplate(tosca_tpl,
 147                                                  parsed_params={
 148                                                      'db_root_pwd': '1234'}))
 149         operation = self._get_operation(webserver.interfaces, 'configure')
 150         wp_endpoint_prot = operation.inputs['wp_endpoint_protocol']
 151         self.assertIsInstance(wp_endpoint_prot, functions.GetProperty)
 152         self.assertEqual('tcp', wp_endpoint_prot.result())
 153         wp_list_prop = operation.inputs['wp_list_prop']
 154         self.assertIsInstance(wp_list_prop, functions.GetProperty)
 155         self.assertEqual(3, wp_list_prop.result())
 156
 157     def test_get_property_with_capabilties_inheritance(self):
 158         tosca_tpl = os.path.join(
 159             os.path.dirname(os.path.abspath(__file__)),
 160             "data/functions/test_capabilties_inheritance.yaml")
 161         some_node = self._get_node('some_node',
 162                                    ToscaTemplate(tosca_tpl,
 163                                                  parsed_params={
 164                                                      'db_root_pwd': '1234'}))
 165         operation = self._get_operation(some_node.interfaces, 'configure')
 166         some_input = operation.inputs['some_input']
 167         self.assertIsInstance(some_input, functions.GetProperty)
 168         self.assertEqual('someval', some_input.result())
 169
 170     def test_get_property_source_target_keywords(self):
 171         tosca_tpl = os.path.join(
 172             os.path.dirname(os.path.abspath(__file__)),
 173             "data/functions/test_get_property_source_target_keywords.yaml")
 174         tosca = ToscaTemplate(tosca_tpl,
 175                               parsed_params={'db_root_pwd': '1234'})
 176
 177         for node in tosca.nodetemplates:
 178             for relationship, trgt in node.relationships.items():
 179                 rel_template = trgt.get_relationship_template()[0]
 180                 break
 181
 182         operation = self._get_operation(rel_template.interfaces,
 183                                         'pre_configure_source')
 184         target_test = operation.inputs['target_test']
 185         self.assertIsInstance(target_test, functions.GetProperty)
 186         self.assertEqual(1, target_test.result())
 187         source_port = operation.inputs['source_port']
 188         self.assertIsInstance(source_port, functions.GetProperty)
 189         self.assertEqual(3306, source_port.result())
 190
 191     def test_get_prop_cap_host(self):
 192         tosca_tpl = os.path.join(
 193             os.path.dirname(os.path.abspath(__file__)),
 194             "data/functions/test_get_prop_cap_host.yaml")
 195         some_node = self._get_node('some_node',
 196                                    ToscaTemplate(tosca_tpl))
 197         some_prop = some_node.get_properties()['some_prop']
 198         self.assertIsInstance(some_prop.value, functions.GetProperty)
 199         self.assertEqual('someval', some_prop.value.result())
 200
 201     def test_get_prop_cap_bool(self):
 202         tosca_tpl = os.path.join(
 203             os.path.dirname(os.path.abspath(__file__)),
 204             "data/functions/test_get_prop_cap_bool.yaml")
 205         some_node = self._get_node('software',
 206                                    ToscaTemplate(tosca_tpl))
 207         some_prop = some_node.get_properties()['some_prop']
 208         self.assertIsInstance(some_prop.value, functions.GetProperty)
 209         self.assertEqual(False, some_prop.value.result())
 210
 211
 212 class GetAttributeTest(TestCase):
 213
 214     def _load_template(self, filename):
 215         return ToscaTemplate(os.path.join(
 216             os.path.dirname(os.path.abspath(__file__)),
 217             'data',
 218             filename),
 219             parsed_params={'db_root_pwd': '1234'})
 220
 221     def _get_operation(self, interfaces, operation):
 222         return [
 223             interface for interface in interfaces
 224             if interface.name == operation][0]
 225
 226     def test_get_attribute_in_outputs(self):
 227         tpl = self._load_template('tosca_single_instance_wordpress.yaml')
 228         website_url_output = [
 229             x for x in tpl.outputs if x.name == 'website_url'][0]
 230         self.assertIsInstance(website_url_output.value, functions.GetAttribute)
 231         self.assertEqual('server', website_url_output.value.node_template_name)
 232         self.assertEqual('private_address',
 233                          website_url_output.value.attribute_name)
 234
 235     def test_get_attribute_invalid_args(self):
 236         expected_msg = _('Illegal arguments for function "get_attribute".'
 237                          ' Expected arguments: "node-template-name", '
 238                          '"req-or-cap"(optional), "property name"')
 239         err = self.assertRaises(ValueError,
 240                                 functions.get_function, None, None,
 241                                 {'get_attribute': []})
 242         self.assertIn(expected_msg, six.text_type(err))
 243         err = self.assertRaises(ValueError,
 244                                 functions.get_function, None, None,
 245                                 {'get_attribute': ['x']})
 246         self.assertIn(expected_msg, six.text_type(err))
 247
 248     def test_get_attribute_unknown_node_template_name(self):
 249         self.assertRaises(
 250             exception.ValidationError, self._load_template,
 251             'functions/test_get_attribute_unknown_node_template_name.yaml')
 252         exception.ExceptionCollector.assertExceptionMessage(
 253             KeyError,
 254             _('\'Node template "unknown_node_template" was not found.\''))
 255
 256     def test_get_attribute_unknown_attribute(self):
 257         self.assertRaises(
 258             exception.ValidationError, self._load_template,
 259             'functions/test_get_attribute_unknown_attribute_name.yaml')
 260         exception.ExceptionCollector.assertExceptionMessage(
 261             KeyError,
 262             _('\'Attribute "unknown_attribute" was not found in node template '
 263               '"server".\''))
 264
 265     def test_get_attribute_host_keyword(self):
 266         tpl = self._load_template(
 267             'functions/test_get_attribute_host_keyword.yaml')
 268
 269         def assert_get_attribute_host_functionality(node_template_name):
 270             node = [x for x in tpl.nodetemplates
 271                     if x.name == node_template_name][0]
 272             configure_op = [
 273                 x for x in node.interfaces if x.name == 'configure'][0]
 274             ip_addr_input = configure_op.inputs['ip_address']
 275             self.assertIsInstance(ip_addr_input, functions.GetAttribute)
 276             self.assertEqual('server',
 277                              ip_addr_input.get_referenced_node_template().name)
 278
 279         assert_get_attribute_host_functionality('dbms')
 280         assert_get_attribute_host_functionality('database')
 281
 282     def test_get_attribute_host_not_found(self):
 283         self.assertRaises(
 284             exception.ValidationError, self._load_template,
 285             'functions/test_get_attribute_host_not_found.yaml')
 286         exception.ExceptionCollector.assertExceptionMessage(
 287             ValueError,
 288             _('"get_attribute: [ HOST, ... ]" was used in node template '
 289               '"server" but "tosca.relationships.HostedOn" was not found in '
 290               'the relationship chain.'))
 291
 292     def test_get_attribute_illegal_host_in_outputs(self):
 293         self.assertRaises(
 294             exception.ValidationError, self._load_template,
 295             'functions/test_get_attribute_illegal_host_in_outputs.yaml')
 296         exception.ExceptionCollector.assertExceptionMessage(
 297             ValueError,
 298             _('"get_attribute: [ HOST, ... ]" is not allowed in "outputs" '
 299               'section of the TOSCA template.'))
 300
 301     def test_get_attribute_with_index(self):
 302         self._load_template(
 303             'functions/test_get_attribute_with_index.yaml')
 304
 305     def test_get_attribute_with_index_error(self):
 306         self.assertRaises(
 307             exception.ValidationError, self._load_template,
 308             'functions/test_get_attribute_with_index_error.yaml')
 309         exception.ExceptionCollector.assertExceptionMessage(
 310             ValueError,
 311             _('Illegal arguments for function "get_attribute". '
 312               'Unexpected attribute/index value "0"'))
 313
 314     def test_get_attribute_source_target_keywords(self):
 315         tosca_tpl = os.path.join(
 316             os.path.dirname(os.path.abspath(__file__)),
 317             "data/functions/test_get_attribute_source_target_keywords.yaml")
 318         tosca = ToscaTemplate(tosca_tpl,
 319                               parsed_params={'db_root_pwd': '12345678'})
 320
 321         for node in tosca.nodetemplates:
 322             for relationship, trgt in node.relationships.items():
 323                 rel_template = trgt.get_relationship_template()[0]
 324                 break
 325
 326         operation = self._get_operation(rel_template.interfaces,
 327                                         'pre_configure_source')
 328         target_test = operation.inputs['target_test']
 329         self.assertIsInstance(target_test, functions.GetAttribute)
 330         source_port = operation.inputs['source_port']
 331         self.assertIsInstance(source_port, functions.GetAttribute)
 332
 333     def test_get_attribute_with_nested_params(self):
 334         self._load_template(
 335             'functions/test_get_attribute_with_nested_params.yaml')
 336
 337     def test_implicit_attribute(self):
 338         self.assertIsNotNone(self._load_template(
 339             'functions/test_get_implicit_attribute.yaml'))
 340
 341     def test_get_attribute_capability_inheritance(self):
 342         self.assertIsNotNone(self._load_template(
 343             'functions/test_container_cap_child.yaml'))
 344
 345
 346 class ConcatTest(TestCase):
 347
 348     def _load_template(self, filename):
 349         return ToscaTemplate(os.path.join(
 350             os.path.dirname(os.path.abspath(__file__)),
 351             filename))
 352
 353     def test_validate_concat(self):
 354         tosca = self._load_template("data/functions/test_concat.yaml")
 355         server_url_output = [
 356             output for output in tosca.outputs if output.name == 'url'][0]
 357         func = functions.get_function(self, tosca.outputs,
 358                                       server_url_output.value)
 359         self.assertIsInstance(func, functions.Concat)
 360
 361         self.assertRaises(exception.ValidationError, self._load_template,
 362                           'data/functions/test_concat_invalid.yaml')
 363         exception.ExceptionCollector.assertExceptionMessage(
 364             ValueError,
 365             _('Invalid arguments for function "concat". Expected at least '
 366               'one arguments.'))
 367
 368
 369 class TokenTest(TestCase):
 370
 371     def _load_template(self, filename):
 372         return ToscaTemplate(os.path.join(
 373             os.path.dirname(os.path.abspath(__file__)),
 374             filename))
 375
 376     def test_validate_token(self):
 377         tosca = self._load_template("data/functions/test_token.yaml")
 378         server_url_output = [
 379             output for output in tosca.outputs if output.name == 'url'][0]
 380         func = functions.get_function(self, tosca.outputs,
 381                                       server_url_output.value)
 382         self.assertIsInstance(func, functions.Token)
 383
 384         self.assertRaises(exception.ValidationError, self._load_template,
 385                           'data/functions/test_token_invalid.yaml')
 386         exception.ExceptionCollector.assertExceptionMessage(
 387             ValueError,
 388             _('Invalid arguments for function "token". Expected at least '
 389               'three arguments.'))
 390         exception.ExceptionCollector.assertExceptionMessage(
 391             ValueError,
 392             _('Invalid arguments for function "token". Expected '
 393               'integer value as third argument.'))
 394         exception.ExceptionCollector.assertExceptionMessage(
 395             ValueError,
 396             _('Invalid arguments for function "token". Expected '
 397               'single char value as second argument.'))