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
 192 class GetAttributeTest(TestCase):
 193
 194     def _load_template(self, filename):
 195         return ToscaTemplate(os.path.join(
 196             os.path.dirname(os.path.abspath(__file__)),
 197             'data',
 198             filename),
 199             parsed_params={'db_root_pwd': '1234'})
 200
 201     def _get_operation(self, interfaces, operation):
 202         return [
 203             interface for interface in interfaces
 204             if interface.name == operation][0]
 205
 206     def test_get_attribute_in_outputs(self):
 207         tpl = self._load_template('tosca_single_instance_wordpress.yaml')
 208         website_url_output = [
 209             x for x in tpl.outputs if x.name == 'website_url'][0]
 210         self.assertIsInstance(website_url_output.value, functions.GetAttribute)
 211         self.assertEqual('server', website_url_output.value.node_template_name)
 212         self.assertEqual('private_address',
 213                          website_url_output.value.attribute_name)
 214
 215     def test_get_attribute_invalid_args(self):
 216         expected_msg = _('Illegal arguments for function "get_attribute".'
 217                          ' Expected arguments: "node-template-name", '
 218                          '"req-or-cap"(optional), "property name"')
 219         err = self.assertRaises(ValueError,
 220                                 functions.get_function, None, None,
 221                                 {'get_attribute': []})
 222         self.assertIn(expected_msg, six.text_type(err))
 223         err = self.assertRaises(ValueError,
 224                                 functions.get_function, None, None,
 225                                 {'get_attribute': ['x']})
 226         self.assertIn(expected_msg, six.text_type(err))
 227
 228     def test_get_attribute_unknown_node_template_name(self):
 229         self.assertRaises(
 230             exception.ValidationError, self._load_template,
 231             'functions/test_get_attribute_unknown_node_template_name.yaml')
 232         exception.ExceptionCollector.assertExceptionMessage(
 233             KeyError,
 234             _('\'Node template "unknown_node_template" was not found.\''))
 235
 236     def test_get_attribute_unknown_attribute(self):
 237         self.assertRaises(
 238             exception.ValidationError, self._load_template,
 239             'functions/test_get_attribute_unknown_attribute_name.yaml')
 240         exception.ExceptionCollector.assertExceptionMessage(
 241             KeyError,
 242             _('\'Attribute "unknown_attribute" was not found in node template '
 243               '"server".\''))
 244
 245     def test_get_attribute_host_keyword(self):
 246         tpl = self._load_template(
 247             'functions/test_get_attribute_host_keyword.yaml')
 248
 249         def assert_get_attribute_host_functionality(node_template_name):
 250             node = [x for x in tpl.nodetemplates
 251                     if x.name == node_template_name][0]
 252             configure_op = [
 253                 x for x in node.interfaces if x.name == 'configure'][0]
 254             ip_addr_input = configure_op.inputs['ip_address']
 255             self.assertIsInstance(ip_addr_input, functions.GetAttribute)
 256             self.assertEqual('server',
 257                              ip_addr_input.get_referenced_node_template().name)
 258
 259         assert_get_attribute_host_functionality('dbms')
 260         assert_get_attribute_host_functionality('database')
 261
 262     def test_get_attribute_host_not_found(self):
 263         self.assertRaises(
 264             exception.ValidationError, self._load_template,
 265             'functions/test_get_attribute_host_not_found.yaml')
 266         exception.ExceptionCollector.assertExceptionMessage(
 267             ValueError,
 268             _('"get_attribute: [ HOST, ... ]" was used in node template '
 269               '"server" but "tosca.relationships.HostedOn" was not found in '
 270               'the relationship chain.'))
 271
 272     def test_get_attribute_illegal_host_in_outputs(self):
 273         self.assertRaises(
 274             exception.ValidationError, self._load_template,
 275             'functions/test_get_attribute_illegal_host_in_outputs.yaml')
 276         exception.ExceptionCollector.assertExceptionMessage(
 277             ValueError,
 278             _('"get_attribute: [ HOST, ... ]" is not allowed in "outputs" '
 279               'section of the TOSCA template.'))
 280
 281     def test_get_attribute_with_index(self):
 282         self._load_template(
 283             'functions/test_get_attribute_with_index.yaml')
 284
 285     def test_get_attribute_with_index_error(self):
 286         self.assertRaises(
 287             exception.ValidationError, self._load_template,
 288             'functions/test_get_attribute_with_index_error.yaml')
 289         exception.ExceptionCollector.assertExceptionMessage(
 290             ValueError,
 291             _('Illegal arguments for function "get_attribute". '
 292               'Unexpected attribute/index value "0"'))
 293
 294     def test_get_attribute_source_target_keywords(self):
 295         tosca_tpl = os.path.join(
 296             os.path.dirname(os.path.abspath(__file__)),
 297             "data/functions/test_get_attribute_source_target_keywords.yaml")
 298         tosca = ToscaTemplate(tosca_tpl,
 299                               parsed_params={'db_root_pwd': '12345678'})
 300
 301         for node in tosca.nodetemplates:
 302             for relationship, trgt in node.relationships.items():
 303                 rel_template = trgt.get_relationship_template()[0]
 304                 break
 305
 306         operation = self._get_operation(rel_template.interfaces,
 307                                         'pre_configure_source')
 308         target_test = operation.inputs['target_test']
 309         self.assertIsInstance(target_test, functions.GetAttribute)
 310         source_port = operation.inputs['source_port']
 311         self.assertIsInstance(source_port, functions.GetAttribute)
 312
 313     def test_get_attribute_with_nested_params(self):
 314         self._load_template(
 315             'functions/test_get_attribute_with_nested_params.yaml')
 316
 317     def test_implicit_attribute(self):
 318         self.assertIsNotNone(self._load_template(
 319             'functions/test_get_implicit_attribute.yaml'))
 320
 321
 322 class ConcatTest(TestCase):
 323
 324     def _load_template(self, filename):
 325         return ToscaTemplate(os.path.join(
 326             os.path.dirname(os.path.abspath(__file__)),
 327             filename))
 328
 329     def test_validate_concat(self):
 330         tosca = self._load_template("data/functions/test_concat.yaml")
 331         server_url_output = [
 332             output for output in tosca.outputs if output.name == 'url'][0]
 333         func = functions.get_function(self, tosca.outputs,
 334                                       server_url_output.value)
 335         self.assertIsInstance(func, functions.Concat)
 336
 337         self.assertRaises(exception.ValidationError, self._load_template,
 338                           'data/functions/test_concat_invalid.yaml')
 339         exception.ExceptionCollector.assertExceptionMessage(
 340             ValueError,
 341             _('Invalid arguments for function "concat". Expected at least '
 342               'one arguments.'))
 343
 344
 345 class TokenTest(TestCase):
 346
 347     def _load_template(self, filename):
 348         return ToscaTemplate(os.path.join(
 349             os.path.dirname(os.path.abspath(__file__)),
 350             filename))
 351
 352     def test_validate_token(self):
 353         tosca = self._load_template("data/functions/test_token.yaml")
 354         server_url_output = [
 355             output for output in tosca.outputs if output.name == 'url'][0]
 356         func = functions.get_function(self, tosca.outputs,
 357                                       server_url_output.value)
 358         self.assertIsInstance(func, functions.Token)
 359
 360         self.assertRaises(exception.ValidationError, self._load_template,
 361                           'data/functions/test_token_invalid.yaml')
 362         exception.ExceptionCollector.assertExceptionMessage(
 363             ValueError,
 364             _('Invalid arguments for function "token". Expected at least '
 365               'three arguments.'))
 366         exception.ExceptionCollector.assertExceptionMessage(
 367             ValueError,
 368             _('Invalid arguments for function "token". Expected '
 369               'integer value as third argument.'))
 370         exception.ExceptionCollector.assertExceptionMessage(
 371             ValueError,
 372             _('Invalid arguments for function "token". Expected '
 373               'single char value as second argument.'))