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