from toscaparser.parameters import Output
from toscaparser.policy import Policy
from toscaparser.relationship_template import RelationshipTemplate
+from toscaparser.repositories import Repository
from toscaparser.tests.base import TestCase
from toscaparser.topology_template import TopologyTemplate
from toscaparser.tosca_template import ToscaTemplate
from toscaparser.triggers import Triggers
from toscaparser.utils.gettextutils import _
-
import toscaparser.utils.yamlparser
tpl_path = 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',
+ 'db_root_pwd': '12345678'}
+ self.assertIsNotNone(ToscaTemplate(tpl_path, params))
+
+ def test_custom_interface_allowed(self):
+ tpl_path = os.path.join(
+ os.path.dirname(os.path.abspath(__file__)),
+ "data/interfaces/test_custom_interface_in_template.yaml")
self.assertIsNotNone(ToscaTemplate(tpl_path))
+ def test_custom_interface_invalid_operation(self):
+ tpl_path = os.path.join(
+ os.path.dirname(os.path.abspath(__file__)),
+ "data/interfaces/test_custom_interface_invalid_operation.yaml")
+ self.assertRaises(exception.ValidationError,
+ ToscaTemplate, tpl_path)
+ exception.ExceptionCollector.assertExceptionMessage(
+ exception.UnknownFieldError,
+ _('"interfaces" of template "customInterfaceTest" '
+ 'contains unknown field "CustomOp4". '
+ 'Refer to the definition to verify valid values.'))
+
def test_first_level_sections(self):
tpl_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
_('The template version "tosca_simple_yaml_1_10 in '
'custom_types/imported_sample.yaml" is invalid. '
'Valid versions are "tosca_simple_yaml_1_0, '
+ 'tosca_simple_yaml_1_1, '
'tosca_simple_profile_for_nfv_1_0_0".'))
exception.ExceptionCollector.assertExceptionMessage(
exception.UnknownFieldError,
_('Policy "mycompany.mytypes.myScalingPolicy" contains unknown '
'field "derived1_from". Refer to the definition to '
'verify valid values.'))
+ exception.ExceptionCollector.assertExceptionMessage(
+ exception.UnknownFieldError,
+ _('Relationshiptype "test.relation.connects" contains unknown '
+ 'field "derived_from4". Refer to the definition to '
+ 'verify valid values.'))
- def test_inputs(self):
+ def test_getoperation_IncorrectValue(self):
+ # test case 1
+ tpl_snippet = '''
+ node_templates:
+ front_end:
+ type: tosca.nodes.Compute
+ interfaces:
+ Standard:
+ create:
+ implementation: scripts/frontend/create.sh
+ configure:
+ implementation: scripts/frontend/configure.sh
+ inputs:
+ data_dir: {get_operation_output: [front_end,Standard1,
+ create,data_dir]}
+ '''
+ tpl = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet))
+ err = self.assertRaises(ValueError,
+ TopologyTemplate, tpl, None)
+ expectedmessage = _('Enter a valid interface name')
+ self.assertEqual(expectedmessage, err.__str__())
+ # test case 2
+ tpl_snippet2 = '''
+ node_templates:
+ front_end:
+ type: tosca.nodes.Compute
+ interfaces:
+ Standard:
+ create:
+ implementation: scripts/frontend/create.sh
+ configure:
+ implementation: scripts/frontend/configure.sh
+ inputs:
+ data_dir: {get_operation_output: [front_end1,Standard,
+ create,data_dir]}
+ '''
+ tpl2 = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet2))
+ err2 = self.assertRaises(KeyError,
+ TopologyTemplate, tpl2, None)
+ expectedmessage2 = _('\'Node template "front_end1" was not found.\'')
+ self.assertEqual(expectedmessage2, err2.__str__())
+ # test case 3
+ tpl_snippet3 = '''
+ node_templates:
+ front_end:
+ type: tosca.nodes.Compute
+ interfaces:
+ Standard:
+ create:
+ implementation: scripts/frontend/create.sh
+ configure:
+ implementation: scripts/frontend/configure.sh
+ inputs:
+ data_dir: {get_operation_output: [front_end,Standard,
+ get_target,data_dir]}
+ '''
+ tpl3 = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet3))
+ err3 = self.assertRaises(ValueError,
+ TopologyTemplate, tpl3, None)
+ expectedmessage3 = _('Enter an operation of Standard interface')
+ self.assertEqual(expectedmessage3, err3.__str__())
+ # test case 4
+ tpl_snippet4 = '''
+ node_templates:
+ front_end:
+ type: tosca.nodes.Compute
+ interfaces:
+ Standard:
+ create:
+ implementation: scripts/frontend/create.sh
+ configure:
+ implementation: scripts/frontend/configure.sh
+ inputs:
+ data_dir: {get_operation_output: [front_end,Configure,
+ create,data_dir]}
+ '''
+ tpl4 = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet4))
+ err4 = self.assertRaises(ValueError,
+ TopologyTemplate, tpl4, None)
+ expectedmessage4 = _('Enter an operation of Configure interface')
+ self.assertEqual(expectedmessage4, err4.__str__())
+ # test case 5
+ tpl_snippet5 = '''
+ node_templates:
+ front_end:
+ type: tosca.nodes.Compute
+ interfaces:
+ Standard:
+ create:
+ implementation: scripts/frontend/create.sh
+ configure:
+ implementation: scripts/frontend/configure.sh
+ inputs:
+ data_dir: {get_operation_output: [front_end,Standard,
+ create]}
+ '''
+ tpl5 = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet5))
+ err5 = self.assertRaises(ValueError,
+ TopologyTemplate, tpl5, None)
+ expectedmessage5 = _('Illegal arguments for function'
+ ' "get_operation_output".'
+ ' Expected arguments: "template_name",'
+ '"interface_name",'
+ '"operation_name","output_variable_name"')
+ self.assertEqual(expectedmessage5, err5.__str__())
+
+ def test_unsupported_type(self):
tpl_snippet = '''
+ node_templates:
+ invalid_type:
+ type: tosca.test.invalidtype
+ properties:
+ size: { get_input: storage_size }
+ snapshot_id: { get_input: storage_snapshot_id }
+ '''
+ tpl = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet))
+ err = self.assertRaises(exception.UnsupportedTypeError,
+ TopologyTemplate, tpl, None)
+ expectedmessage = _('Type "tosca.test.invalidtype" is valid'
+ ' TOSCA type but not supported at this time.')
+ self.assertEqual(expectedmessage, err.__str__())
+
+ def test_inputs(self):
+ tpl_snippet1 = '''
inputs:
cpus:
type: integer
description: Number of CPUs for the server.
constraint:
- - valid_values: [ 1, 2, 4, 8 ]
+ - valid_values: [ 1, 2, 4 ]
+ required: yes
+ status: supported
+ '''
+ tpl_snippet2 = '''
+ inputs:
+ cpus:
+ type: integer
+ description: Number of CPUs for the server.
+ constraints:
+ - valid_values: [ 1, 2, 4 ]
+ required: yes
+ status: supported
'''
- inputs = (toscaparser.utils.yamlparser.
- simple_parse(tpl_snippet)['inputs'])
- name, attrs = list(inputs.items())[0]
- input = Input(name, attrs)
- err = self.assertRaises(exception.UnknownFieldError, input.validate)
- self.assertEqual(_('Input "cpus" contains unknown field "constraint". '
- 'Refer to the definition to verify valid values.'),
- err.__str__())
+ tpl_snippet3 = '''
+ inputs:
+ some_list:
+ type: list
+ description: List of items
+ entry_schema:
+ type: string
+ default: []
+ '''
+ inputs1 = (toscaparser.utils.yamlparser.
+ simple_parse(tpl_snippet1)['inputs'])
+ name1, attrs1 = list(inputs1.items())[0]
+ inputs2 = (toscaparser.utils.yamlparser.
+ simple_parse(tpl_snippet2)['inputs'])
+ name2, attrs2 = list(inputs2.items())[0]
+ try:
+ Input(name1, attrs1)
+ except Exception as err:
+ self.assertEqual(_('Input "cpus" contains unknown field '
+ '"constraint". Refer to the definition to '
+ 'verify valid values.'),
+ err.__str__())
+ input2 = Input(name2, attrs2)
+ self.assertTrue(input2.required)
+ toscaparser.utils.yamlparser.simple_parse(tpl_snippet3)['inputs']
def _imports_content_test(self, tpl_snippet, path, custom_type_def):
imports = (toscaparser.utils.yamlparser.
self.assertTrue(custom_defs.get("mycompany.tosca.nodes."
"WebApplication.WordPress"))
+ def test_imports_file_with_suffix_yml(self):
+ tpl_snippet = '''
+ imports:
+ - custom_types/wordpress.yml
+ '''
+ path = 'toscaparser/tests/data/tosca_elk.yaml'
+ custom_defs = self._imports_content_test(tpl_snippet,
+ path,
+ "node_types")
+ self.assertTrue(custom_defs.get("tosca.nodes."
+ "WebApplication.WordPress"))
+
def test_import_error_file_uri(self):
tpl_snippet = '''
imports:
try:
output.validate()
except Exception as err:
- self.assertTrue(isinstance(err, exception.UnknownFieldError))
+ self.assertIsInstance(err, exception.UnknownFieldError)
self.assertEqual(_('Output "server_address" contains unknown '
'field "descriptions". Refer to the definition '
'to verify valid values.'),
err.__str__())
+ def _repo_content(self, path):
+ repositories = path['repositories']
+ reposit = []
+ for name, val in repositories.items():
+ reposits = Repository(name, val)
+ reposit.append(reposits)
+ return reposit
+
+ def test_repositories(self):
+ tpl_snippet = '''
+ repositories:
+ repo_code0: https://raw.githubusercontent.com/nandinivemula/intern
+ repo_code1:
+ description: My project's code Repository in github usercontent.
+ url: https://github.com/nandinivemula/intern
+ credential:
+ user: nandini
+ password: tcs@12345
+ repo_code2:
+ description: My Project's code Repository in github.
+ url: https://github.com/nandinivemula/intern
+ credential:
+ user: xyzw
+ password: xyz@123
+ '''
+ tpl = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet))
+ repoobject = self._repo_content(tpl)
+ actualrepo_names = []
+ for repo in repoobject:
+ repos = repo.name
+ actualrepo_names.append(repos)
+ reposname = list(tpl.values())
+ reposnames = reposname[0]
+ expected_reponames = list(reposnames.keys())
+ self.assertEqual(expected_reponames, actualrepo_names)
+
+ def test_repositories_with_missing_required_field(self):
+ tpl_snippet = '''
+ repositories:
+ repo_code0: https://raw.githubusercontent.com/nandinivemula/intern
+ repo_code1:
+ description: My project's code Repository in github usercontent.
+ credential:
+ user: nandini
+ password: tcs@12345
+ repo_code2:
+ description: My Project's code Repository in github.
+ url: https://github.com/nandinivemula/intern
+ credential:
+ user: xyzw
+ password: xyz@123
+ '''
+ tpl = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet))
+ err = self.assertRaises(exception.MissingRequiredFieldError,
+ self._repo_content, tpl)
+ expectedmessage = _('Repository "repo_code1" is missing '
+ 'required field "url".')
+ self.assertEqual(expectedmessage, err.__str__())
+
+ def test_repositories_with_unknown_field(self):
+ tpl_snippet = '''
+ repositories:
+ repo_code0: https://raw.githubusercontent.com/nandinivemula/intern
+ repo_code1:
+ description: My project's code Repository in github usercontent.
+ url: https://github.com/nandinivemula/intern
+ credential:
+ user: nandini
+ password: tcs@12345
+ repo_code2:
+ descripton: My Project's code Repository in github.
+ url: https://github.com/nandinivemula/intern
+ credential:
+ user: xyzw
+ password: xyz@123
+ '''
+ tpl = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet))
+ err = self.assertRaises(exception.UnknownFieldError,
+ self._repo_content, tpl)
+ expectedmessage = _('repositories "repo_code2" contains unknown field'
+ ' "descripton". Refer to the definition to verify'
+ ' valid values.')
+ self.assertEqual(expectedmessage, err.__str__())
+
+ def test_repositories_with_invalid_url(self):
+ tpl_snippet = '''
+ repositories:
+ repo_code0: https://raw.githubusercontent.com/nandinivemula/intern
+ repo_code1:
+ description: My project's code Repository in github usercontent.
+ url: h
+ credential:
+ user: nandini
+ password: tcs@12345
+ repo_code2:
+ description: My Project's code Repository in github.
+ url: https://github.com/nandinivemula/intern
+ credential:
+ user: xyzw
+ password: xyz@123
+ '''
+ tpl = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet))
+ err = self.assertRaises(exception.URLException,
+ self._repo_content, tpl)
+ expectedmessage = _('repsositories "repo_code1" Invalid Url')
+ self.assertEqual(expectedmessage, err.__str__())
+
def test_groups(self):
tpl_snippet = '''
node_templates:
lambda: Policy(name, policies[name], None, None))
self.assertEqual(expectedmessage, err.__str__())
- def test_policy_trigger_valid_keyname(self):
+ def test_policy_trigger_valid_keyname_senlin_resources(self):
tpl_snippet = '''
triggers:
- resize_compute:
requirement: host
capability: Container
condition:
- constraint: utilization greater_than 50%
+ constraint: { greater_than: 50 }
period: 60
evaluations: 1
method : average
name = list(triggers.keys())[0]
Triggers(name, triggers[name])
- def test_policy_trigger_invalid_keyname(self):
+ def test_policy_trigger_valid_keyname_heat_resources(self):
+ tpl_snippet = '''
+ triggers:
+ - high_cpu_usage:
+ description: trigger
+ meter_name: cpu_util
+ condition:
+ constraint: utilization greater_than 60%
+ threshold: 60
+ period: 600
+ evaluations: 1
+ method: average
+ comparison_operator: gt
+ metadata: SG1
+ action: [SP1]
+ '''
+ triggers = (toscaparser.utils.yamlparser.
+ simple_parse(tpl_snippet))['triggers'][0]
+ name = list(triggers.keys())[0]
+ Triggers(name, triggers[name])
+
+ def test_policy_trigger_invalid_keyname_senlin_resources(self):
tpl_snippet = '''
triggers:
- resize_compute:
lambda: Triggers(name, triggers[name]))
self.assertEqual(expectedmessage, err.__str__())
+ def test_policy_trigger_invalid_keyname_heat_resources(self):
+ tpl_snippet = '''
+ triggers:
+ - high_cpu_usage:
+ description: trigger
+ meter_name: cpu_util
+ condition:
+ constraint: utilization greater_than 60%
+ threshold: 60
+ period: 600
+ evaluations: 1
+ method: average
+ comparison_operator: gt
+ metadata1: SG1
+ action: [SP1]
+ '''
+ triggers = (toscaparser.utils.yamlparser.
+ simple_parse(tpl_snippet))['triggers'][0]
+ name = list(triggers.keys())[0]
+ expectedmessage = _(
+ 'Triggers "high_cpu_usage" contains unknown field '
+ '"metadata1". Refer to the definition '
+ 'to verify valid values.')
+ err = self.assertRaises(
+ exception.UnknownFieldError,
+ lambda: Triggers(name, triggers[name]))
+ self.assertEqual(expectedmessage, err.__str__())
+
def test_policy_missing_required_keyname(self):
tpl_snippet = '''
policies:
exception.MissingRequiredFieldError,
lambda: Policy(name, policies[name], None, None))
self.assertEqual(expectedmessage, err.__str__())
+
+ def test_credential_datatype(self):
+ tosca_tpl = os.path.join(
+ os.path.dirname(os.path.abspath(__file__)),
+ "data/test_credential_datatype.yaml")
+ self.assertIsNotNone(ToscaTemplate(tosca_tpl))
+
+ def test_invalid_default_value(self):
+ tpl_path = os.path.join(
+ os.path.dirname(os.path.abspath(__file__)),
+ "data/test_invalid_input_defaults.yaml")
+ self.assertRaises(exception.ValidationError, ToscaTemplate, tpl_path)
+ exception.ExceptionCollector.assertExceptionMessage(
+ ValueError, _('"two" is not an integer.'))
+
+ def test_invalid_capability(self):
+ tpl_snippet = '''
+ node_templates:
+ server:
+ type: tosca.nodes.Compute
+ capabilities:
+ oss:
+ properties:
+ architecture: x86_64
+ '''
+ tpl = (toscaparser.utils.yamlparser.simple_parse(tpl_snippet))
+ err = self.assertRaises(exception.UnknownFieldError,
+ TopologyTemplate, tpl, None)
+ expectedmessage = _('"capabilities" of template "server" contains '
+ 'unknown field "oss". Refer to the definition '
+ 'to verify valid values.')
+ self.assertEqual(expectedmessage, err.__str__())
+
+ def test_qualified_name(self):
+ tpl_snippet_full_name = '''
+ node_templates:
+ supported_type:
+ type: tosca.nodes.Compute
+ '''
+ tpl = (
+ toscaparser.utils.yamlparser.simple_parse(
+ tpl_snippet_full_name))
+ TopologyTemplate(tpl, None)
+
+ tpl_snippet_short_name = '''
+ node_templates:
+ supported_type:
+ type: Compute
+ '''
+ tpl = (
+ toscaparser.utils.yamlparser.simple_parse(
+ tpl_snippet_short_name))
+ TopologyTemplate(tpl, None)
+
+ tpl_snippet_qualified_name = '''
+ node_templates:
+ supported_type:
+ type: tosca:Compute
+ '''
+ tpl = (
+ toscaparser.utils.yamlparser.simple_parse(
+ tpl_snippet_qualified_name))
+ TopologyTemplate(tpl, None)
+
+ def test_requirements_as_list(self):
+ """Node template with requirements provided with or without list
+
+ Node template requirements are required to be provided as list.
+ """
+
+ expectedmessage = _('"requirements" of template "my_webserver"'
+ ' must be of type "list".')
+
+ # requirements provided as dictionary
+ tpl_snippet1 = '''
+ node_templates:
+ my_webserver:
+ type: tosca.nodes.WebServer
+ requirements:
+ host: server
+ server:
+ type: tosca.nodes.Compute
+ '''
+ err1 = self.assertRaises(
+ exception.TypeMismatchError,
+ lambda: self._single_node_template_content_test(tpl_snippet1))
+ self.assertEqual(expectedmessage, err1.__str__())
+
+ # requirements provided as string
+ tpl_snippet2 = '''
+ node_templates:
+ my_webserver:
+ type: tosca.nodes.WebServer
+ requirements: server
+ server:
+ type: tosca.nodes.Compute
+ '''
+ err2 = self.assertRaises(
+ exception.TypeMismatchError,
+ lambda: self._single_node_template_content_test(tpl_snippet2))
+ self.assertEqual(expectedmessage, err2.__str__())
+
+ # requirements provided as list
+ tpl_snippet3 = '''
+ node_templates:
+ my_webserver:
+ type: tosca.nodes.WebServer
+ requirements:
+ - host: server
+ server:
+ type: tosca.nodes.Compute
+ '''
+ self.assertIsNone(
+ self._single_node_template_content_test(tpl_snippet3))
+
+ def test_properties_override_with_flavor_and_image(self):
+ tpl_path = os.path.join(
+ os.path.dirname(os.path.abspath(__file__)),
+ "data/test_normative_type_properties_override.yaml")
+ self.assertIsNotNone(ToscaTemplate(tpl_path))
Code Review / parser.git / blobdiff