b669c8184d
The TOSCA interfaces operation keyword 'input' is changed to 'inputs' in the TOSCA spec as it was a bug in the spec. Change-Id: Id7d60c7d2c23258938388d2fccd1569f64050f3c Closes-Bug: #1376333
183 lines
7.8 KiB
Python
183 lines
7.8 KiB
Python
# 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 translator.toscalib.common import exception
|
|
from translator.toscalib import functions
|
|
from translator.toscalib.tests.base import TestCase
|
|
from translator.toscalib.tosca_template import ToscaTemplate
|
|
|
|
|
|
class IntrinsicFunctionsTest(TestCase):
|
|
|
|
tosca_tpl = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)),
|
|
"data/tosca_single_instance_wordpress.yaml")
|
|
tosca = ToscaTemplate(tosca_tpl)
|
|
|
|
def _get_node(self, node_name):
|
|
return [
|
|
node for node in self.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 test_get_property(self):
|
|
mysql_dbms = self._get_node('mysql_dbms')
|
|
operation = self._get_operation(mysql_dbms.interfaces, 'configure')
|
|
db_root_password = operation.inputs['db_root_password']
|
|
self.assertTrue(isinstance(db_root_password, functions.GetProperty))
|
|
result = db_root_password.result()
|
|
self.assertTrue(isinstance(result, functions.GetInput))
|
|
|
|
def test_get_requirement_property(self):
|
|
wordpress = self._get_node('wordpress')
|
|
operation = self._get_operation(wordpress.interfaces, 'configure')
|
|
wp_db_port = operation.inputs['wp_db_port']
|
|
self.assertTrue(isinstance(wp_db_port, functions.GetProperty))
|
|
result = wp_db_port.result()
|
|
self.assertTrue(isinstance(result, functions.GetInput))
|
|
self.assertEqual('db_port', result.input_name)
|
|
|
|
def test_get_capability_property(self):
|
|
mysql_database = self._get_node('mysql_database')
|
|
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.assertTrue(isinstance(result, functions.GetInput))
|
|
self.assertEqual('db_port', result.input_name)
|
|
|
|
def test_unknown_capability_property(self):
|
|
err = self.assertRaises(
|
|
KeyError,
|
|
self._load_template,
|
|
'functions/test_unknown_capability_property.yaml')
|
|
self.assertIn("'unknown'", six.text_type(err))
|
|
self.assertIn("'database_endpoint'", six.text_type(err))
|
|
self.assertIn("'database'", six.text_type(err))
|
|
|
|
def test_get_input_in_properties(self):
|
|
mysql_dbms = self._get_node('mysql_dbms')
|
|
self.assertTrue(isinstance(mysql_dbms.properties[0].value,
|
|
functions.GetInput))
|
|
self.assertTrue(mysql_dbms.properties[0].value.input_name,
|
|
'db_root_pwd')
|
|
self.assertTrue(isinstance(mysql_dbms.properties[1].value,
|
|
functions.GetInput))
|
|
self.assertTrue(mysql_dbms.properties[1].value.input_name,
|
|
'db_port')
|
|
|
|
def test_get_input_in_interface(self):
|
|
mysql_dbms = self._get_node('mysql_dbms')
|
|
operation = self._get_operation(mysql_dbms.interfaces, 'configure')
|
|
db_user = operation.inputs['db_user']
|
|
self.assertTrue(isinstance(db_user, functions.GetInput))
|
|
|
|
def test_get_input_validation(self):
|
|
self.assertRaises(exception.UnknownInputError,
|
|
self._load_template,
|
|
'functions/test_unknown_input_in_property.yaml')
|
|
self.assertRaises(exception.UnknownInputError,
|
|
self._load_template,
|
|
'functions/test_unknown_input_in_interface.yaml')
|
|
|
|
|
|
class GetAttributeTest(TestCase):
|
|
|
|
def _load_template(self, filename):
|
|
return ToscaTemplate(os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)),
|
|
'data',
|
|
filename))
|
|
|
|
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('ip_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']})
|
|
self.assertIn(expected_msg, six.text_type(err))
|
|
|
|
def test_get_attribute_unknown_node_template_name(self):
|
|
err = self.assertRaises(
|
|
KeyError,
|
|
self._load_template,
|
|
'functions/test_get_attribute_unknown_node_template_name.yaml')
|
|
self.assertIn('unknown_node_template', six.text_type(err))
|
|
|
|
def test_get_attribute_unknown_attribute(self):
|
|
err = self.assertRaises(
|
|
KeyError,
|
|
self._load_template,
|
|
'functions/test_get_attribute_unknown_attribute_name.yaml')
|
|
self.assertIn('unknown_attribute', six.text_type(err))
|
|
self.assertIn('server', six.text_type(err))
|
|
|
|
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):
|
|
err = self.assertRaises(
|
|
ValueError,
|
|
self._load_template,
|
|
'functions/test_get_attribute_host_not_found.yaml')
|
|
self.assertIn(
|
|
"get_attribute HOST keyword is used in 'server' node template but "
|
|
"tosca.relationships.HostedOn was not found in relationship chain",
|
|
six.text_type(err))
|
|
|
|
def test_get_attribute_illegal_host_in_outputs(self):
|
|
err = self.assertRaises(
|
|
ValueError,
|
|
self._load_template,
|
|
'functions/test_get_attribute_illegal_host_in_outputs.yaml')
|
|
self.assertIn(
|
|
"get_attribute HOST keyword is not allowed within the outputs "
|
|
"section of the TOSCA template",
|
|
six.text_type(err))
|