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
5 # http://www.apache.org/licenses/LICENSE-2.0
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
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 _
22 class IntrinsicFunctionsTest(TestCase):
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)
30 def _get_node(self, node_name, tosca=None):
34 node for node in tosca.nodetemplates
35 if node.name == node_name][0]
37 def _get_operation(self, interfaces, operation):
39 interface for interface in interfaces
40 if interface.name == operation][0]
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]
46 def _get_inputs_dict(self):
48 for input in self.tosca.inputs:
49 inputs[input.name] = input.default
52 def _get_input(self, name):
53 self._get_inputs_dict()[name]
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)
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)
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(
76 _('\'Property "unknown" was not found in capability '
77 '"database_endpoint" of node template "database" referenced '
78 'from node template "database".\''))
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():
86 self.assertTrue(isinstance(prop.value, functions.GetInput))
87 expected_inputs.remove(prop.value.input_name)
88 self.assertListEqual(expected_inputs, [])
90 def test_get_input_validation(self):
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".'))
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".'))
106 exception.ValidationError, self._load_template,
107 'functions/test_invalid_function_signature.yaml')
108 exception.ExceptionCollector.assertExceptionMessage(
110 _('Expected one argument for function "get_input" but received '
111 '"[\'cpus\', \'cpus\']".'))
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,
119 self.assertIsNone(dbms_root_password.result())
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)
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())
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())
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)
166 for node in tosca.nodetemplates:
167 for relationship, trgt in node.relationships.items():
168 rel_template = trgt.get_relationship_template()[0]
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())
181 class GetAttributeTest(TestCase):
183 def _load_template(self, filename):
184 return ToscaTemplate(os.path.join(
185 os.path.dirname(os.path.abspath(__file__)),
189 def _get_operation(self, interfaces, operation):
191 interface for interface in interfaces
192 if interface.name == operation][0]
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)
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))
216 def test_get_attribute_unknown_node_template_name(self):
218 exception.ValidationError, self._load_template,
219 'functions/test_get_attribute_unknown_node_template_name.yaml')
220 exception.ExceptionCollector.assertExceptionMessage(
222 _('\'Node template "unknown_node_template" was not found.\''))
224 def test_get_attribute_unknown_attribute(self):
226 exception.ValidationError, self._load_template,
227 'functions/test_get_attribute_unknown_attribute_name.yaml')
228 exception.ExceptionCollector.assertExceptionMessage(
230 _('\'Attribute "unknown_attribute" was not found in node template '
233 def test_get_attribute_host_keyword(self):
234 tpl = self._load_template(
235 'functions/test_get_attribute_host_keyword.yaml')
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]
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)
247 assert_get_attribute_host_functionality('dbms')
248 assert_get_attribute_host_functionality('database')
250 def test_get_attribute_host_not_found(self):
252 exception.ValidationError, self._load_template,
253 'functions/test_get_attribute_host_not_found.yaml')
254 exception.ExceptionCollector.assertExceptionMessage(
256 _('"get_attribute: [ HOST, ... ]" was used in node template '
257 '"server" but "tosca.relationships.HostedOn" was not found in '
258 'the relationship chain.'))
260 def test_get_attribute_illegal_host_in_outputs(self):
262 exception.ValidationError, self._load_template,
263 'functions/test_get_attribute_illegal_host_in_outputs.yaml')
264 exception.ExceptionCollector.assertExceptionMessage(
266 _('"get_attribute: [ HOST, ... ]" is not allowed in "outputs" '
267 'section of the TOSCA template.'))
269 def test_get_attribute_with_index(self):
271 'functions/test_get_attribute_with_index.yaml')
273 def test_get_attribute_with_index_error(self):
275 exception.ValidationError, self._load_template,
276 'functions/test_get_attribute_with_index_error.yaml')
277 exception.ExceptionCollector.assertExceptionMessage(
279 _('Illegal arguments for function "get_attribute". '
280 'Unexpected attribute/index value "0"'))
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)
288 for node in tosca.nodetemplates:
289 for relationship, trgt in node.relationships.items():
290 rel_template = trgt.get_relationship_template()[0]
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))
300 def test_get_attribute_with_nested_params(self):
302 'functions/test_get_attribute_with_nested_params.yaml')
305 class ConcatTest(TestCase):
307 def _load_template(self, filename):
308 return ToscaTemplate(os.path.join(
309 os.path.dirname(os.path.abspath(__file__)),
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)
320 self.assertRaises(exception.ValidationError, self._load_template,
321 'data/functions/test_concat_invalid.yaml')
322 exception.ExceptionCollector.assertExceptionMessage(
324 _('Invalid arguments for function "concat". Expected at least '
328 class TokenTest(TestCase):
330 def _load_template(self, filename):
331 return ToscaTemplate(os.path.join(
332 os.path.dirname(os.path.abspath(__file__)),
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)
343 self.assertRaises(exception.ValidationError, self._load_template,
344 'data/functions/test_token_invalid.yaml')
345 exception.ExceptionCollector.assertExceptionMessage(
347 _('Invalid arguments for function "token". Expected at least '
349 exception.ExceptionCollector.assertExceptionMessage(
351 _('Invalid arguments for function "token". Expected '
352 'integer value as third argument.'))
353 exception.ExceptionCollector.assertExceptionMessage(
355 _('Invalid arguments for function "token". Expected '
356 'single char value as second argument.'))