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 'db_root_pwd': '12345678'}
29 tosca = ToscaTemplate(tosca_tpl, parsed_params=params)
31 def _get_node(self, node_name, tosca=None):
35 node for node in tosca.nodetemplates
36 if node.name == node_name][0]
38 def _get_operation(self, interfaces, operation):
40 interface for interface in interfaces
41 if interface.name == operation][0]
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]
47 def _get_inputs_dict(self):
49 for input in self.tosca.inputs:
50 inputs[input.name] = input.default
53 def _get_input(self, name):
54 self._get_inputs_dict()[name]
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)
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)
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(
77 _('\'Property "unknown" was not found in capability '
78 '"database_endpoint" of node template "database" referenced '
79 'from node template "database".\''))
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():
87 self.assertIsInstance(prop.value, functions.GetInput)
88 expected_inputs.remove(prop.value.input_name)
89 self.assertListEqual(expected_inputs, [])
91 def test_get_input_validation(self):
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".'))
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".'))
107 exception.ValidationError, self._load_template,
108 'functions/test_invalid_function_signature.yaml')
109 exception.ExceptionCollector.assertExceptionMessage(
111 _('Expected one argument for function "get_input" but received '
112 '"[\'cpus\', \'cpus\']".'))
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,
120 self.assertEqual(dbms_root_password.result(), '12345678')
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,
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)
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,
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())
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,
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())
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'})
177 for node in tosca.nodetemplates:
178 for relationship, trgt in node.relationships.items():
179 rel_template = trgt.get_relationship_template()[0]
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())
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())
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())
212 class GetAttributeTest(TestCase):
214 def _load_template(self, filename):
215 return ToscaTemplate(os.path.join(
216 os.path.dirname(os.path.abspath(__file__)),
219 parsed_params={'db_root_pwd': '1234'})
221 def _get_operation(self, interfaces, operation):
223 interface for interface in interfaces
224 if interface.name == operation][0]
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)
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))
248 def test_get_attribute_unknown_node_template_name(self):
250 exception.ValidationError, self._load_template,
251 'functions/test_get_attribute_unknown_node_template_name.yaml')
252 exception.ExceptionCollector.assertExceptionMessage(
254 _('\'Node template "unknown_node_template" was not found.\''))
256 def test_get_attribute_unknown_attribute(self):
258 exception.ValidationError, self._load_template,
259 'functions/test_get_attribute_unknown_attribute_name.yaml')
260 exception.ExceptionCollector.assertExceptionMessage(
262 _('\'Attribute "unknown_attribute" was not found in node template '
265 def test_get_attribute_host_keyword(self):
266 tpl = self._load_template(
267 'functions/test_get_attribute_host_keyword.yaml')
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]
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)
279 assert_get_attribute_host_functionality('dbms')
280 assert_get_attribute_host_functionality('database')
282 def test_get_attribute_host_not_found(self):
284 exception.ValidationError, self._load_template,
285 'functions/test_get_attribute_host_not_found.yaml')
286 exception.ExceptionCollector.assertExceptionMessage(
288 _('"get_attribute: [ HOST, ... ]" was used in node template '
289 '"server" but "tosca.relationships.HostedOn" was not found in '
290 'the relationship chain.'))
292 def test_get_attribute_illegal_host_in_outputs(self):
294 exception.ValidationError, self._load_template,
295 'functions/test_get_attribute_illegal_host_in_outputs.yaml')
296 exception.ExceptionCollector.assertExceptionMessage(
298 _('"get_attribute: [ HOST, ... ]" is not allowed in "outputs" '
299 'section of the TOSCA template.'))
301 def test_get_attribute_with_index(self):
303 'functions/test_get_attribute_with_index.yaml')
305 def test_get_attribute_with_index_error(self):
307 exception.ValidationError, self._load_template,
308 'functions/test_get_attribute_with_index_error.yaml')
309 exception.ExceptionCollector.assertExceptionMessage(
311 _('Illegal arguments for function "get_attribute". '
312 'Unexpected attribute/index value "0"'))
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'})
321 for node in tosca.nodetemplates:
322 for relationship, trgt in node.relationships.items():
323 rel_template = trgt.get_relationship_template()[0]
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)
333 def test_get_attribute_with_nested_params(self):
335 'functions/test_get_attribute_with_nested_params.yaml')
337 def test_implicit_attribute(self):
338 self.assertIsNotNone(self._load_template(
339 'functions/test_get_implicit_attribute.yaml'))
341 def test_get_attribute_capability_inheritance(self):
342 self.assertIsNotNone(self._load_template(
343 'functions/test_container_cap_child.yaml'))
346 class ConcatTest(TestCase):
348 def _load_template(self, filename):
349 return ToscaTemplate(os.path.join(
350 os.path.dirname(os.path.abspath(__file__)),
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)
361 self.assertRaises(exception.ValidationError, self._load_template,
362 'data/functions/test_concat_invalid.yaml')
363 exception.ExceptionCollector.assertExceptionMessage(
365 _('Invalid arguments for function "concat". Expected at least '
369 class TokenTest(TestCase):
371 def _load_template(self, filename):
372 return ToscaTemplate(os.path.join(
373 os.path.dirname(os.path.abspath(__file__)),
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)
384 self.assertRaises(exception.ValidationError, self._load_template,
385 'data/functions/test_token_invalid.yaml')
386 exception.ExceptionCollector.assertExceptionMessage(
388 _('Invalid arguments for function "token". Expected at least '
390 exception.ExceptionCollector.assertExceptionMessage(
392 _('Invalid arguments for function "token". Expected '
393 'integer value as third argument.'))
394 exception.ExceptionCollector.assertExceptionMessage(
396 _('Invalid arguments for function "token". Expected '
397 'single char value as second argument.'))