diff options
Diffstat (limited to 'tosca2heat/tosca-parser/toscaparser/tests/test_functions.py')
| -rw-r--r-- | tosca2heat/tosca-parser/toscaparser/tests/test_functions.py | 324 |
1 files changed, 324 insertions, 0 deletions
diff --git a/tosca2heat/tosca-parser/toscaparser/tests/test_functions.py b/tosca2heat/tosca-parser/toscaparser/tests/test_functions.py new file mode 100644 index 0000000..2a6225d --- /dev/null +++ b/tosca2heat/tosca-parser/toscaparser/tests/test_functions.py @@ -0,0 +1,324 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import os +import six +from toscaparser.common import exception +from toscaparser import functions +from toscaparser.tests.base import TestCase +from toscaparser.tosca_template import ToscaTemplate +from toscaparser.utils.gettextutils import _ + + +class IntrinsicFunctionsTest(TestCase): + + tosca_tpl = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/tosca_single_instance_wordpress.yaml") + params = {'db_name': 'my_wordpress', 'db_user': 'my_db_user'} + tosca = ToscaTemplate(tosca_tpl, parsed_params=params) + + def _get_node(self, node_name, tosca=None): + if tosca is None: + tosca = self.tosca + return [ + node for node in tosca.nodetemplates + if node.name == node_name][0] + + def _get_operation(self, interfaces, operation): + return [ + interface for interface in interfaces + if interface.name == operation][0] + + def _get_property(self, node_template, property_name): + return [prop.value for prop in node_template.get_properties_objects() + if prop.name == property_name][0] + + def _get_inputs_dict(self): + inputs = {} + for input in self.tosca.inputs: + inputs[input.name] = input.default + return inputs + + def _get_input(self, name): + self._get_inputs_dict()[name] + + def test_get_property(self): + wordpress = self._get_node('wordpress') + operation = self._get_operation(wordpress.interfaces, 'configure') + wp_db_password = operation.inputs['wp_db_password'] + self.assertTrue(isinstance(wp_db_password, functions.GetProperty)) + result = wp_db_password.result() + self.assertEqual('wp_pass', result) + + def test_get_property_with_input_param(self): + wordpress = self._get_node('wordpress') + operation = self._get_operation(wordpress.interfaces, 'configure') + wp_db_user = operation.inputs['wp_db_user'] + self.assertTrue(isinstance(wp_db_user, functions.GetProperty)) + result = wp_db_user.result() + self.assertEqual('my_db_user', result) + + def test_unknown_capability_property(self): + self.assertRaises(exception.ValidationError, self._load_template, + 'functions/test_unknown_capability_property.yaml') + exception.ExceptionCollector.assertExceptionMessage( + KeyError, + _('\'Property "unknown" was not found in capability ' + '"database_endpoint" of node template "database" referenced ' + 'from node template "database".\'')) + + def test_get_input_in_properties(self): + mysql_dbms = self._get_node('mysql_dbms') + expected_inputs = ['db_root_pwd', 'db_port'] + props = mysql_dbms.get_properties() + for key in props.keys(): + prop = props[key] + self.assertTrue(isinstance(prop.value, functions.GetInput)) + expected_inputs.remove(prop.value.input_name) + self.assertListEqual(expected_inputs, []) + + def test_get_input_validation(self): + self.assertRaises( + exception.ValidationError, self._load_template, + 'functions/test_unknown_input_in_property.yaml') + exception.ExceptionCollector.assertExceptionMessage( + exception.UnknownInputError, + _('Unknown input "objectstore_name".')) + + self.assertRaises( + exception.ValidationError, self._load_template, + 'functions/test_unknown_input_in_interface.yaml') + exception.ExceptionCollector.assertExceptionMessage( + exception.UnknownInputError, + _('Unknown input "image_id".')) + + self.assertRaises( + exception.ValidationError, self._load_template, + 'functions/test_invalid_function_signature.yaml') + exception.ExceptionCollector.assertExceptionMessage( + ValueError, + _('Expected one argument for function "get_input" but received ' + '"[\'cpus\', \'cpus\']".')) + + def test_get_input_default_value_result(self): + mysql_dbms = self._get_node('mysql_dbms') + dbms_port = self._get_property(mysql_dbms, 'port') + self.assertEqual(3306, dbms_port.result()) + dbms_root_password = self._get_property(mysql_dbms, + 'root_password') + self.assertIsNone(dbms_root_password.result()) + + def test_get_property_with_host(self): + tosca_tpl = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/functions/test_get_property_with_host.yaml") + mysql_database = self._get_node('mysql_database', + ToscaTemplate(tosca_tpl)) + operation = self._get_operation(mysql_database.interfaces, 'configure') + db_port = operation.inputs['db_port'] + self.assertTrue(isinstance(db_port, functions.GetProperty)) + result = db_port.result() + self.assertEqual(3306, result) + test = operation.inputs['test'] + self.assertTrue(isinstance(test, functions.GetProperty)) + result = test.result() + self.assertEqual(1, result) + + def test_get_property_with_nested_params(self): + tosca_tpl = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/functions/tosca_nested_property_names_indexes.yaml") + webserver = self._get_node('wordpress', ToscaTemplate(tosca_tpl)) + operation = self._get_operation(webserver.interfaces, 'configure') + wp_endpoint_prot = operation.inputs['wp_endpoint_protocol'] + self.assertTrue(isinstance(wp_endpoint_prot, functions.GetProperty)) + self.assertEqual('tcp', wp_endpoint_prot.result()) + wp_list_prop = operation.inputs['wp_list_prop'] + self.assertTrue(isinstance(wp_list_prop, functions.GetProperty)) + self.assertEqual(3, wp_list_prop.result()) + + def test_get_property_with_capabilties_inheritance(self): + tosca_tpl = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/functions/test_capabilties_inheritance.yaml") + some_node = self._get_node('some_node', ToscaTemplate(tosca_tpl)) + operation = self._get_operation(some_node.interfaces, 'configure') + some_input = operation.inputs['some_input'] + self.assertTrue(isinstance(some_input, functions.GetProperty)) + self.assertEqual('someval', some_input.result()) + + def test_get_property_source_target_keywords(self): + tosca_tpl = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/functions/test_get_property_source_target_keywords.yaml") + tosca = ToscaTemplate(tosca_tpl) + + for node in tosca.nodetemplates: + for relationship, trgt in node.relationships.items(): + rel_template = trgt.get_relationship_template()[0] + break + + operation = self._get_operation(rel_template.interfaces, + 'pre_configure_source') + target_test = operation.inputs['target_test'] + self.assertTrue(isinstance(target_test, functions.GetProperty)) + self.assertEqual(1, target_test.result()) + source_port = operation.inputs['source_port'] + self.assertTrue(isinstance(source_port, functions.GetProperty)) + self.assertEqual(3306, source_port.result()) + + +class GetAttributeTest(TestCase): + + def _load_template(self, filename): + return ToscaTemplate(os.path.join( + os.path.dirname(os.path.abspath(__file__)), + 'data', + filename)) + + def _get_operation(self, interfaces, operation): + return [ + interface for interface in interfaces + if interface.name == operation][0] + + def test_get_attribute_in_outputs(self): + tpl = self._load_template('tosca_single_instance_wordpress.yaml') + website_url_output = [ + x for x in tpl.outputs if x.name == 'website_url'][0] + self.assertIsInstance(website_url_output.value, functions.GetAttribute) + self.assertEqual('server', website_url_output.value.node_template_name) + self.assertEqual('private_address', + website_url_output.value.attribute_name) + + def test_get_attribute_invalid_args(self): + expected_msg = _('Expected arguments: "node-template-name", ' + '"attribute-name"') + err = self.assertRaises(ValueError, + functions.get_function, None, None, + {'get_attribute': []}) + self.assertIn(expected_msg, six.text_type(err)) + err = self.assertRaises(ValueError, + functions.get_function, None, None, + {'get_attribute': ['x']}) + self.assertIn(expected_msg, six.text_type(err)) + err = self.assertRaises(ValueError, + functions.get_function, None, None, + {'get_attribute': ['x', 'y', 'z', 'k']}) + self.assertIn(expected_msg, six.text_type(err)) + + def test_get_attribute_unknown_node_template_name(self): + self.assertRaises( + exception.ValidationError, self._load_template, + 'functions/test_get_attribute_unknown_node_template_name.yaml') + exception.ExceptionCollector.assertExceptionMessage( + KeyError, + _('\'Node template "unknown_node_template" was not found.\'')) + + def test_get_attribute_unknown_attribute(self): + self.assertRaises( + exception.ValidationError, self._load_template, + 'functions/test_get_attribute_unknown_attribute_name.yaml') + exception.ExceptionCollector.assertExceptionMessage( + KeyError, + _('\'Attribute "unknown_attribute" was not found in node template ' + '"server".\'')) + + def test_get_attribute_host_keyword(self): + tpl = self._load_template( + 'functions/test_get_attribute_host_keyword.yaml') + + def assert_get_attribute_host_functionality(node_template_name): + node = [x for x in tpl.nodetemplates + if x.name == node_template_name][0] + configure_op = [ + x for x in node.interfaces if x.name == 'configure'][0] + ip_addr_input = configure_op.inputs['ip_address'] + self.assertIsInstance(ip_addr_input, functions.GetAttribute) + self.assertEqual('server', + ip_addr_input.get_referenced_node_template().name) + + assert_get_attribute_host_functionality('dbms') + assert_get_attribute_host_functionality('database') + + def test_get_attribute_host_not_found(self): + self.assertRaises( + exception.ValidationError, self._load_template, + 'functions/test_get_attribute_host_not_found.yaml') + exception.ExceptionCollector.assertExceptionMessage( + ValueError, + _('"get_attribute: [ HOST, ... ]" was used in node template ' + '"server" but "tosca.relationships.HostedOn" was not found in ' + 'the relationship chain.')) + + def test_get_attribute_illegal_host_in_outputs(self): + self.assertRaises( + exception.ValidationError, self._load_template, + 'functions/test_get_attribute_illegal_host_in_outputs.yaml') + exception.ExceptionCollector.assertExceptionMessage( + ValueError, + _('"get_attribute: [ HOST, ... ]" is not allowed in "outputs" ' + 'section of the TOSCA template.')) + + def test_get_attribute_with_index(self): + self._load_template( + 'functions/test_get_attribute_with_index.yaml') + + def test_get_attribute_with_index_error(self): + self.assertRaises( + exception.ValidationError, self._load_template, + 'functions/test_get_attribute_with_index_error.yaml') + exception.ExceptionCollector.assertExceptionMessage( + ValueError, + _('Illegal arguments for function "get_attribute". ' + 'Expected arguments: "node-template-name", "attribute-name"')) + + def test_get_attribute_source_target_keywords(self): + tosca_tpl = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "data/functions/test_get_attribute_source_target_keywords.yaml") + tosca = ToscaTemplate(tosca_tpl) + + for node in tosca.nodetemplates: + for relationship, trgt in node.relationships.items(): + rel_template = trgt.get_relationship_template()[0] + break + + operation = self._get_operation(rel_template.interfaces, + 'pre_configure_source') + target_test = operation.inputs['target_test'] + self.assertTrue(isinstance(target_test, functions.GetAttribute)) + source_port = operation.inputs['source_port'] + self.assertTrue(isinstance(source_port, functions.GetAttribute)) + + +class ConcatTest(TestCase): + + def _load_template(self, filename): + return ToscaTemplate(os.path.join( + os.path.dirname(os.path.abspath(__file__)), + filename)) + + def test_validate_concat(self): + tosca = self._load_template("data/functions/test_concat.yaml") + server_url_output = [ + output for output in tosca.outputs if output.name == 'url'][0] + func = functions.get_function(self, tosca.outputs, + server_url_output.value) + self.assertIsInstance(func, functions.Concat) + + self.assertRaises(exception.ValidationError, self._load_template, + 'data/functions/test_concat_invalid.yaml') + exception.ExceptionCollector.assertExceptionMessage( + ValueError, + _('Invalid arguments for function "concat". Expected at least ' + 'one arguments.')) |