Enable Translator with new TOSCA library (3)
Remove tests which are specific to TOSCA parser. They are alredy part of TOSCA Parse repository. Partially Implements: blueprint use-tosca-parser-library Change-Id: I64b948270b896b5cd24ff5dd798da8484ba9e1bf
This commit is contained in:
parent
717b855180
commit
e5f40f0022
@ -1,67 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2010-2011 OpenStack Foundation
|
||||
# Copyright (c) 2013 Hewlett-Packard Development Company, L.P.
|
||||
#
|
||||
# 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 fixtures
|
||||
import testscenarios
|
||||
import testtools
|
||||
|
||||
from toscaparser.tosca_template import ToscaTemplate
|
||||
|
||||
_TRUE_VALUES = ('True', 'true', '1', 'yes')
|
||||
|
||||
|
||||
class TestCase(testscenarios.TestWithScenarios, testtools.TestCase):
|
||||
|
||||
"""Test case base class for all unit tests."""
|
||||
|
||||
def setUp(self):
|
||||
"""Run before each test method to initialize test environment."""
|
||||
|
||||
super(TestCase, self).setUp()
|
||||
test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
|
||||
try:
|
||||
test_timeout = int(test_timeout)
|
||||
except ValueError:
|
||||
# If timeout value is invalid do not set a timeout.
|
||||
test_timeout = 0
|
||||
if test_timeout > 0:
|
||||
self.useFixture(fixtures.Timeout(test_timeout, gentle=True))
|
||||
|
||||
self.useFixture(fixtures.NestedTempfile())
|
||||
self.useFixture(fixtures.TempHomeDir())
|
||||
|
||||
if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
|
||||
stdout = self.useFixture(fixtures.StringStream('stdout')).stream
|
||||
self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
|
||||
if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
|
||||
stderr = self.useFixture(fixtures.StringStream('stderr')).stream
|
||||
self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))
|
||||
|
||||
self.log_fixture = self.useFixture(fixtures.FakeLogger())
|
||||
|
||||
def _load_template(self, filename):
|
||||
"""Load a Tosca template from tests data folder.
|
||||
|
||||
:param filename: Tosca template file name to load.
|
||||
:return: ToscaTemplate
|
||||
"""
|
||||
return ToscaTemplate(os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
'data',
|
||||
filename))
|
@ -1,336 +0,0 @@
|
||||
# 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 datetime
|
||||
import yaml
|
||||
|
||||
from toscaparser.common import exception
|
||||
from toscaparser.elements.constraints import Constraint
|
||||
from toscaparser.elements.constraints import Schema
|
||||
from toscaparser.tests.base import TestCase
|
||||
from toscaparser.utils import yamlparser
|
||||
|
||||
|
||||
class ConstraintTest(TestCase):
|
||||
|
||||
def test_schema_dict(self):
|
||||
tpl_snippet = '''
|
||||
cpus:
|
||||
type: integer
|
||||
description: Number of CPUs for the server.
|
||||
'''
|
||||
schema = yamlparser.simple_parse(tpl_snippet)
|
||||
cpus_schema = Schema('cpus', schema['cpus'])
|
||||
self.assertEqual(len(cpus_schema), 2)
|
||||
self.assertEqual('integer', cpus_schema.type)
|
||||
self.assertEqual('Number of CPUs for the server.',
|
||||
cpus_schema.description)
|
||||
self.assertEqual(True, cpus_schema.required)
|
||||
self.assertIsNone(cpus_schema.default)
|
||||
|
||||
def test_schema_not_dict(self):
|
||||
tpl_snippet = '''
|
||||
cpus:
|
||||
- type: integer
|
||||
- description: Number of CPUs for the server.
|
||||
'''
|
||||
schema = yamlparser.simple_parse(tpl_snippet)
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Schema,
|
||||
'cpus', schema['cpus'])
|
||||
self.assertEqual('Schema cpus must be a dict.', str(error))
|
||||
|
||||
def test_schema_miss_type(self):
|
||||
tpl_snippet = '''
|
||||
cpus:
|
||||
description: Number of CPUs for the server.
|
||||
'''
|
||||
schema = yamlparser.simple_parse(tpl_snippet)
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Schema,
|
||||
'cpus', schema['cpus'])
|
||||
self.assertEqual('Schema cpus must have type.', str(error))
|
||||
|
||||
def test_schema_none_description(self):
|
||||
tpl_snippet = '''
|
||||
cpus:
|
||||
type: integer
|
||||
'''
|
||||
schema = yamlparser.simple_parse(tpl_snippet)
|
||||
cpus_schema = Schema('cpus', schema['cpus'])
|
||||
self.assertEqual('', cpus_schema.description)
|
||||
|
||||
def test_invalid_constraint_type(self):
|
||||
schema = {'invalid_type': 2}
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.INTEGER,
|
||||
schema)
|
||||
self.assertEqual('Invalid constraint type "invalid_type".',
|
||||
str(error))
|
||||
|
||||
def test_invalid_prop_type(self):
|
||||
schema = {'length': 5}
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.INTEGER,
|
||||
schema)
|
||||
self.assertEqual('Constraint type "length" is not valid for '
|
||||
'data type "integer".', str(error))
|
||||
|
||||
def test_invalid_validvalues(self):
|
||||
schema = {'valid_values': 2}
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.INTEGER,
|
||||
schema)
|
||||
self.assertEqual('valid_values must be a list.', str(error))
|
||||
|
||||
def test_validvalues_validate(self):
|
||||
schema = {'valid_values': [2, 4, 6, 8]}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
self.assertIsNone(constraint.validate(4))
|
||||
|
||||
def test_validvalues_validate_fail(self):
|
||||
schema = {'valid_values': [2, 4, 6, 8]}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 5)
|
||||
self.assertEqual('prop: 5 is not an valid value "[2, 4, 6, 8]".',
|
||||
str(error))
|
||||
|
||||
def test_invalid_in_range(self):
|
||||
snippet = 'in_range: {2, 6}'
|
||||
schema = yaml.load(snippet)
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.INTEGER,
|
||||
schema)
|
||||
self.assertEqual('in_range must be a list.', str(error))
|
||||
|
||||
def test_in_range_min_max(self):
|
||||
schema = {'in_range': [2, 6]}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
self.assertEqual(2, constraint.min)
|
||||
self.assertEqual(6, constraint.max)
|
||||
|
||||
def test_in_range_validate(self):
|
||||
schema = {'in_range': [2, 6]}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
self.assertIsNone(constraint.validate(2))
|
||||
self.assertIsNone(constraint.validate(4))
|
||||
self.assertIsNone(constraint.validate(6))
|
||||
|
||||
def test_in_range_validate_fail(self):
|
||||
schema = {'in_range': [2, 6]}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 8)
|
||||
self.assertEqual('prop: 8 is out of range (min:2, max:6).',
|
||||
str(error))
|
||||
|
||||
def test_equal_validate(self):
|
||||
schema = {'equal': 4}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
self.assertIsNone(constraint.validate(4))
|
||||
|
||||
def test_equal_validate_fail(self):
|
||||
schema = {'equal': 4}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 8)
|
||||
self.assertEqual('prop: 8 is not equal to "4".', str(error))
|
||||
|
||||
def test_greater_than_validate(self):
|
||||
schema = {'greater_than': 4}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
self.assertIsNone(constraint.validate(6))
|
||||
|
||||
def test_greater_than_validate_fail(self):
|
||||
schema = {'greater_than': 4}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 3)
|
||||
self.assertEqual('prop: 3 must be greater than "4".', str(error))
|
||||
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 4)
|
||||
self.assertEqual('prop: 4 must be greater than "4".', str(error))
|
||||
|
||||
def test_greater_than_invalid(self):
|
||||
snippet = 'greater_than: {4}'
|
||||
schema = yaml.load(snippet)
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.INTEGER,
|
||||
schema)
|
||||
self.assertEqual('greater_than must be comparable.', str(error))
|
||||
|
||||
def test_greater_or_equal_validate(self):
|
||||
schema = {'greater_or_equal': 3.9}
|
||||
constraint = Constraint('prop', Schema.FLOAT, schema)
|
||||
self.assertIsNone(constraint.validate(3.9))
|
||||
self.assertIsNone(constraint.validate(4.0))
|
||||
|
||||
def test_greater_or_equal_validate_fail(self):
|
||||
schema = {'greater_or_equal': 3.9}
|
||||
constraint = Constraint('prop', Schema.FLOAT, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 3.0)
|
||||
self.assertEqual('prop: 3.0 must be greater or equal to "3.9".',
|
||||
str(error))
|
||||
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 3.8)
|
||||
self.assertEqual('prop: 3.8 must be greater or equal to "3.9".',
|
||||
str(error))
|
||||
|
||||
def test_greater_or_equal_invalid(self):
|
||||
snippet = 'greater_or_equal: {3.9}'
|
||||
schema = yaml.load(snippet)
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.INTEGER,
|
||||
schema)
|
||||
self.assertEqual('greater_or_equal must be comparable.', str(error))
|
||||
|
||||
def test_less_than_validate(self):
|
||||
schema = {'less_than': datetime.date(2014, 0o7, 25)}
|
||||
constraint = Constraint('prop', Schema.TIMESTAMP, schema)
|
||||
self.assertIsNone(constraint.validate(datetime.date(2014, 0o7, 20)))
|
||||
self.assertIsNone(constraint.validate(datetime.date(2014, 0o7, 24)))
|
||||
|
||||
def test_less_than_validate_fail(self):
|
||||
schema = {'less_than': datetime.date(2014, 0o7, 25)}
|
||||
constraint = Constraint('prop', Schema.TIMESTAMP, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate,
|
||||
datetime.date(2014, 0o7, 25))
|
||||
self.assertEqual('prop: 2014-07-25 must be '
|
||||
'less than "2014-07-25".',
|
||||
str(error))
|
||||
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate,
|
||||
datetime.date(2014, 0o7, 27))
|
||||
self.assertEqual('prop: 2014-07-27 must be '
|
||||
'less than "2014-07-25".',
|
||||
str(error))
|
||||
|
||||
def test_less_than_invalid(self):
|
||||
snippet = 'less_than: {3.9}'
|
||||
schema = yaml.load(snippet)
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.INTEGER,
|
||||
schema)
|
||||
self.assertEqual('less_than must be comparable.', str(error))
|
||||
|
||||
def test_less_or_equal_validate(self):
|
||||
schema = {'less_or_equal': 4}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
self.assertIsNone(constraint.validate(4))
|
||||
self.assertIsNone(constraint.validate(3))
|
||||
|
||||
def test_less_or_equal_validate_fail(self):
|
||||
schema = {'less_or_equal': 4}
|
||||
constraint = Constraint('prop', Schema.INTEGER, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 5)
|
||||
self.assertEqual('prop: 5 must be less or equal to "4".', str(error))
|
||||
|
||||
def test_less_or_equal_invalid(self):
|
||||
snippet = 'less_or_equal: {3.9}'
|
||||
schema = yaml.load(snippet)
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.INTEGER,
|
||||
schema)
|
||||
self.assertEqual('less_or_equal must be comparable.', str(error))
|
||||
|
||||
def test_invalid_length(self):
|
||||
schema = {'length': 'four'}
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.STRING,
|
||||
schema)
|
||||
self.assertEqual('length must be integer.', str(error))
|
||||
|
||||
schema = {'length': 4.5}
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.STRING,
|
||||
schema)
|
||||
self.assertEqual('length must be integer.', str(error))
|
||||
|
||||
def test_length_validate(self):
|
||||
schema = {'length': 4}
|
||||
constraint = Constraint('prop', Schema.STRING, schema)
|
||||
self.assertIsNone(constraint.validate('abcd'))
|
||||
|
||||
def test_length_validate_fail(self):
|
||||
schema = {'length': 4}
|
||||
constraint = Constraint('prop', Schema.STRING, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 'abc')
|
||||
self.assertEqual('length of prop: abc must be equal to "4".',
|
||||
str(error))
|
||||
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate,
|
||||
'abcde')
|
||||
self.assertEqual('length of prop: abcde must be equal to "4".',
|
||||
str(error))
|
||||
|
||||
def test_invalid_min_length(self):
|
||||
schema = {'min_length': 'four'}
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.STRING,
|
||||
schema)
|
||||
self.assertEqual('min_length must be integer.', str(error))
|
||||
|
||||
def test_min_length_validate(self):
|
||||
schema = {'min_length': 4}
|
||||
constraint = Constraint('prop', Schema.STRING, schema)
|
||||
self.assertIsNone(constraint.validate('abcd'))
|
||||
self.assertIsNone(constraint.validate('abcde'))
|
||||
|
||||
def test_min_length_validate_fail(self):
|
||||
schema = {'min_length': 4}
|
||||
constraint = Constraint('prop', Schema.STRING, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 'abc')
|
||||
self.assertEqual('length of prop: abc must be at least "4".',
|
||||
str(error))
|
||||
|
||||
def test_invalid_max_length(self):
|
||||
schema = {'max_length': 'four'}
|
||||
error = self.assertRaises(exception.InvalidSchemaError, Constraint,
|
||||
'prop', Schema.STRING,
|
||||
schema)
|
||||
self.assertEqual('max_length must be integer.', str(error))
|
||||
|
||||
def test_max_length_validate(self):
|
||||
schema = {'max_length': 4}
|
||||
constraint = Constraint('prop', Schema.STRING, schema)
|
||||
self.assertIsNone(constraint.validate('abcd'))
|
||||
self.assertIsNone(constraint.validate('abc'))
|
||||
|
||||
def test_max_length_validate_fail(self):
|
||||
schema = {'max_length': 4}
|
||||
constraint = Constraint('prop', Schema.STRING, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate,
|
||||
'abcde')
|
||||
self.assertEqual('length of prop: abcde must be no greater than "4".',
|
||||
str(error))
|
||||
|
||||
def test_pattern_validate(self):
|
||||
schema = {'pattern': '[0-9]*'}
|
||||
constraint = Constraint('prop', Schema.STRING, schema)
|
||||
self.assertIsNone(constraint.validate('123'))
|
||||
|
||||
def test_pattern_validate_fail(self):
|
||||
schema = {'pattern': '[0-9]*'}
|
||||
constraint = Constraint('prop', Schema.STRING, schema)
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
constraint.validate, 'abc')
|
||||
self.assertEqual('prop: "abc" does not match pattern "[0-9]*".',
|
||||
str(error))
|
@ -1,299 +0,0 @@
|
||||
# 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
|
||||
|
||||
from testtools.testcase import skip
|
||||
from toscaparser.common import exception
|
||||
from toscaparser.dataentity import DataEntity
|
||||
from toscaparser.elements.datatype import DataType
|
||||
from toscaparser.parameters import Input
|
||||
from toscaparser.tests.base import TestCase
|
||||
from toscaparser.tosca_template import ToscaTemplate
|
||||
from toscaparser.utils import yamlparser
|
||||
|
||||
|
||||
class DataTypeTest(TestCase):
|
||||
|
||||
custom_type_schema = '''
|
||||
tosca.my.datatypes.PeopleBase:
|
||||
properties:
|
||||
name:
|
||||
type: string
|
||||
required: true
|
||||
constraints:
|
||||
- min_length: 2
|
||||
gender:
|
||||
type: string
|
||||
default: unknown
|
||||
|
||||
tosca.my.datatypes.People:
|
||||
derived_from: tosca.my.datatypes.PeopleBase
|
||||
properties:
|
||||
addresses:
|
||||
type: map
|
||||
entry_schema:
|
||||
type: string
|
||||
contacts:
|
||||
type: list
|
||||
entry_schema:
|
||||
type: tosca.my.datatypes.ContactInfo
|
||||
|
||||
tosca.my.datatypes.ContactInfo:
|
||||
description: simple contact information
|
||||
properties:
|
||||
contact_name:
|
||||
type: string
|
||||
required: true
|
||||
constraints:
|
||||
- min_length: 2
|
||||
contact_email:
|
||||
type: string
|
||||
contact_phone:
|
||||
type: string
|
||||
'''
|
||||
custom_type_def = yamlparser.simple_parse(custom_type_schema)
|
||||
|
||||
def test_empty_template(self):
|
||||
value_snippet = ''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
self.assertEqual(value, {})
|
||||
|
||||
def test_built_in_datatype(self):
|
||||
value_snippet = '''
|
||||
private_network:
|
||||
network_name: private
|
||||
network_id: 3e54214f-5c09-1bc9-9999-44100326da1b
|
||||
addresses: [ 10.111.128.10 ]
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.datatypes.network.NetworkInfo',
|
||||
value.get('private_network'))
|
||||
self.assertIsNotNone(data.validate())
|
||||
|
||||
def test_built_in_datatype_with_short_name(self):
|
||||
value_snippet = '''
|
||||
ethernet_port:
|
||||
port_name: port1
|
||||
port_id: 2c0c7a37-691a-23a6-7709-2d10ad041467
|
||||
network_id: 3e54214f-5c09-1bc9-9999-44100326da1b
|
||||
mac_address: f1:18:3b:41:92:1e
|
||||
addresses: [ 172.24.9.102 ]
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('PortInfo', value.get('ethernet_port'))
|
||||
self.assertIsNotNone(data.validate())
|
||||
|
||||
def test_built_in_datatype_without_properties(self):
|
||||
value_snippet = '''
|
||||
2
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
datatype = DataType('PortDef')
|
||||
self.assertEqual('integer', datatype.value_type)
|
||||
data = DataEntity('PortDef', value)
|
||||
self.assertIsNotNone(data.validate())
|
||||
|
||||
@skip('The example in TOSCA spec may have some problem.')
|
||||
def test_built_in_nested_datatype(self):
|
||||
value_snippet = '''
|
||||
user_port:
|
||||
protocol: tcp
|
||||
target: [50000]
|
||||
source: [9000]
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('PortSpec', value.get('user_port'))
|
||||
self.assertIsNotNone(data.validate())
|
||||
|
||||
def test_built_in_nested_datatype_portdef(self):
|
||||
tpl_snippet = '''
|
||||
inputs:
|
||||
db_port:
|
||||
type: PortDef
|
||||
description: Port for the MySQL database
|
||||
'''
|
||||
inputs = yamlparser.simple_parse(tpl_snippet)['inputs']
|
||||
name, attrs = list(inputs.items())[0]
|
||||
input = Input(name, attrs)
|
||||
self.assertIsNone(input.validate(3360))
|
||||
try:
|
||||
input.validate(336000)
|
||||
except Exception as err:
|
||||
self.assertTrue(isinstance(err, exception.ValidationError))
|
||||
self.assertEqual('None: 336000 is out of range (min:1, '
|
||||
'max:65535).', err.__str__())
|
||||
|
||||
def test_custom_datatype(self):
|
||||
value_snippet = '''
|
||||
name: Mike
|
||||
gender: male
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.PeopleBase', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
self.assertIsNotNone(data.validate())
|
||||
|
||||
def test_custom_datatype_with_parent(self):
|
||||
value_snippet = '''
|
||||
name: Mike
|
||||
gender: male
|
||||
contacts:
|
||||
- {contact_name: Tom,
|
||||
contact_email: tom@email.com,
|
||||
contact_phone: '123456789'}
|
||||
- {contact_name: Jerry,
|
||||
contact_email: jerry@email.com,
|
||||
contact_phone: '321654987'}
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.People', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
self.assertIsNotNone(data.validate())
|
||||
|
||||
# [Tom, Jerry] is not a dict, it can't be a value of datatype PeopleBase
|
||||
def test_non_dict_value_for_datatype(self):
|
||||
value_snippet = '''
|
||||
[Tom, Jerry]
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.PeopleBase', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
error = self.assertRaises(exception.TypeMismatchError, data.validate)
|
||||
self.assertEqual('[\'Tom\', \'Jerry\'] must be of type: '
|
||||
'"tosca.my.datatypes.PeopleBase".', error.__str__())
|
||||
|
||||
# 'nema' is an invalid field name
|
||||
def test_field_error_in_dataentity(self):
|
||||
value_snippet = '''
|
||||
nema: Mike
|
||||
gender: male
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.PeopleBase', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
error = self.assertRaises(exception.UnknownFieldError, data.validate)
|
||||
self.assertEqual('Data value of type tosca.my.datatypes.PeopleBase '
|
||||
'contain(s) unknown field: "nema", refer to the '
|
||||
'definition to verify valid values.',
|
||||
error.__str__())
|
||||
|
||||
def test_default_field_in_dataentity(self):
|
||||
value_snippet = '''
|
||||
name: Mike
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.PeopleBase', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
data = data.validate()
|
||||
self.assertEqual('unknown', data.get('gender'))
|
||||
|
||||
# required field 'name' is missing
|
||||
def test_missing_field_in_dataentity(self):
|
||||
value_snippet = '''
|
||||
gender: male
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.PeopleBase', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
error = self.assertRaises(exception.MissingRequiredFieldError,
|
||||
data.validate)
|
||||
self.assertEqual('Data value of type tosca.my.datatypes.PeopleBase '
|
||||
'is missing required field: "[\'name\']".',
|
||||
error.__str__())
|
||||
|
||||
# the value of name field is not a string
|
||||
def test_type_error_in_dataentity(self):
|
||||
value_snippet = '''
|
||||
name: 123
|
||||
gender: male
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.PeopleBase', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
error = self.assertRaises(ValueError, data.validate)
|
||||
self.assertEqual('"123" is not a string', error.__str__())
|
||||
|
||||
# the value of name doesn't meet the defined constraint
|
||||
def test_value_error_in_dataentity(self):
|
||||
value_snippet = '''
|
||||
name: M
|
||||
gender: male
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.PeopleBase', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
error = self.assertRaises(exception.ValidationError, data.validate)
|
||||
self.assertEqual('length of name: M must be at least "2".',
|
||||
error.__str__())
|
||||
|
||||
# value of addresses doesn't fit the entry_schema
|
||||
def test_validation_in_collection_entry(self):
|
||||
value_snippet = '''
|
||||
name: Mike
|
||||
gender: male
|
||||
addresses: {Home: 1, Office: 9 bar avenue}
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.People', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
error = self.assertRaises(ValueError, data.validate)
|
||||
self.assertEqual('"1" is not a string', error.__str__())
|
||||
|
||||
# contact_pone is an invalid field name in nested datatype
|
||||
def test_validation_in_nested_datatype(self):
|
||||
value_snippet = '''
|
||||
name: Mike
|
||||
gender: male
|
||||
contacts:
|
||||
- {contact_name: Tom,
|
||||
contact_email: tom@email.com,
|
||||
contact_pone: '123456789'}
|
||||
- {contact_name: Jerry,
|
||||
contact_email: jerry@email.com,
|
||||
contact_phone: '321654987'}
|
||||
'''
|
||||
value = yamlparser.simple_parse(value_snippet)
|
||||
data = DataEntity('tosca.my.datatypes.People', value,
|
||||
DataTypeTest.custom_type_def)
|
||||
error = self.assertRaises(exception.UnknownFieldError, data.validate)
|
||||
self.assertEqual('Data value of type tosca.my.datatypes.ContactInfo '
|
||||
'contain(s) unknown field: "contact_pone", refer to '
|
||||
'the definition to verify valid values.',
|
||||
error.__str__())
|
||||
|
||||
def test_datatype_in_current_template(self):
|
||||
tpl_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/datatypes/test_custom_datatypes_in_current_template.yaml")
|
||||
self.assertIsNotNone(ToscaTemplate(tpl_path))
|
||||
|
||||
def test_datatype_in_template_positive(self):
|
||||
tpl_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/datatypes/test_custom_datatypes_positive.yaml")
|
||||
self.assertIsNotNone(ToscaTemplate(tpl_path))
|
||||
|
||||
def test_datatype_in_template_invalid_value(self):
|
||||
tpl_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/datatypes/test_custom_datatypes_value_error.yaml")
|
||||
error = self.assertRaises(ValueError, ToscaTemplate, tpl_path)
|
||||
self.assertEqual('"[\'1 foo street\', \'9 bar avenue\']" '
|
||||
'is not a map', error.__str__())
|
||||
|
||||
def test_datatype_in_template_nested_datatype_error(self):
|
||||
tpl_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/datatypes/test_custom_datatypes_nested_datatype_error.yaml")
|
||||
error = self.assertRaises(ValueError, ToscaTemplate, tpl_path)
|
||||
self.assertEqual('"123456789" is not a string', error.__str__())
|
@ -1,41 +0,0 @@
|
||||
# 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.
|
||||
|
||||
from toscaparser.common import exception
|
||||
from toscaparser.tests.base import TestCase
|
||||
|
||||
|
||||
class ExceptionTest(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestCase, self).setUp()
|
||||
exception.TOSCAException.set_fatal_format_exception(False)
|
||||
|
||||
def test_message(self):
|
||||
ex = exception.MissingRequiredFieldError(what='Template',
|
||||
required='type')
|
||||
self.assertEqual('Template is missing required field: "type".',
|
||||
ex.__str__())
|
||||
|
||||
def test_set_flag(self):
|
||||
exception.TOSCAException.set_fatal_format_exception('True')
|
||||
self.assertFalse(
|
||||
exception.TOSCAException._FATAL_EXCEPTION_FORMAT_ERRORS)
|
||||
|
||||
def test_format_error(self):
|
||||
ex = exception.UnknownFieldError(what='Template')
|
||||
self.assertEqual('An unknown exception occurred.', ex.__str__(),)
|
||||
self.assertRaises(KeyError, self._formate_exception)
|
||||
|
||||
def _formate_exception(self):
|
||||
exception.UnknownFieldError.set_fatal_format_exception(True)
|
||||
raise exception.UnknownFieldError(what='Template')
|
@ -1,199 +0,0 @@
|
||||
# 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
|
||||
|
||||
|
||||
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 _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 test_get_property(self):
|
||||
TestCase.skip(self, 'bug #1440247')
|
||||
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):
|
||||
TestCase.skip(self, 'bug #1440247')
|
||||
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):
|
||||
TestCase.skip(self, 'bug #1440247')
|
||||
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')
|
||||
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_in_interface(self):
|
||||
TestCase.skip(self, 'bug #1440247')
|
||||
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')
|
||||
self.assertRaises(ValueError,
|
||||
self._load_template,
|
||||
'functions/test_invalid_function_signature.yaml')
|
||||
|
||||
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())
|
||||
|
||||
|
||||
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('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']})
|
||||
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))
|
@ -1,190 +0,0 @@
|
||||
# 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.
|
||||
|
||||
from toscaparser.common import exception
|
||||
from toscaparser.properties import Property
|
||||
from toscaparser.tests.base import TestCase
|
||||
from toscaparser.utils import yamlparser
|
||||
|
||||
|
||||
class PropertyTest(TestCase):
|
||||
|
||||
def test_type(self):
|
||||
test_property_schema = {'type': 'string'}
|
||||
propertyInstance = Property('test_property', 'Hughes',
|
||||
test_property_schema)
|
||||
self.assertEqual('string', propertyInstance.type)
|
||||
|
||||
def test_type_invalid(self):
|
||||
test_property_schema = {'type': 'Fish'}
|
||||
propertyInstance = Property('test_property', 'Hughes',
|
||||
test_property_schema)
|
||||
error = self.assertRaises(exception.InvalidTypeError,
|
||||
propertyInstance.validate)
|
||||
self.assertEqual('Type "Fish" is not a valid type.', str(error))
|
||||
|
||||
def test_list(self):
|
||||
test_property_schema = {'type': 'list'}
|
||||
propertyInstance = Property('test_property', ['a', 'b'],
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual(['a', 'b'], propertyInstance.value)
|
||||
|
||||
def test_list_invalid(self):
|
||||
test_property_schema = {'type': 'list'}
|
||||
propertyInstance = Property('test_property', 'a',
|
||||
test_property_schema)
|
||||
error = self.assertRaises(ValueError, propertyInstance.validate)
|
||||
self.assertEqual('"a" is not a list', str(error))
|
||||
|
||||
def test_list_entry_schema(self):
|
||||
test_property_schema = {'type': 'list',
|
||||
'entry_schema': {'type': 'string'}}
|
||||
propertyInstance = Property('test_property', ['a', 'b'],
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual(['a', 'b'], propertyInstance.value)
|
||||
|
||||
schema_snippet = '''
|
||||
type: list
|
||||
entry_schema:
|
||||
type: string
|
||||
constraints:
|
||||
- min_length: 2
|
||||
'''
|
||||
test_property_schema = yamlparser.simple_parse(schema_snippet)
|
||||
propertyInstance = Property('test_property', ['ab', 'cd'],
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual(['ab', 'cd'], propertyInstance.value)
|
||||
|
||||
def test_list_entry_schema_invalid(self):
|
||||
test_property_schema = {'type': 'list',
|
||||
'entry_schema': {'type': 'integer'}}
|
||||
propertyInstance = Property('test_property', [1, 'b'],
|
||||
test_property_schema)
|
||||
error = self.assertRaises(ValueError, propertyInstance.validate)
|
||||
self.assertEqual('"b" is not an integer',
|
||||
str(error))
|
||||
|
||||
def test_map(self):
|
||||
test_property_schema = {'type': 'map'}
|
||||
propertyInstance = Property('test_property', {'a': 'b'},
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual({'a': 'b'}, propertyInstance.value)
|
||||
|
||||
def test_map_invalid(self):
|
||||
test_property_schema = {'type': 'map'}
|
||||
propertyInstance = Property('test_property', 12,
|
||||
test_property_schema)
|
||||
error = self.assertRaises(ValueError, propertyInstance.validate)
|
||||
self.assertEqual('"12" is not a map', str(error))
|
||||
|
||||
def test_map_entry_schema(self):
|
||||
test_property_schema = {'type': 'map',
|
||||
'entry_schema': {'type': 'boolean'}}
|
||||
propertyInstance = Property('test_property',
|
||||
{'valid': True, 'required': True},
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual({'valid': True, 'required': True},
|
||||
propertyInstance.value)
|
||||
|
||||
def test_map_entry_schema_invalid(self):
|
||||
test_property_schema = {'type': 'map',
|
||||
'entry_schema': {'type': 'boolean'}}
|
||||
propertyInstance = Property('test_property',
|
||||
{'valid': True, 'contact_name': 123},
|
||||
test_property_schema)
|
||||
error = self.assertRaises(ValueError, propertyInstance.validate)
|
||||
self.assertEqual('"123" is not a boolean', str(error))
|
||||
|
||||
def test_boolean(self):
|
||||
test_property_schema = {'type': 'boolean'}
|
||||
propertyInstance = Property('test_property', 'true',
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
propertyInstance = Property('test_property', True,
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual(True, propertyInstance.value)
|
||||
|
||||
def test_boolean_invalid(self):
|
||||
test_property_schema = {'type': 'boolean'}
|
||||
propertyInstance = Property('test_property', 12,
|
||||
test_property_schema)
|
||||
error = self.assertRaises(ValueError, propertyInstance.validate)
|
||||
self.assertEqual('"12" is not a boolean', str(error))
|
||||
|
||||
def test_float(self):
|
||||
test_property_schema = {'type': 'float'}
|
||||
propertyInstance = Property('test_property', 0.1,
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual(0.1, propertyInstance.value)
|
||||
|
||||
def test_float_invalid(self):
|
||||
test_property_schema = {'type': 'float'}
|
||||
propertyInstance = Property('test_property', 12,
|
||||
test_property_schema)
|
||||
error = self.assertRaises(ValueError, propertyInstance.validate)
|
||||
self.assertEqual('"12" is not a float', str(error))
|
||||
|
||||
def test_timestamp(self):
|
||||
test_property_schema = {'type': 'timestamp'}
|
||||
# canonical timestamp
|
||||
propertyInstance = Property('test_property', '2015-04-01T02:59:43.1Z',
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual("2015-04-01T02:59:43.1Z", propertyInstance.value)
|
||||
|
||||
# iso8601 timestamp
|
||||
propertyInstance = Property('test_property',
|
||||
'2015-04-01t21:59:43.10-05:00',
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual("2015-04-01t21:59:43.10-05:00",
|
||||
propertyInstance.value)
|
||||
|
||||
# space separated timestamp
|
||||
propertyInstance = Property('test_property',
|
||||
'2015-04-01 21:59:43.10 -5',
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual("2015-04-01 21:59:43.10 -5", propertyInstance.value)
|
||||
|
||||
# no time zone timestamp
|
||||
propertyInstance = Property('test_property', '2015-04-01 21:59:43.10',
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual("2015-04-01 21:59:43.10", propertyInstance.value)
|
||||
|
||||
# date (00:00:00Z)
|
||||
propertyInstance = Property('test_property', '2015-04-01',
|
||||
test_property_schema)
|
||||
self.assertIsNone(propertyInstance.validate())
|
||||
self.assertEqual("2015-04-01", propertyInstance.value)
|
||||
|
||||
def test_timestamp_invalid(self):
|
||||
test_property_schema = {'type': 'timestamp'}
|
||||
# invalid timestamp - day out of range
|
||||
propertyInstance = Property('test_property', '2015-04-115T02:59:43.1Z',
|
||||
test_property_schema)
|
||||
error = self.assertRaises(ValueError, propertyInstance.validate)
|
||||
self.assertEqual('day is out of range for month', str(error))
|
||||
|
||||
def test_required(self):
|
||||
test_property_schema = {'type': 'string'}
|
||||
propertyInstance = Property('test_property', 'Foo',
|
||||
test_property_schema)
|
||||
self.assertEqual(True, propertyInstance.required)
|
@ -1,352 +0,0 @@
|
||||
# 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.
|
||||
|
||||
from toscaparser.common import exception
|
||||
from toscaparser.elements.scalarunit import ScalarUnit_Frequency
|
||||
from toscaparser.elements.scalarunit import ScalarUnit_Size
|
||||
from toscaparser.elements.scalarunit import ScalarUnit_Time
|
||||
from toscaparser.nodetemplate import NodeTemplate
|
||||
from toscaparser.tests.base import TestCase
|
||||
from toscaparser.utils import yamlparser
|
||||
|
||||
|
||||
class ScalarUnitPositiveTest(TestCase):
|
||||
|
||||
scenarios = [
|
||||
(
|
||||
# tpl_snippet with mem_size given as number+space+MB
|
||||
'mem_size_is_number_Space_MB',
|
||||
dict(tpl_snippet='''
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
mem_size: 1024 MB
|
||||
''',
|
||||
property='mem_size',
|
||||
expected='1024 MB')
|
||||
),
|
||||
(
|
||||
# tpl_snippet with mem_size given as number+spaces+GB
|
||||
'mem_size_is_number_Space_GB',
|
||||
dict(tpl_snippet='''
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
mem_size: 1 GB
|
||||
''',
|
||||
property='mem_size',
|
||||
expected='1 GB')
|
||||
),
|
||||
(
|
||||
# tpl_snippet with mem_size given as number+tiB
|
||||
'mem_size_is_number_NoSpace_GB',
|
||||
dict(tpl_snippet='''
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
mem_size: 1tiB
|
||||
''',
|
||||
property='mem_size',
|
||||
expected='1 TiB')
|
||||
),
|
||||
(
|
||||
# tpl_snippet with mem_size given as number+Spaces+GIB
|
||||
'mem_size_is_number_Spaces_GB',
|
||||
dict(tpl_snippet='''
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
mem_size: 1 GIB
|
||||
''',
|
||||
property='mem_size',
|
||||
expected='1 GiB')
|
||||
),
|
||||
(
|
||||
# tpl_snippet with mem_size given as number+Space+tib
|
||||
'mem_size_is_number_Spaces_GB',
|
||||
dict(tpl_snippet='''
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
mem_size: 1 tib
|
||||
''',
|
||||
property='mem_size',
|
||||
expected='1 TiB')
|
||||
),
|
||||
(
|
||||
'cpu_frequency_is_float_Space_GHz',
|
||||
dict(tpl_snippet='''
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
cpu_frequency: 2.5 GHz
|
||||
''',
|
||||
property='cpu_frequency',
|
||||
expected='2.5 GHz')
|
||||
),
|
||||
(
|
||||
'cpu_frequency_is_float_Space_MHz',
|
||||
dict(tpl_snippet='''
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
cpu_frequency: 800 MHz
|
||||
''',
|
||||
property='cpu_frequency',
|
||||
expected='800 MHz')
|
||||
),
|
||||
]
|
||||
|
||||
def test_scenario_scalar_unit_positive(self):
|
||||
tpl = self.tpl_snippet
|
||||
nodetemplates = yamlparser.simple_parse(tpl)
|
||||
nodetemplate = NodeTemplate('server', nodetemplates)
|
||||
props = nodetemplate.get_capability('host').get_properties()
|
||||
prop_name = self.property
|
||||
if props and prop_name in props.keys():
|
||||
prop = props[prop_name]
|
||||
self.assertIsNone(prop.validate())
|
||||
resolved = prop.value
|
||||
self.assertEqual(resolved, self.expected)
|
||||
|
||||
|
||||
class GetNumFromScalarUnitSizePositive(TestCase):
|
||||
|
||||
scenarios = [
|
||||
( # Note that (1 TB) / (1 GB) = 1000
|
||||
'Input is TB, user input is GB',
|
||||
dict(InputMemSize='1 TB',
|
||||
UserInputUnit='gB',
|
||||
expected=1000)
|
||||
),
|
||||
( # Note that (1 Tib)/ (1 GB) = 1099
|
||||
'Input is TiB, user input is GB',
|
||||
dict(InputMemSize='1 TiB',
|
||||
UserInputUnit='gB',
|
||||
expected=1099.511627776)
|
||||
),
|
||||
]
|
||||
|
||||
def test_scenario_get_num_from_scalar_unit_size(self):
|
||||
resolved = (ScalarUnit_Size(self.InputMemSize).
|
||||
get_num_from_scalar_unit(self.UserInputUnit))
|
||||
self.assertEqual(resolved, self.expected)
|
||||
|
||||
|
||||
class GetNumFromScalarUnitFrequencyPositive(TestCase):
|
||||
|
||||
scenarios = [
|
||||
( # Note that (1 GHz) / (1 Hz) = 1000000000
|
||||
'Input is GHz, user input is Hz',
|
||||
dict(InputMemSize='1 GHz',
|
||||
UserInputUnit='Hz',
|
||||
expected=1000000000)
|
||||
),
|
||||
(
|
||||
'Input is GHz, user input is Hz',
|
||||
dict(InputMemSize='2.4 GHz',
|
||||
UserInputUnit='Hz',
|
||||
expected=2400000000)
|
||||
),
|
||||
( # Note that (1 GHz)/ (1 MHz) = 1000
|
||||
'Input is MHz, user input is GHz',
|
||||
dict(InputMemSize='800 MHz',
|
||||
UserInputUnit='GHz',
|
||||
expected=0.8)
|
||||
),
|
||||
(
|
||||
'Input is GHz, user input is Hz',
|
||||
dict(InputMemSize='0.9 GHz',
|
||||
UserInputUnit='MHz',
|
||||
expected=900)
|
||||
),
|
||||
(
|
||||
'Input is GHz, user input is Hz',
|
||||
dict(InputMemSize='2.7GHz',
|
||||
UserInputUnit='MHz',
|
||||
expected=2700)
|
||||
),
|
||||
]
|
||||
|
||||
def test_scenario_get_num_from_scalar_unit_frequency(self):
|
||||
resolved = (ScalarUnit_Frequency(self.InputMemSize).
|
||||
get_num_from_scalar_unit(self.UserInputUnit))
|
||||
self.assertEqual(resolved, self.expected)
|
||||
|
||||
|
||||
class GetNumFromScalarUnitTimePositive(TestCase):
|
||||
|
||||
scenarios = [
|
||||
( # Note that (1 s) / (1 ms) = 1000
|
||||
'Input is 500ms, user input is s',
|
||||
dict(InputMemSize='500 ms',
|
||||
UserInputUnit='s',
|
||||
expected=0.5)
|
||||
),
|
||||
( # Note that (1 h)/ (1 s) = 3600
|
||||
'Input is h, user input is s',
|
||||
dict(InputMemSize='1 h',
|
||||
UserInputUnit='s',
|
||||
expected=3600)
|
||||
),
|
||||
( # Note that (1 m)/ (1 s) = 60
|
||||
'Input is m, user input is s',
|
||||
dict(InputMemSize='0.5 m',
|
||||
UserInputUnit='s',
|
||||
expected=30)
|
||||
),
|
||||
( # Note that (1 d)/ (1 h) = 24
|
||||
'Input is d, user input is h',
|
||||
dict(InputMemSize='1 d',
|
||||
UserInputUnit='h',
|
||||
expected=24)
|
||||
),
|
||||
]
|
||||
|
||||
def test_scenario_get_num_from_scalar_unit_time(self):
|
||||
resolved = (ScalarUnit_Time(self.InputMemSize).
|
||||
get_num_from_scalar_unit(self.UserInputUnit))
|
||||
self.assertEqual(resolved, self.expected)
|
||||
|
||||
|
||||
class GetNumFromScalarUnitSizeNegative(TestCase):
|
||||
|
||||
InputMemSize = '1 GB'
|
||||
UserInputUnit = 'qB'
|
||||
|
||||
def test_get_num_from_scalar_unit_size_negative(self):
|
||||
try:
|
||||
(ScalarUnit_Size(self.InputMemSize).
|
||||
get_num_from_scalar_unit(self.UserInputUnit))
|
||||
except Exception as error:
|
||||
self.assertTrue(isinstance(error, ValueError))
|
||||
self.assertEqual('Provided unit "qB" is not valid. The valid units'
|
||||
' are [\'B\', \'GB\', \'GiB\', \'KiB\', \'MB\','
|
||||
' \'MiB\', \'TB\', \'TiB\', \'kB\']',
|
||||
error.__str__())
|
||||
|
||||
|
||||
class GetNumFromScalarUnitFrequencyNegative(TestCase):
|
||||
|
||||
InputFrequency = '2.7 GHz'
|
||||
UserInputUnit = 'Jz'
|
||||
|
||||
def test_get_num_from_scalar_unit_frequency_negative(self):
|
||||
try:
|
||||
(ScalarUnit_Frequency(self.InputFrequency).
|
||||
get_num_from_scalar_unit(self.UserInputUnit))
|
||||
except Exception as error:
|
||||
self.assertTrue(isinstance(error, ValueError))
|
||||
self.assertEqual('Provided unit "Jz" is not valid. The valid'
|
||||
' units are [\'GHz\', \'Hz\', \'MHz\', \'kHz\']',
|
||||
error.__str__())
|
||||
|
||||
|
||||
class GetNumFromScalarUnitTimeNegative(TestCase):
|
||||
|
||||
InputTime = '5 ms'
|
||||
UserInputUnit = 'D'
|
||||
|
||||
def test_get_num_from_scalar_unit_frequency_negative(self):
|
||||
try:
|
||||
(ScalarUnit_Time(self.InputTime).
|
||||
get_num_from_scalar_unit(self.UserInputUnit))
|
||||
except Exception as error:
|
||||
self.assertTrue(isinstance(error, ValueError))
|
||||
self.assertEqual('"Jz" is not a valid scalar-unit',
|
||||
error.__str__())
|
||||
|
||||
|
||||
class ScalarUnitNegativeTest(TestCase):
|
||||
|
||||
custom_def_snippet = '''
|
||||
tosca.my.nodes.Compute:
|
||||
derived_from: tosca.nodes.Root
|
||||
properties:
|
||||
cpu_frequency:
|
||||
required: no
|
||||
type: scalar-unit.frequency
|
||||
constraints:
|
||||
- greater_or_equal: 0.1 GHz
|
||||
disk_size:
|
||||
required: no
|
||||
type: scalar-unit.size
|
||||
constraints:
|
||||
- greater_or_equal: 1 GB
|
||||
mem_size:
|
||||
required: no
|
||||
type: scalar-unit.size
|
||||
constraints:
|
||||
- in_range: [1 MiB, 1 GiB]
|
||||
'''
|
||||
custom_def = yamlparser.simple_parse(custom_def_snippet)
|
||||
|
||||
# disk_size doesn't provide a value, mem_size uses an invalid unit.
|
||||
def test_invalid_scalar_unit(self):
|
||||
tpl_snippet = '''
|
||||
server:
|
||||
type: tosca.my.nodes.Compute
|
||||
properties:
|
||||
cpu_frequency: 50.3.6 GHZ
|
||||
disk_size: MB
|
||||
mem_size: 1 QB
|
||||
'''
|
||||
nodetemplates = yamlparser.simple_parse(tpl_snippet)
|
||||
nodetemplate = NodeTemplate('server', nodetemplates, self.custom_def)
|
||||
for p in nodetemplate.get_properties_objects():
|
||||
self.assertRaises(ValueError, p.validate)
|
||||
|
||||
# disk_size is less than 1 GB, mem_size is not in the required range.
|
||||
# Note: in the spec, the minimum value of mem_size is 1 MiB (> 1 MB)
|
||||
def test_constraint_for_scalar_unit(self):
|
||||
tpl_snippet = '''
|
||||
server:
|
||||
type: tosca.my.nodes.Compute
|
||||
properties:
|
||||
cpu_frequency: 0.05 GHz
|
||||
disk_size: 500 MB
|
||||
mem_size: 1 MB
|
||||
'''
|
||||
nodetemplates = yamlparser.simple_parse(tpl_snippet)
|
||||
nodetemplate = NodeTemplate('server', nodetemplates, self.custom_def)
|
||||
props = nodetemplate.get_properties()
|
||||
if 'cpu_frequency' in props.keys():
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
props['cpu_frequency'].validate)
|
||||
self.assertEqual('cpu_frequency: 0.05 GHz must be greater or '
|
||||
'equal to "0.1 GHz".', error.__str__())
|
||||
if 'disk_size' in props.keys():
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
props['disk_size'].validate)
|
||||
self.assertEqual('disk_size: 500 MB must be greater or '
|
||||
'equal to "1 GB".', error.__str__())
|
||||
|
||||
if 'mem_size' in props.keys():
|
||||
error = self.assertRaises(exception.ValidationError,
|
||||
props['mem_size'].validate)
|
||||
self.assertEqual('mem_size: 1 MB is out of range '
|
||||
'(min:1 MiB, '
|
||||
'max:1 GiB).', error.__str__())
|
@ -1,153 +0,0 @@
|
||||
# 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
|
||||
|
||||
from toscaparser.tests.base import TestCase
|
||||
from toscaparser.topology_template import TopologyTemplate
|
||||
import toscaparser.utils.yamlparser
|
||||
|
||||
YAML_LOADER = toscaparser.utils.yamlparser.load_yaml
|
||||
|
||||
|
||||
class TopologyTemplateTest(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
TestCase.setUp(self)
|
||||
'''TOSCA template.'''
|
||||
self.tosca_tpl_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/topology_template/subsystem.yaml")
|
||||
self.tpl = YAML_LOADER(self.tosca_tpl_path)
|
||||
self.topo_tpl = self.tpl.get('topology_template')
|
||||
self.imports = self.tpl.get('imports')
|
||||
self.topo = TopologyTemplate(self.topo_tpl,
|
||||
self._get_all_custom_def())
|
||||
|
||||
def _get_custom_def(self, type_definition):
|
||||
custom_defs = {}
|
||||
for definition in self.imports:
|
||||
if os.path.isabs(definition):
|
||||
def_file = definition
|
||||
else:
|
||||
tpl_dir = os.path.dirname(os.path.abspath(self.tosca_tpl_path))
|
||||
def_file = os.path.join(tpl_dir, definition)
|
||||
custom_type = YAML_LOADER(def_file)
|
||||
custom_defs.update(custom_type.get(type_definition))
|
||||
return custom_defs
|
||||
|
||||
def _get_all_custom_def(self):
|
||||
custom_defs = {}
|
||||
custom_defs.update(self._get_custom_def('node_types'))
|
||||
custom_defs.update(self._get_custom_def('capability_types'))
|
||||
return custom_defs
|
||||
|
||||
def test_description(self):
|
||||
expected_desc = 'Template of a database including its hosting stack.'
|
||||
self.assertEqual(expected_desc, self.topo.description)
|
||||
|
||||
def test_inputs(self):
|
||||
self.assertEqual(
|
||||
['mq_server_ip', 'my_cpus', 'receiver_port'],
|
||||
sorted([input.name for input in self.topo.inputs]))
|
||||
|
||||
input_name = "receiver_port"
|
||||
expected_description = "Port to be used for receiving messages."
|
||||
for input in self.topo.inputs:
|
||||
if input.name == input_name:
|
||||
self.assertEqual(expected_description, input.description)
|
||||
|
||||
def test_node_tpls(self):
|
||||
'''Test nodetemplate names.'''
|
||||
self.assertEqual(
|
||||
['app', 'server', 'websrv'],
|
||||
sorted([tpl.name for tpl in self.topo.nodetemplates]))
|
||||
|
||||
tpl_name = "app"
|
||||
expected_type = "example.SomeApp"
|
||||
expected_properties = ['admin_user', 'pool_size']
|
||||
expected_capabilities = ['message_receiver']
|
||||
expected_requirements = [{'host': {'node': 'websrv'}}]
|
||||
expected_relationshp = ['tosca.relationships.HostedOn']
|
||||
expected_host = ['websrv']
|
||||
for tpl in self.topo.nodetemplates:
|
||||
if tpl_name == tpl.name:
|
||||
'''Test node type.'''
|
||||
self.assertEqual(tpl.type, expected_type)
|
||||
|
||||
'''Test properties.'''
|
||||
self.assertEqual(
|
||||
expected_properties,
|
||||
sorted(tpl.get_properties().keys()))
|
||||
|
||||
'''Test capabilities.'''
|
||||
self.assertEqual(
|
||||
expected_capabilities,
|
||||
sorted(tpl.get_capabilities().keys()))
|
||||
|
||||
'''Test requirements.'''
|
||||
self.assertEqual(
|
||||
expected_requirements, tpl.requirements)
|
||||
|
||||
'''Test relationship.'''
|
||||
''' TODO : skip tempororily. need to fix it
|
||||
'''
|
||||
self.assertEqual(
|
||||
expected_relationshp,
|
||||
[x.type for x in tpl.relationships.keys()])
|
||||
self.assertEqual(
|
||||
expected_host,
|
||||
[y.name for y in tpl.relationships.values()])
|
||||
'''Test interfaces.'''
|
||||
# TODO(hurf) add interface test when new template is available
|
||||
|
||||
if tpl.name == 'server':
|
||||
'''Test property value'''
|
||||
props = tpl.get_properties()
|
||||
if props and 'mem_size' in props.keys():
|
||||
self.assertEqual(props['mem_size'].value, '4096 MB')
|
||||
'''Test capability'''
|
||||
caps = tpl.get_capabilities()
|
||||
self.assertIn('os', caps.keys())
|
||||
os_props_objs = None
|
||||
os_props = None
|
||||
os_type_prop = None
|
||||
if caps and 'os' in caps.keys():
|
||||
capability = caps['os']
|
||||
os_props_objs = capability.get_properties_objects()
|
||||
os_props = capability.get_properties()
|
||||
os_type_prop = capability.get_property_value('type')
|
||||
break
|
||||
self.assertEqual(
|
||||
['Linux'],
|
||||
[p.value for p in os_props_objs if p.name == 'type'])
|
||||
self.assertEqual(
|
||||
'Linux',
|
||||
os_props['type'].value if 'type' in os_props else '')
|
||||
self.assertEqual('Linux', os_props['type'].value)
|
||||
self.assertEqual('Linux', os_type_prop)
|
||||
|
||||
def test_outputs(self):
|
||||
self.assertEqual(
|
||||
['receiver_ip'],
|
||||
sorted([output.name for output in self.topo.outputs]))
|
||||
|
||||
def test_groups(self):
|
||||
group = self.topo.groups[0]
|
||||
self.assertEqual('webserver_group', group.name)
|
||||
self.assertEqual(['websrv', 'server'], group.member_names)
|
||||
for node in group.members:
|
||||
if node.name == 'server':
|
||||
'''Test property value'''
|
||||
props = node.get_properties()
|
||||
if props and 'mem_size' in props.keys():
|
||||
self.assertEqual(props['mem_size'].value, '4096 MB')
|
@ -1,260 +0,0 @@
|
||||
# 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.
|
||||
|
||||
from toscaparser.common import exception
|
||||
from toscaparser.elements.artifacttype import ArtifactTypeDef
|
||||
import toscaparser.elements.interfaces as ifaces
|
||||
from toscaparser.elements.nodetype import NodeType
|
||||
from toscaparser.tests.base import TestCase
|
||||
|
||||
compute_type = NodeType('tosca.nodes.Compute')
|
||||
component_type = NodeType('tosca.nodes.SoftwareComponent')
|
||||
network_type = NodeType('tosca.nodes.network.Network')
|
||||
network_port_type = NodeType('tosca.nodes.network.Port')
|
||||
webserver_type = NodeType('tosca.nodes.WebServer')
|
||||
database_type = NodeType('tosca.nodes.Database')
|
||||
artif_root_type = ArtifactTypeDef('tosca.artifacts.Root')
|
||||
artif_file_type = ArtifactTypeDef('tosca.artifacts.File')
|
||||
artif_bash_type = ArtifactTypeDef('tosca.artifacts.Implementation.Bash')
|
||||
artif_python_type = ArtifactTypeDef('tosca.artifacts.Implementation.Python')
|
||||
artif_container_docker_type = ArtifactTypeDef('tosca.artifacts.'
|
||||
'Deployment.Image.'
|
||||
'Container.Docker')
|
||||
artif_vm_iso_type = ArtifactTypeDef('tosca.artifacts.'
|
||||
'Deployment.Image.VM.ISO')
|
||||
artif_vm_qcow2_type = ArtifactTypeDef('tosca.artifacts.'
|
||||
'Deployment.Image.VM.QCOW2')
|
||||
|
||||
|
||||
class ToscaDefTest(TestCase):
|
||||
def test_type(self):
|
||||
self.assertEqual(compute_type.type, "tosca.nodes.Compute")
|
||||
self.assertRaises(exception.InvalidTypeError, NodeType,
|
||||
'tosca.nodes.Invalid')
|
||||
self.assertEqual(network_type.type, "tosca.nodes.network.Network")
|
||||
self.assertEqual(network_port_type.type, "tosca.nodes.network.Port")
|
||||
|
||||
def test_parent_type(self):
|
||||
self.assertEqual(compute_type.parent_type.type, "tosca.nodes.Root")
|
||||
self.assertEqual(network_type.parent_type.type, "tosca.nodes.Root")
|
||||
self.assertEqual(network_port_type.parent_type.type,
|
||||
"tosca.nodes.Root")
|
||||
|
||||
def test_capabilities(self):
|
||||
self.assertEqual(
|
||||
sorted(['tosca.capabilities.Container',
|
||||
'tosca.capabilities.OperatingSystem',
|
||||
'tosca.capabilities.network.Bindable',
|
||||
'tosca.capabilities.Scalable']),
|
||||
sorted([c.type for c in compute_type.get_capabilities_objects()]))
|
||||
self.assertEqual(
|
||||
['tosca.capabilities.network.Linkable'],
|
||||
[c.type for c in network_type.get_capabilities_objects()])
|
||||
endpoint_properties = ['initiator', 'network_name', 'port',
|
||||
'port_name', 'ports', 'protocol',
|
||||
'secure', 'url_path']
|
||||
endpoint_props_def_objects = \
|
||||
self._get_capability_properties_def_objects(
|
||||
webserver_type.get_capabilities_objects(),
|
||||
'tosca.capabilities.Endpoint')
|
||||
self.assertEqual(
|
||||
endpoint_properties,
|
||||
sorted([p.name for p in endpoint_props_def_objects]))
|
||||
for p in endpoint_props_def_objects:
|
||||
if p.name in endpoint_properties:
|
||||
self.assertFalse(p.required)
|
||||
endpoint_props_def = self._get_capability_properties_def(
|
||||
webserver_type.get_capabilities_objects(),
|
||||
'tosca.capabilities.Endpoint')
|
||||
self.assertEqual(
|
||||
endpoint_properties,
|
||||
sorted(endpoint_props_def.keys()))
|
||||
endpoint_prop_def = self._get_capability_property_def(
|
||||
webserver_type.get_capabilities_objects(),
|
||||
'tosca.capabilities.Endpoint',
|
||||
'initiator')
|
||||
self.assertEqual(None, endpoint_prop_def)
|
||||
|
||||
os_props = self._get_capability_properties_def_objects(
|
||||
compute_type.get_capabilities_objects(),
|
||||
'tosca.capabilities.OperatingSystem')
|
||||
self.assertEqual(
|
||||
[('architecture', False), ('distribution', False), ('type', False),
|
||||
('version', False)],
|
||||
sorted([(p.name, p.required) for p in os_props]))
|
||||
|
||||
host_props = self._get_capability_properties_def_objects(
|
||||
compute_type.get_capabilities_objects(),
|
||||
'tosca.capabilities.Container')
|
||||
self.assertEqual(
|
||||
[('cpu_frequency', False), ('disk_size', False),
|
||||
('mem_size', False), ('num_cpus', False)],
|
||||
sorted([(p.name, p.required) for p in host_props]))
|
||||
|
||||
def _get_capability_properties_def_objects(self, caps, type):
|
||||
properties_def = None
|
||||
for cap in caps:
|
||||
if cap.type == type:
|
||||
properties_def = cap.get_properties_def_objects()
|
||||
break
|
||||
return properties_def
|
||||
|
||||
def _get_capability_properties_def(self, caps, type):
|
||||
properties_def = None
|
||||
for cap in caps:
|
||||
if cap.type == type:
|
||||
properties_def = cap.get_properties_def()
|
||||
break
|
||||
return properties_def
|
||||
|
||||
def _get_capability_property_def(self, caps, type, property):
|
||||
property_def = None
|
||||
for cap in caps:
|
||||
if cap.type == type:
|
||||
property_def = cap.get_property_def_value(property)
|
||||
break
|
||||
return property_def
|
||||
|
||||
def test_properties_def(self):
|
||||
self.assertEqual(
|
||||
['name', 'password', 'user'],
|
||||
sorted(database_type.get_properties_def().keys()))
|
||||
|
||||
def test_attributes_def(self):
|
||||
self.assertEqual(
|
||||
['private_address', 'public_address'],
|
||||
sorted(compute_type.get_attributes_def().keys()))
|
||||
|
||||
def test_requirements(self):
|
||||
self.assertEqual(
|
||||
[{'host': {'capability': 'tosca.capabilities.Container',
|
||||
'node': 'tosca.nodes.Compute',
|
||||
'relationship': 'tosca.relationships.HostedOn'}}],
|
||||
[r for r in component_type.requirements])
|
||||
|
||||
def test_relationship(self):
|
||||
self.assertEqual(
|
||||
[('tosca.relationships.HostedOn', 'tosca.nodes.Compute')],
|
||||
[(relation.type, node.type) for
|
||||
relation, node in component_type.relationship.items()])
|
||||
self.assertIn(
|
||||
('tosca.relationships.HostedOn', ['tosca.capabilities.Container']),
|
||||
[(relation.type, relation.valid_target_types) for
|
||||
relation in list(component_type.relationship.keys())])
|
||||
self.assertIn(
|
||||
('tosca.relationships.network.BindsTo', 'tosca.nodes.Compute'),
|
||||
[(relation.type, node.type) for
|
||||
relation, node in network_port_type.relationship.items()])
|
||||
self.assertIn(
|
||||
('tosca.relationships.network.LinksTo',
|
||||
'tosca.nodes.network.Network'),
|
||||
[(relation.type, node.type) for
|
||||
relation, node in network_port_type.relationship.items()])
|
||||
|
||||
def test_interfaces(self):
|
||||
self.assertEqual(compute_type.interfaces, None)
|
||||
root_node = NodeType('tosca.nodes.Root')
|
||||
self.assertIn(ifaces.LIFECYCLE_SHORTNAME, root_node.interfaces)
|
||||
|
||||
def test_artifacts(self):
|
||||
self.assertEqual('tosca.artifacts.Root',
|
||||
artif_file_type.parent_type)
|
||||
self.assertEqual({}, artif_file_type.parent_artifacts)
|
||||
self.assertEqual(sorted(['tosca.artifacts.Root'],
|
||||
key=lambda x: str(x)),
|
||||
sorted([artif_file_type.get_artifact(name)
|
||||
for name in artif_file_type.defs],
|
||||
key=lambda x: str(x)))
|
||||
|
||||
self.assertEqual('tosca.artifacts.Implementation',
|
||||
artif_bash_type.parent_type)
|
||||
self.assertEqual({'tosca.artifacts.Implementation':
|
||||
{'derived_from': 'tosca.artifacts.Root',
|
||||
'description':
|
||||
'TOSCA base type for implementation artifacts'}},
|
||||
artif_bash_type.parent_artifacts)
|
||||
self.assertEqual(sorted([['sh'], 'tosca.artifacts.Implementation',
|
||||
'Script artifact for the Unix Bash shell',
|
||||
'application/x-sh'], key=lambda x: str(x)),
|
||||
sorted([artif_bash_type.get_artifact(name)
|
||||
for name in artif_bash_type.defs],
|
||||
key=lambda x: str(x)))
|
||||
|
||||
self.assertEqual('tosca.artifacts.Implementation',
|
||||
artif_python_type.parent_type)
|
||||
self.assertEqual({'tosca.artifacts.Implementation':
|
||||
{'derived_from': 'tosca.artifacts.Root',
|
||||
'description':
|
||||
'TOSCA base type for implementation artifacts'}},
|
||||
artif_python_type.parent_artifacts)
|
||||
self.assertEqual(sorted([['py'], 'tosca.artifacts.Implementation',
|
||||
'Artifact for the interpreted Python'
|
||||
' language', 'application/x-python'],
|
||||
key=lambda x: str(x)),
|
||||
sorted([artif_python_type.get_artifact(name)
|
||||
for name in artif_python_type.defs],
|
||||
key=lambda x: str(x)))
|
||||
|
||||
self.assertEqual('tosca.artifacts.Deployment.Image',
|
||||
artif_container_docker_type.parent_type)
|
||||
self.assertEqual({'tosca.artifacts.Deployment':
|
||||
{'derived_from': 'tosca.artifacts.Root',
|
||||
'description':
|
||||
'TOSCA base type for deployment artifacts'},
|
||||
'tosca.artifacts.Deployment.Image':
|
||||
{'derived_from': 'tosca.artifacts.Deployment'}},
|
||||
artif_container_docker_type.parent_artifacts)
|
||||
self.assertEqual(sorted(['tosca.artifacts.Deployment.Image',
|
||||
'Docker container image'],
|
||||
key=lambda x: str(x)),
|
||||
sorted([artif_container_docker_type.
|
||||
get_artifact(name) for name in
|
||||
artif_container_docker_type.defs],
|
||||
key=lambda x: str(x)))
|
||||
|
||||
self.assertEqual('tosca.artifacts.Deployment.Image',
|
||||
artif_vm_iso_type.parent_type)
|
||||
self.assertEqual({'tosca.artifacts.Deployment':
|
||||
{'derived_from': 'tosca.artifacts.Root',
|
||||
'description':
|
||||
'TOSCA base type for deployment artifacts'},
|
||||
'tosca.artifacts.Deployment.Image':
|
||||
{'derived_from': 'tosca.artifacts.Deployment'}},
|
||||
artif_vm_iso_type.parent_artifacts)
|
||||
self.assertEqual(sorted(['tosca.artifacts.Deployment.Image',
|
||||
'Virtual Machine (VM) image in '
|
||||
'ISO disk format',
|
||||
'application/octet-stream', ['iso']],
|
||||
key=lambda x: str(x)),
|
||||
sorted([artif_vm_iso_type.
|
||||
get_artifact(name) for name in
|
||||
artif_vm_iso_type.defs],
|
||||
key=lambda x: str(x)))
|
||||
|
||||
self.assertEqual('tosca.artifacts.Deployment.Image',
|
||||
artif_vm_qcow2_type.parent_type)
|
||||
self.assertEqual({'tosca.artifacts.Deployment':
|
||||
{'derived_from': 'tosca.artifacts.Root',
|
||||
'description':
|
||||
'TOSCA base type for deployment artifacts'},
|
||||
'tosca.artifacts.Deployment.Image':
|
||||
{'derived_from': 'tosca.artifacts.Deployment'}},
|
||||
artif_vm_qcow2_type.parent_artifacts)
|
||||
self.assertEqual(sorted(['tosca.artifacts.Deployment.Image',
|
||||
'Virtual Machine (VM) image in QCOW v2 '
|
||||
'standard disk format',
|
||||
'application/octet-stream', ['qcow2']],
|
||||
key=lambda x: str(x)),
|
||||
sorted([artif_vm_qcow2_type.
|
||||
get_artifact(name) for name in
|
||||
artif_vm_qcow2_type.defs],
|
||||
key=lambda x: str(x)))
|
@ -1,384 +0,0 @@
|
||||
# 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
|
||||
import toscaparser.elements.interfaces as ifaces
|
||||
from toscaparser.elements.nodetype import NodeType
|
||||
from toscaparser.functions import GetInput
|
||||
from toscaparser.functions import GetProperty
|
||||
from toscaparser.nodetemplate import NodeTemplate
|
||||
from toscaparser.tests.base import TestCase
|
||||
from toscaparser.tosca_template import ToscaTemplate
|
||||
import toscaparser.utils.yamlparser
|
||||
|
||||
|
||||
class ToscaTemplateTest(TestCase):
|
||||
|
||||
'''TOSCA template.'''
|
||||
tosca_tpl = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/tosca_single_instance_wordpress.yaml")
|
||||
tosca = ToscaTemplate(tosca_tpl)
|
||||
|
||||
tosca_elk_tpl = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/tosca_elk.yaml")
|
||||
|
||||
def test_version(self):
|
||||
self.assertEqual(self.tosca.version, "tosca_simple_yaml_1_0")
|
||||
|
||||
def test_description(self):
|
||||
expected_description = "TOSCA simple profile with wordpress, " \
|
||||
"web server and mysql on the same server."
|
||||
self.assertEqual(self.tosca.description, expected_description)
|
||||
|
||||
def test_inputs(self):
|
||||
self.assertEqual(
|
||||
['cpus', 'db_name', 'db_port',
|
||||
'db_pwd', 'db_root_pwd', 'db_user'],
|
||||
sorted([input.name for input in self.tosca.inputs]))
|
||||
|
||||
input_name = "db_port"
|
||||
expected_description = "Port for the MySQL database."
|
||||
for input in self.tosca.inputs:
|
||||
if input.name == input_name:
|
||||
self.assertEqual(input.description, expected_description)
|
||||
|
||||
def test_node_tpls(self):
|
||||
'''Test nodetemplate names.'''
|
||||
self.assertEqual(
|
||||
['mysql_database', 'mysql_dbms', 'server',
|
||||
'webserver', 'wordpress'],
|
||||
sorted([tpl.name for tpl in self.tosca.nodetemplates]))
|
||||
|
||||
tpl_name = "mysql_database"
|
||||
expected_type = "tosca.nodes.Database"
|
||||
expected_properties = ['name', 'password', 'user']
|
||||
expected_capabilities = ['database_endpoint']
|
||||
expected_requirements = [{'host': {'node': 'mysql_dbms'}}]
|
||||
''' TODO: needs enhancement in tosca_elk.yaml..
|
||||
expected_relationshp = ['tosca.relationships.HostedOn']
|
||||
expected_host = ['mysql_dbms']
|
||||
'''
|
||||
expected_interface = [ifaces.LIFECYCLE_SHORTNAME]
|
||||
|
||||
for tpl in self.tosca.nodetemplates:
|
||||
if tpl_name == tpl.name:
|
||||
'''Test node type.'''
|
||||
self.assertEqual(tpl.type, expected_type)
|
||||
|
||||
'''Test properties.'''
|
||||
self.assertEqual(
|
||||
expected_properties,
|
||||
sorted(tpl.get_properties().keys()))
|
||||
|
||||
'''Test capabilities.'''
|
||||
self.assertEqual(
|
||||
expected_capabilities,
|
||||
sorted(tpl.get_capabilities().keys()))
|
||||
|
||||
'''Test requirements.'''
|
||||
self.assertEqual(
|
||||
expected_requirements, tpl.requirements)
|
||||
|
||||
'''Test relationship.'''
|
||||
''' needs enhancements in tosca_elk.yaml
|
||||
self.assertEqual(
|
||||
expected_relationshp,
|
||||
[x.type for x in tpl.relationships.keys()])
|
||||
self.assertEqual(
|
||||
expected_host,
|
||||
[y.name for y in tpl.relationships.values()])
|
||||
'''
|
||||
'''Test interfaces.'''
|
||||
self.assertEqual(
|
||||
expected_interface,
|
||||
[x.type for x in tpl.interfaces])
|
||||
|
||||
if tpl.name == 'server':
|
||||
'''Test property value'''
|
||||
props = tpl.get_properties()
|
||||
if props and 'mem_size' in props.keys():
|
||||
self.assertEqual(props['mem_size'].value, '4096 MB')
|
||||
'''Test capability'''
|
||||
caps = tpl.get_capabilities()
|
||||
self.assertIn('os', caps.keys())
|
||||
os_props_objs = None
|
||||
os_props = None
|
||||
os_type_prop = None
|
||||
if caps and 'os' in caps.keys():
|
||||
capability = caps['os']
|
||||
os_props_objs = capability.get_properties_objects()
|
||||
os_props = capability.get_properties()
|
||||
os_type_prop = capability.get_property_value('type')
|
||||
break
|
||||
self.assertEqual(
|
||||
['Linux'],
|
||||
[p.value for p in os_props_objs if p.name == 'type'])
|
||||
self.assertEqual(
|
||||
'Linux',
|
||||
os_props['type'].value if 'type' in os_props else '')
|
||||
self.assertEqual('Linux', os_props['type'].value)
|
||||
self.assertEqual('Linux', os_type_prop)
|
||||
|
||||
def test_outputs(self):
|
||||
self.assertEqual(
|
||||
['website_url'],
|
||||
sorted([output.name for output in self.tosca.outputs]))
|
||||
|
||||
def test_interfaces(self):
|
||||
wordpress_node = [
|
||||
node for node in self.tosca.nodetemplates
|
||||
if node.name == 'wordpress'][0]
|
||||
interfaces = wordpress_node.interfaces
|
||||
self.assertEqual(2, len(interfaces))
|
||||
for interface in interfaces:
|
||||
if interface.name == 'create':
|
||||
self.assertEqual(ifaces.LIFECYCLE_SHORTNAME,
|
||||
interface.type)
|
||||
self.assertEqual('wordpress/wordpress_install.sh',
|
||||
interface.implementation)
|
||||
self.assertIsNone(interface.inputs)
|
||||
elif interface.name == 'configure':
|
||||
self.assertEqual(ifaces.LIFECYCLE_SHORTNAME,
|
||||
interface.type)
|
||||
self.assertEqual('wordpress/wordpress_configure.sh',
|
||||
interface.implementation)
|
||||
self.assertEqual(3, len(interface.inputs))
|
||||
TestCase.skip(self, 'bug #1440247')
|
||||
wp_db_port = interface.inputs['wp_db_port']
|
||||
self.assertTrue(isinstance(wp_db_port, GetProperty))
|
||||
self.assertEqual('get_property', wp_db_port.name)
|
||||
self.assertEqual(['SELF',
|
||||
'database_endpoint',
|
||||
'port'],
|
||||
wp_db_port.args)
|
||||
result = wp_db_port.result()
|
||||
self.assertTrue(isinstance(result, GetInput))
|
||||
else:
|
||||
raise AssertionError(
|
||||
'Unexpected interface: {0}'.format(interface.name))
|
||||
|
||||
def test_normative_type_by_short_name(self):
|
||||
# test template with a short name Compute
|
||||
template = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/test_tosca_normative_type_by_shortname.yaml")
|
||||
|
||||
tosca_tpl = ToscaTemplate(template)
|
||||
expected_type = "tosca.nodes.Compute"
|
||||
for tpl in tosca_tpl.nodetemplates:
|
||||
self.assertEqual(tpl.type, expected_type)
|
||||
for tpl in tosca_tpl.nodetemplates:
|
||||
compute_type = NodeType(tpl.type)
|
||||
self.assertEqual(
|
||||
sorted(['tosca.capabilities.Container',
|
||||
'tosca.capabilities.OperatingSystem',
|
||||
'tosca.capabilities.network.Bindable',
|
||||
'tosca.capabilities.Scalable']),
|
||||
sorted([c.type
|
||||
for c in compute_type.get_capabilities_objects()]))
|
||||
|
||||
def test_template_with_no_inputs(self):
|
||||
tosca_tpl = self._load_template('test_no_inputs_in_template.yaml')
|
||||
self.assertEqual(0, len(tosca_tpl.inputs))
|
||||
|
||||
def test_template_with_no_outputs(self):
|
||||
tosca_tpl = self._load_template('test_no_outputs_in_template.yaml')
|
||||
self.assertEqual(0, len(tosca_tpl.outputs))
|
||||
|
||||
def test_relationship_interface(self):
|
||||
template = ToscaTemplate(self.tosca_elk_tpl)
|
||||
for node_tpl in template.nodetemplates:
|
||||
if node_tpl.name == 'logstash':
|
||||
config_interface = 'Configure'
|
||||
artifact = 'logstash/configure_elasticsearch.py'
|
||||
relation = node_tpl.relationships
|
||||
for key in relation.keys():
|
||||
rel_tpl = relation.get(key).get_relationship_template()
|
||||
if rel_tpl:
|
||||
interfaces = rel_tpl[0].interfaces
|
||||
for interface in interfaces:
|
||||
self.assertEqual(config_interface,
|
||||
interface.type)
|
||||
self.assertEqual('pre_configure_source',
|
||||
interface.name)
|
||||
self.assertEqual(artifact,
|
||||
interface.implementation)
|
||||
|
||||
def test_template_macro(self):
|
||||
template = ToscaTemplate(self.tosca_elk_tpl)
|
||||
for node_tpl in template.nodetemplates:
|
||||
if node_tpl.name == 'mongo_server':
|
||||
self.assertEqual(
|
||||
['disk_size', 'mem_size', 'num_cpus'],
|
||||
sorted(node_tpl.get_capability('host').
|
||||
get_properties().keys()))
|
||||
|
||||
def test_template_requirements(self):
|
||||
"""Test different formats of requirements
|
||||
|
||||
The requirements can be defined in few different ways,
|
||||
1. Requirement expressed as a capability with an implicit relationship.
|
||||
2. Requirement expressed with explicit relationship.
|
||||
3. Requirement expressed with a relationship template.
|
||||
4. Requirement expressed via TOSCA types to provision a node
|
||||
with explicit relationship.
|
||||
5. Requirement expressed via TOSCA types with a filter.
|
||||
"""
|
||||
tosca_tpl = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/test_requirements.yaml")
|
||||
tosca = ToscaTemplate(tosca_tpl)
|
||||
for node_tpl in tosca.nodetemplates:
|
||||
if node_tpl.name == 'my_app':
|
||||
expected_relationship = [
|
||||
('tosca.relationships.ConnectsTo', 'mysql_database'),
|
||||
('tosca.relationships.HostedOn', 'my_webserver')]
|
||||
actual_relationship = sorted([
|
||||
(relation.type, node.name) for
|
||||
relation, node in node_tpl.relationships.items()])
|
||||
self.assertEqual(expected_relationship, actual_relationship)
|
||||
if node_tpl.name == 'mysql_database':
|
||||
self.assertEqual(
|
||||
[('tosca.relationships.HostedOn', 'my_dbms')],
|
||||
[(relation.type, node.name) for
|
||||
relation,
|
||||
node in node_tpl.relationships.items()])
|
||||
if node_tpl.name == 'my_server':
|
||||
self.assertEqual(
|
||||
[('tosca.relationships.AttachesTo', 'my_storage')],
|
||||
[(relation.type, node.name) for
|
||||
relation,
|
||||
node in node_tpl.relationships.items()])
|
||||
|
||||
def test_template_requirements_not_implemented(self):
|
||||
# TODO(spzala): replace this test with new one once TOSCA types look up
|
||||
# support is implemented.
|
||||
"""Requirements that yet need to be implemented
|
||||
|
||||
The following requirement formats are not yet implemented,
|
||||
due to look up dependency:
|
||||
1. Requirement expressed via TOSCA types to provision a node
|
||||
with explicit relationship.
|
||||
2. Requirement expressed via TOSCA types with a filter.
|
||||
"""
|
||||
tpl_snippet_1 = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
description: Requires a particular node type and relationship.
|
||||
To be full-filled via lookup into node repository.
|
||||
requirements:
|
||||
- req1:
|
||||
node: tosca.nodes.DBMS
|
||||
relationship: tosca.relationships.HostedOn
|
||||
'''
|
||||
|
||||
tpl_snippet_2 = '''
|
||||
node_templates:
|
||||
my_webserver:
|
||||
type: tosca.nodes.WebServer
|
||||
description: Requires a particular node type with a filter.
|
||||
To be full-filled via lookup into node repository.
|
||||
requirements:
|
||||
- req1:
|
||||
node: tosca.nodes.Compute
|
||||
target_filter:
|
||||
properties:
|
||||
num_cpus: { in_range: [ 1, 4 ] }
|
||||
mem_size: { greater_or_equal: 2 }
|
||||
capabilities:
|
||||
- tosca.capabilities.OS:
|
||||
properties:
|
||||
architecture: x86_64
|
||||
type: linux
|
||||
'''
|
||||
|
||||
tpl_snippet_3 = '''
|
||||
node_templates:
|
||||
my_webserver2:
|
||||
type: tosca.nodes.WebServer
|
||||
description: Requires a node type with a particular capability.
|
||||
To be full-filled via lookup into node repository.
|
||||
requirements:
|
||||
- req1:
|
||||
node: tosca.nodes.Compute
|
||||
relationship: tosca.relationships.HostedOn
|
||||
capability: tosca.capabilities.Container
|
||||
'''
|
||||
self._requirements_not_implemented(tpl_snippet_1, 'mysql_database')
|
||||
self._requirements_not_implemented(tpl_snippet_2, 'my_webserver')
|
||||
self._requirements_not_implemented(tpl_snippet_3, 'my_webserver2')
|
||||
|
||||
def _requirements_not_implemented(self, tpl_snippet, tpl_name):
|
||||
nodetemplates = (toscaparser.utils.yamlparser.
|
||||
simple_parse(tpl_snippet))['node_templates']
|
||||
self.assertRaises(
|
||||
NotImplementedError,
|
||||
lambda: NodeTemplate(tpl_name, nodetemplates).relationships)
|
||||
|
||||
def test_custom_capability_type_definition(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
test_app:
|
||||
type: tosca.nodes.WebApplication.TestApp
|
||||
capabilities:
|
||||
test_cap:
|
||||
properties:
|
||||
test: 1
|
||||
'''
|
||||
# custom definition with capability type definition
|
||||
custom_def = '''
|
||||
tosca.nodes.WebApplication.TestApp:
|
||||
derived_from: tosca.nodes.WebApplication
|
||||
capabilities:
|
||||
test_cap:
|
||||
type: tosca.capabilities.TestCapability
|
||||
tosca.capabilities.TestCapability:
|
||||
derived_from: tosca.capabilities.Root
|
||||
properties:
|
||||
test:
|
||||
type: integer
|
||||
required: no
|
||||
'''
|
||||
expected_capabilities = ['test_cap']
|
||||
nodetemplates = (toscaparser.utils.yamlparser.
|
||||
simple_parse(tpl_snippet))['node_templates']
|
||||
custom_def = (toscaparser.utils.yamlparser.
|
||||
simple_parse(custom_def))
|
||||
name = list(nodetemplates.keys())[0]
|
||||
tpl = NodeTemplate(name, nodetemplates, custom_def)
|
||||
self.assertEqual(
|
||||
expected_capabilities,
|
||||
sorted(tpl.get_capabilities().keys()))
|
||||
|
||||
# custom definition without capability type definition
|
||||
custom_def = '''
|
||||
tosca.nodes.WebApplication.TestApp:
|
||||
derived_from: tosca.nodes.WebApplication
|
||||
capabilities:
|
||||
test_cap:
|
||||
type: tosca.capabilities.TestCapability
|
||||
'''
|
||||
custom_def = (toscaparser.utils.yamlparser.
|
||||
simple_parse(custom_def))
|
||||
tpl = NodeTemplate(name, nodetemplates, custom_def)
|
||||
err = self.assertRaises(
|
||||
exception.InvalidTypeError,
|
||||
lambda: NodeTemplate(name, nodetemplates,
|
||||
custom_def).get_capabilities_objects())
|
||||
self.assertEqual('Type "tosca.capabilities.TestCapability" is not '
|
||||
'a valid type.', six.text_type(err))
|
@ -1,794 +0,0 @@
|
||||
# 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.nodetemplate import NodeTemplate
|
||||
from toscaparser.parameters import Input
|
||||
from toscaparser.parameters import Output
|
||||
from toscaparser.relationship_template import RelationshipTemplate
|
||||
from toscaparser.tests.base import TestCase
|
||||
from toscaparser.tosca_template import ToscaTemplate
|
||||
import toscaparser.utils.yamlparser
|
||||
|
||||
|
||||
class ToscaTemplateValidationTest(TestCase):
|
||||
|
||||
def test_well_defined_template(self):
|
||||
tpl_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/tosca_single_instance_wordpress.yaml")
|
||||
self.assertIsNotNone(ToscaTemplate(tpl_path))
|
||||
|
||||
def test_first_level_sections(self):
|
||||
tpl_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/test_tosca_top_level_error1.yaml")
|
||||
err = self.assertRaises(exception.MissingRequiredFieldError,
|
||||
ToscaTemplate, tpl_path)
|
||||
self.assertEqual('Template is missing required field: '
|
||||
'"tosca_definitions_version".', err.__str__())
|
||||
|
||||
tpl_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/test_tosca_top_level_error2.yaml")
|
||||
err = self.assertRaises(exception.UnknownFieldError,
|
||||
ToscaTemplate, tpl_path)
|
||||
self.assertEqual('Template contain(s) unknown field: '
|
||||
'"node_template", refer to the definition '
|
||||
'to verify valid values.', err.__str__())
|
||||
|
||||
def test_inputs(self):
|
||||
tpl_snippet = '''
|
||||
inputs:
|
||||
cpus:
|
||||
type: integer
|
||||
description: Number of CPUs for the server.
|
||||
constraint:
|
||||
- valid_values: [ 1, 2, 4, 8 ]
|
||||
'''
|
||||
inputs = (toscaparser.utils.yamlparser.
|
||||
simple_parse(tpl_snippet)['inputs'])
|
||||
name, attrs = list(inputs.items())[0]
|
||||
input = Input(name, attrs)
|
||||
try:
|
||||
input.validate()
|
||||
except Exception as err:
|
||||
self.assertTrue(isinstance(err, exception.UnknownFieldError))
|
||||
self.assertEqual('Input cpus contain(s) unknown field: '
|
||||
'"constraint", refer to the definition to '
|
||||
'verify valid values.', err.__str__())
|
||||
|
||||
def test_outputs(self):
|
||||
tpl_snippet = '''
|
||||
outputs:
|
||||
server_address:
|
||||
description: IP address of server instance.
|
||||
values: { get_property: [server, private_address] }
|
||||
'''
|
||||
outputs = (toscaparser.utils.yamlparser.
|
||||
simple_parse(tpl_snippet)['outputs'])
|
||||
name, attrs = list(outputs.items())[0]
|
||||
output = Output(name, attrs)
|
||||
try:
|
||||
output.validate()
|
||||
except Exception as err:
|
||||
self.assertTrue(
|
||||
isinstance(err, exception.MissingRequiredFieldError))
|
||||
self.assertEqual('Output server_address is missing required '
|
||||
'field: "value".', err.__str__())
|
||||
|
||||
tpl_snippet = '''
|
||||
outputs:
|
||||
server_address:
|
||||
descriptions: IP address of server instance.
|
||||
value: { get_property: [server, private_address] }
|
||||
'''
|
||||
outputs = (toscaparser.utils.yamlparser.
|
||||
simple_parse(tpl_snippet)['outputs'])
|
||||
name, attrs = list(outputs.items())[0]
|
||||
output = Output(name, attrs)
|
||||
try:
|
||||
output.validate()
|
||||
except Exception as err:
|
||||
self.assertTrue(isinstance(err, exception.UnknownFieldError))
|
||||
self.assertEqual('Output server_address contain(s) unknown '
|
||||
'field: "descriptions", refer to the definition '
|
||||
'to verify valid values.',
|
||||
err.__str__())
|
||||
|
||||
def _custom_types(self):
|
||||
custom_types = {}
|
||||
def_file = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/custom_types/wordpress.yaml")
|
||||
custom_type = toscaparser.utils.yamlparser.load_yaml(def_file)
|
||||
node_types = custom_type['node_types']
|
||||
for name in node_types:
|
||||
defintion = node_types[name]
|
||||
custom_types[name] = defintion
|
||||
return custom_types
|
||||
|
||||
def _single_node_template_content_test(self, tpl_snippet, expectederror,
|
||||
expectedmessage):
|
||||
nodetemplates = (toscaparser.utils.yamlparser.
|
||||
simple_ordered_parse(tpl_snippet))['node_templates']
|
||||
name = list(nodetemplates.keys())[0]
|
||||
try:
|
||||
nodetemplate = NodeTemplate(name, nodetemplates,
|
||||
self._custom_types())
|
||||
nodetemplate.validate()
|
||||
nodetemplate.requirements
|
||||
nodetemplate.get_capabilities_objects()
|
||||
nodetemplate.get_properties_objects()
|
||||
nodetemplate.interfaces
|
||||
|
||||
except Exception as err:
|
||||
self.assertTrue(isinstance(err, expectederror))
|
||||
self.assertEqual(expectedmessage, err.__str__())
|
||||
|
||||
def test_node_templates(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
server:
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
disk_size: 10
|
||||
num_cpus: 4
|
||||
mem_size: 4096
|
||||
os:
|
||||
properties:
|
||||
architecture: x86_64
|
||||
type: Linux
|
||||
distribution: Fedora
|
||||
version: 18.0
|
||||
'''
|
||||
expectedmessage = ('Template server is missing '
|
||||
'required field: "type".')
|
||||
self._single_node_template_content_test(
|
||||
tpl_snippet,
|
||||
exception.MissingRequiredFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_with_wrong_properties_keyname(self):
|
||||
"""Node template keyname 'properties' given as 'propertiessss'."""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_dbms:
|
||||
type: tosca.nodes.DBMS
|
||||
propertiessss:
|
||||
root_password: aaa
|
||||
port: 3376
|
||||
'''
|
||||
expectedmessage = ('Node template mysql_dbms '
|
||||
'contain(s) unknown field: "propertiessss", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_with_wrong_requirements_keyname(self):
|
||||
"""Node template keyname 'requirements' given as 'requirement'."""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_dbms:
|
||||
type: tosca.nodes.DBMS
|
||||
properties:
|
||||
root_password: aaa
|
||||
port: 3376
|
||||
requirement:
|
||||
- host: server
|
||||
'''
|
||||
expectedmessage = ('Node template mysql_dbms '
|
||||
'contain(s) unknown field: "requirement", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_with_wrong_interfaces_keyname(self):
|
||||
"""Node template keyname 'interfaces' given as 'interfac'."""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_dbms:
|
||||
type: tosca.nodes.DBMS
|
||||
properties:
|
||||
root_password: aaa
|
||||
port: 3376
|
||||
requirements:
|
||||
- host: server
|
||||
interfac:
|
||||
Standard:
|
||||
configure: mysql_database_configure.sh
|
||||
'''
|
||||
expectedmessage = ('Node template mysql_dbms '
|
||||
'contain(s) unknown field: "interfac", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_with_wrong_capabilities_keyname(self):
|
||||
"""Node template keyname 'capabilities' given as 'capabilitiis'."""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
properties:
|
||||
db_name: { get_input: db_name }
|
||||
db_user: { get_input: db_user }
|
||||
db_password: { get_input: db_pwd }
|
||||
capabilitiis:
|
||||
database_endpoint:
|
||||
properties:
|
||||
port: { get_input: db_port }
|
||||
'''
|
||||
expectedmessage = ('Node template mysql_database '
|
||||
'contain(s) unknown field: "capabilitiis", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_with_wrong_artifacts_keyname(self):
|
||||
"""Node template keyname 'artifacts' given as 'artifactsss'."""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
artifactsss:
|
||||
db_content:
|
||||
implementation: files/my_db_content.txt
|
||||
type: tosca.artifacts.File
|
||||
'''
|
||||
expectedmessage = ('Node template mysql_database '
|
||||
'contain(s) unknown field: "artifactsss", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_with_multiple_wrong_keynames(self):
|
||||
"""Node templates given with multiple wrong keynames."""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_dbms:
|
||||
type: tosca.nodes.DBMS
|
||||
propertieees:
|
||||
root_password: aaa
|
||||
port: 3376
|
||||
requirements:
|
||||
- host: server
|
||||
interfacs:
|
||||
Standard:
|
||||
configure: mysql_database_configure.sh
|
||||
'''
|
||||
expectedmessage = ('Node template mysql_dbms '
|
||||
'contain(s) unknown field: "propertieees", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
properties:
|
||||
name: { get_input: db_name }
|
||||
user: { get_input: db_user }
|
||||
password: { get_input: db_pwd }
|
||||
capabilitiiiies:
|
||||
database_endpoint:
|
||||
properties:
|
||||
port: { get_input: db_port }
|
||||
requirementsss:
|
||||
- host:
|
||||
node: mysql_dbms
|
||||
interfac:
|
||||
Standard:
|
||||
configure: mysql_database_configure.sh
|
||||
|
||||
'''
|
||||
expectedmessage = ('Node template mysql_database '
|
||||
'contain(s) unknown field: "capabilitiiiies", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_type(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Databases
|
||||
properties:
|
||||
db_name: { get_input: db_name }
|
||||
db_user: { get_input: db_user }
|
||||
db_password: { get_input: db_pwd }
|
||||
capabilities:
|
||||
database_endpoint:
|
||||
properties:
|
||||
port: { get_input: db_port }
|
||||
requirements:
|
||||
- host: mysql_dbms
|
||||
interfaces:
|
||||
Standard:
|
||||
configure: mysql_database_configure.sh
|
||||
'''
|
||||
expectedmessage = ('Type "tosca.nodes.Databases" is not '
|
||||
'a valid type.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.InvalidTypeError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_requirements(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
webserver:
|
||||
type: tosca.nodes.WebServer
|
||||
requirements:
|
||||
host: server
|
||||
interfaces:
|
||||
Standard:
|
||||
create: webserver_install.sh
|
||||
start: d.sh
|
||||
'''
|
||||
expectedmessage = ('Requirements of template webserver '
|
||||
'must be of type: "list".')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.TypeMismatchError,
|
||||
expectedmessage)
|
||||
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
properties:
|
||||
db_name: { get_input: db_name }
|
||||
db_user: { get_input: db_user }
|
||||
db_password: { get_input: db_pwd }
|
||||
capabilities:
|
||||
database_endpoint:
|
||||
properties:
|
||||
port: { get_input: db_port }
|
||||
requirements:
|
||||
- host: mysql_dbms
|
||||
- database_endpoint: mysql_database
|
||||
interfaces:
|
||||
Standard:
|
||||
configure: mysql_database_configure.sh
|
||||
'''
|
||||
expectedmessage = ('Requirements of template mysql_database '
|
||||
'contain(s) unknown field: "database_endpoint", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_requirements_with_wrong_node_keyname(self):
|
||||
"""Node template requirements keyname 'node' given as 'nodes'."""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
requirements:
|
||||
- host:
|
||||
nodes: mysql_dbms
|
||||
|
||||
'''
|
||||
expectedmessage = ('Requirements of template mysql_database '
|
||||
'contain(s) unknown field: "nodes", refer to the '
|
||||
'definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_requirements_with_wrong_capability_keyname(self):
|
||||
"""Incorrect node template requirements keyname
|
||||
|
||||
Node template requirements keyname 'capability' given as
|
||||
'capabilityy'.
|
||||
"""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
requirements:
|
||||
- host:
|
||||
node: mysql_dbms
|
||||
- log_endpoint:
|
||||
node: logstash
|
||||
capabilityy: log_endpoint
|
||||
relationship:
|
||||
type: tosca.relationships.ConnectsTo
|
||||
|
||||
'''
|
||||
expectedmessage = ('Requirements of template mysql_database '
|
||||
'contain(s) unknown field: "capabilityy", refer to '
|
||||
'the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_requirements_with_wrong_relationship_keyname(self):
|
||||
"""Incorrect node template requirements keyname
|
||||
|
||||
Node template requirements keyname 'relationship' given as
|
||||
'relationshipppp'.
|
||||
"""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
requirements:
|
||||
- host:
|
||||
node: mysql_dbms
|
||||
- log_endpoint:
|
||||
node: logstash
|
||||
capability: log_endpoint
|
||||
relationshipppp:
|
||||
type: tosca.relationships.ConnectsTo
|
||||
|
||||
'''
|
||||
expectedmessage = ('Requirements of template mysql_database '
|
||||
'contain(s) unknown field: "relationshipppp", refer'
|
||||
' to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_requirements_with_multiple_wrong_keynames(self):
|
||||
"""Node templates given with multiple wrong requirements keynames."""
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
requirements:
|
||||
- host:
|
||||
node: mysql_dbms
|
||||
- log_endpoint:
|
||||
nod: logstash
|
||||
capabilit: log_endpoint
|
||||
relationshipppp:
|
||||
type: tosca.relationships.ConnectsTo
|
||||
|
||||
'''
|
||||
expectedmessage = ('Requirements of template mysql_database '
|
||||
'contain(s) unknown field: "nod", refer'
|
||||
' to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
requirements:
|
||||
- host:
|
||||
node: mysql_dbms
|
||||
- log_endpoint:
|
||||
node: logstash
|
||||
capabilit: log_endpoint
|
||||
relationshipppp:
|
||||
type: tosca.relationships.ConnectsTo
|
||||
|
||||
'''
|
||||
expectedmessage = ('Requirements of template mysql_database '
|
||||
'contain(s) unknown field: "capabilit", refer'
|
||||
' to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_capabilities(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
mysql_database:
|
||||
type: tosca.nodes.Database
|
||||
properties:
|
||||
db_name: { get_input: db_name }
|
||||
db_user: { get_input: db_user }
|
||||
db_password: { get_input: db_pwd }
|
||||
capabilities:
|
||||
http_endpoint:
|
||||
properties:
|
||||
port: { get_input: db_port }
|
||||
requirements:
|
||||
- host: mysql_dbms
|
||||
interfaces:
|
||||
Standard:
|
||||
configure: mysql_database_configure.sh
|
||||
'''
|
||||
expectedmessage = ('Capabilities of template mysql_database '
|
||||
'contain(s) unknown field: "http_endpoint", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_properties(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
properties:
|
||||
os_image: F18_x86_64
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
disk_size: 10 GB
|
||||
num_cpus: { get_input: cpus }
|
||||
mem_size: 4096 MB
|
||||
os:
|
||||
properties:
|
||||
architecture: x86_64
|
||||
type: Linux
|
||||
distribution: Fedora
|
||||
version: 18.0
|
||||
'''
|
||||
expectedmessage = ('Properties of template server contain(s) '
|
||||
'unknown field: "os_image", refer to the '
|
||||
'definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_interfaces(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
wordpress:
|
||||
type: tosca.nodes.WebApplication.WordPress
|
||||
requirements:
|
||||
- host: webserver
|
||||
- database_endpoint: mysql_database
|
||||
interfaces:
|
||||
Standards:
|
||||
create: wordpress_install.sh
|
||||
configure:
|
||||
implementation: wordpress_configure.sh
|
||||
inputs:
|
||||
wp_db_name: { get_property: [ mysql_database, db_name ] }
|
||||
wp_db_user: { get_property: [ mysql_database, db_user ] }
|
||||
wp_db_password: { get_property: [ mysql_database, \
|
||||
db_password ] }
|
||||
wp_db_port: { get_property: [ SELF, \
|
||||
database_endpoint, port ] }
|
||||
'''
|
||||
expectedmessage = ('Interfaces of template wordpress '
|
||||
'contain(s) unknown field: '
|
||||
'"Standards", '
|
||||
'refer to the definition to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
wordpress:
|
||||
type: tosca.nodes.WebApplication.WordPress
|
||||
requirements:
|
||||
- host: webserver
|
||||
- database_endpoint: mysql_database
|
||||
interfaces:
|
||||
Standard:
|
||||
create: wordpress_install.sh
|
||||
config:
|
||||
implementation: wordpress_configure.sh
|
||||
inputs:
|
||||
wp_db_name: { get_property: [ mysql_database, db_name ] }
|
||||
wp_db_user: { get_property: [ mysql_database, db_user ] }
|
||||
wp_db_password: { get_property: [ mysql_database, \
|
||||
db_password ] }
|
||||
wp_db_port: { get_property: [ SELF, \
|
||||
database_endpoint, port ] }
|
||||
'''
|
||||
expectedmessage = ('Interfaces of template wordpress contain(s) '
|
||||
'unknown field: "config", refer to the definition'
|
||||
' to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
wordpress:
|
||||
type: tosca.nodes.WebApplication.WordPress
|
||||
requirements:
|
||||
- host: webserver
|
||||
- database_endpoint: mysql_database
|
||||
interfaces:
|
||||
Standard:
|
||||
create: wordpress_install.sh
|
||||
configure:
|
||||
implementation: wordpress_configure.sh
|
||||
inputs:
|
||||
wp_db_name: { get_property: [ mysql_database, db_name ] }
|
||||
wp_db_user: { get_property: [ mysql_database, db_user ] }
|
||||
wp_db_password: { get_property: [ mysql_database, \
|
||||
db_password ] }
|
||||
wp_db_port: { get_ref_property: [ database_endpoint, \
|
||||
database_endpoint, port ] }
|
||||
'''
|
||||
expectedmessage = ('Interfaces of template wordpress contain(s) '
|
||||
'unknown field: "inputs", refer to the definition'
|
||||
' to verify valid values.')
|
||||
self._single_node_template_content_test(tpl_snippet,
|
||||
exception.UnknownFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_relationship_template_properties(self):
|
||||
tpl_snippet = '''
|
||||
relationship_templates:
|
||||
storage_attachto:
|
||||
type: AttachesTo
|
||||
properties:
|
||||
device: test_device
|
||||
'''
|
||||
expectedmessage = ('Properties of template '
|
||||
'storage_attachto is missing required field: '
|
||||
'"[\'location\']".')
|
||||
self._single_rel_template_content_test(
|
||||
tpl_snippet,
|
||||
exception.MissingRequiredFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def _single_rel_template_content_test(self, tpl_snippet, expectederror,
|
||||
expectedmessage):
|
||||
rel_template = (toscaparser.utils.yamlparser.
|
||||
simple_parse(tpl_snippet))['relationship_templates']
|
||||
name = list(rel_template.keys())[0]
|
||||
rel_template = RelationshipTemplate(rel_template[name], name)
|
||||
err = self.assertRaises(expectederror, rel_template.validate)
|
||||
self.assertEqual(expectedmessage, six.text_type(err))
|
||||
|
||||
def test_invalid_template_version(self):
|
||||
tosca_tpl = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data/test_invalid_template_version.yaml")
|
||||
err = self.assertRaises(exception.InvalidTemplateVersion,
|
||||
ToscaTemplate, tosca_tpl)
|
||||
valid_versions = ', '.join(ToscaTemplate.VALID_TEMPLATE_VERSIONS)
|
||||
ex_err_msg = ('The template version "tosca_xyz" is invalid. '
|
||||
'The valid versions are: "%s"' % valid_versions)
|
||||
self.assertEqual(six.text_type(err), ex_err_msg)
|
||||
|
||||
def test_node_template_capabilities_properties(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
disk_size: 10 GB
|
||||
num_cpus: { get_input: cpus }
|
||||
mem_size: 4096 MB
|
||||
os:
|
||||
properties:
|
||||
architecture: x86_64
|
||||
type: Linux
|
||||
distribution: Fedora
|
||||
version: 18.0
|
||||
scalable:
|
||||
properties:
|
||||
min_instances: 1
|
||||
default_instances: 5
|
||||
'''
|
||||
expectedmessage = ('Properties of template server is missing '
|
||||
'required field: '
|
||||
'"[\'max_instances\']".')
|
||||
|
||||
self._single_node_template_content_test(
|
||||
tpl_snippet,
|
||||
exception.MissingRequiredFieldError,
|
||||
expectedmessage)
|
||||
|
||||
# validatating capability property values
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
server:
|
||||
type: tosca.nodes.WebServer
|
||||
capabilities:
|
||||
data_endpoint:
|
||||
properties:
|
||||
initiator: test
|
||||
'''
|
||||
expectedmessage = ('initiator: test is not an valid value '
|
||||
'"[source, target, peer]".')
|
||||
|
||||
self._single_node_template_content_test(
|
||||
tpl_snippet,
|
||||
exception.ValidationError,
|
||||
expectedmessage)
|
||||
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
server:
|
||||
type: tosca.nodes.Compute
|
||||
capabilities:
|
||||
host:
|
||||
properties:
|
||||
disk_size: 10 GB
|
||||
num_cpus: { get_input: cpus }
|
||||
mem_size: 4096 MB
|
||||
os:
|
||||
properties:
|
||||
architecture: x86_64
|
||||
type: Linux
|
||||
distribution: Fedora
|
||||
version: 18.0
|
||||
scalable:
|
||||
properties:
|
||||
min_instances: 1
|
||||
max_instances: 3
|
||||
default_instances: 5
|
||||
'''
|
||||
expectedmessage = ('Properties of template server : '
|
||||
'default_instances value is not between'
|
||||
' min_instances and max_instances')
|
||||
|
||||
self._single_node_template_content_test(
|
||||
tpl_snippet,
|
||||
exception.ValidationError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_objectstorage_without_required_property(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
server:
|
||||
type: tosca.nodes.ObjectStorage
|
||||
properties:
|
||||
maxsize: 1 GB
|
||||
'''
|
||||
expectedmessage = ('Properties of template server is missing '
|
||||
'required field: '
|
||||
'"[\'name\']".')
|
||||
|
||||
self._single_node_template_content_test(
|
||||
tpl_snippet,
|
||||
exception.MissingRequiredFieldError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_objectstorage_with_invalid_scalar_unit(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
server:
|
||||
type: tosca.nodes.ObjectStorage
|
||||
properties:
|
||||
name: test
|
||||
maxsize: -1
|
||||
'''
|
||||
expectedmessage = ('"-1" is not a valid scalar-unit')
|
||||
self._single_node_template_content_test(
|
||||
tpl_snippet,
|
||||
ValueError,
|
||||
expectedmessage)
|
||||
|
||||
def test_node_template_objectstorage_with_invalid_scalar_type(self):
|
||||
tpl_snippet = '''
|
||||
node_templates:
|
||||
server:
|
||||
type: tosca.nodes.ObjectStorage
|
||||
properties:
|
||||
name: test
|
||||
maxsize: 1 XB
|
||||
'''
|
||||
expectedmessage = ('"1 XB" is not a valid scalar-unit')
|
||||
self._single_node_template_content_test(
|
||||
tpl_snippet,
|
||||
ValueError,
|
||||
expectedmessage)
|
@ -1,139 +0,0 @@
|
||||
# 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.
|
||||
|
||||
from toscaparser.common.exception import (
|
||||
InvalidTOSCAVersionPropertyException)
|
||||
from toscaparser.tests.base import TestCase
|
||||
from toscaparser.utils.validateutils import TOSCAVersionProperty
|
||||
|
||||
|
||||
class TOSCAVersionPropertyTest(TestCase):
|
||||
|
||||
def test_tosca_version_property(self):
|
||||
version = '18.0.3.beta-1'
|
||||
expected_output = '18.0.3.beta-1'
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
version = 18
|
||||
expected_output = '18.0'
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
version = 18.0
|
||||
expected_output = '18.0'
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
version = '18.0.3'
|
||||
expected_output = '18.0.3'
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
version = 0
|
||||
expected_output = None
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
version = 00
|
||||
expected_output = None
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
version = 0.0
|
||||
expected_output = None
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
version = 00.00
|
||||
expected_output = None
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
version = '0.0.0'
|
||||
expected_output = None
|
||||
output = TOSCAVersionProperty(version).get_version()
|
||||
self.assertEqual(output, expected_output)
|
||||
|
||||
def test_tosca_version_property_invalid_major_version(self):
|
||||
|
||||
version = 'x'
|
||||
exp_msg = ('Value of TOSCA version property "x" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
||||
|
||||
def test_tosca_version_property_invalid_minor_version(self):
|
||||
|
||||
version = '18.x'
|
||||
exp_msg = ('Value of TOSCA version property "18.x" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
||||
|
||||
version = '18.x.y'
|
||||
exp_msg = ('Value of TOSCA version property "18.x.y" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
||||
|
||||
version = '18-2'
|
||||
exp_msg = ('Value of TOSCA version property "18-2" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
||||
|
||||
def test_tosca_version_property_invalid_fix_version(self):
|
||||
|
||||
version = '18.0.a'
|
||||
exp_msg = ('Value of TOSCA version property "18.0.a" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
||||
|
||||
def test_tosca_version_property_invalid_qualifier(self):
|
||||
|
||||
version = '18.0.1-xyz'
|
||||
exp_msg = ('Value of TOSCA version property "18.0.1-xyz" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
||||
|
||||
version = '0.0.0.abc'
|
||||
exp_msg = ('Value of TOSCA version property "0.0.0.abc" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
||||
|
||||
def test_tosca_version_property_invalid_build_version(self):
|
||||
|
||||
version = '18.0.1.abc-x'
|
||||
exp_msg = ('Value of TOSCA version property '
|
||||
'"18.0.1.abc-x" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
||||
|
||||
version = '0.0.0.abc-x'
|
||||
exp_msg = ('Value of TOSCA version property "0.0.0.abc-x" is invalid.')
|
||||
try:
|
||||
TOSCAVersionProperty(version).get_version()
|
||||
except InvalidTOSCAVersionPropertyException as err:
|
||||
self.assertEqual(exp_msg, err.__str__())
|
Loading…
x
Reference in New Issue
Block a user