cafd53be11
Add data type for the attributes. This will help 1) validating the attibute against its type 2) resource-schema api will show the attribute type 3) template authors will know what type of value to expect making indexing and mapping easier 4) generating docs for the attribute type Implements: blueprint add-type-in-attributes-schema Change-Id: Ifc92c57ec1ddd2ab5f810587a1d33e762308dd8a
217 lines
8.7 KiB
Python
217 lines
8.7 KiB
Python
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import mock
|
|
import six
|
|
|
|
from heat.engine import attributes
|
|
from heat.engine import resources
|
|
from heat.engine import support
|
|
from heat.tests import common
|
|
|
|
|
|
class AttributeSchemaTest(common.HeatTestCase):
|
|
def test_schema_all(self):
|
|
d = {'description': 'A attribute'}
|
|
s = attributes.Schema('A attribute')
|
|
self.assertEqual(d, dict(s))
|
|
|
|
d = {'description': 'Another attribute',
|
|
'type': 'string'}
|
|
s = attributes.Schema('Another attribute',
|
|
type=attributes.Schema.STRING)
|
|
self.assertEqual(d, dict(s))
|
|
|
|
def test_all_resource_schemata(self):
|
|
for resource_type in resources.global_env().get_types():
|
|
for schema in six.itervalues(getattr(resource_type,
|
|
'attributes_schema',
|
|
{})):
|
|
attributes.Schema.from_attribute(schema)
|
|
|
|
def test_from_attribute_new_schema_format(self):
|
|
s = attributes.Schema('Test description.')
|
|
self.assertIs(s, attributes.Schema.from_attribute(s))
|
|
self.assertEqual('Test description.',
|
|
attributes.Schema.from_attribute(s).description)
|
|
|
|
s = attributes.Schema('Test description.',
|
|
type=attributes.Schema.MAP)
|
|
self.assertIs(s, attributes.Schema.from_attribute(s))
|
|
self.assertEqual(attributes.Schema.MAP,
|
|
attributes.Schema.from_attribute(s).type)
|
|
|
|
def test_schema_support_status(self):
|
|
schema = {
|
|
'foo_sup': attributes.Schema(
|
|
'Description1'
|
|
),
|
|
'bar_dep': attributes.Schema(
|
|
'Description2',
|
|
support_status=support.SupportStatus(
|
|
support.DEPRECATED,
|
|
'Do not use this ever')
|
|
)
|
|
}
|
|
attrs = attributes.Attributes('test_rsrc', schema, lambda d: d)
|
|
self.assertEqual(support.SUPPORTED,
|
|
attrs._attributes['foo_sup'].support_status().status)
|
|
self.assertEqual(support.DEPRECATED,
|
|
attrs._attributes['bar_dep'].support_status().status)
|
|
self.assertEqual('Do not use this ever',
|
|
attrs._attributes['bar_dep'].support_status().message)
|
|
|
|
def test_old_attribute_schema_format(self):
|
|
with mock.patch('heat.engine.attributes.warnings'):
|
|
s = 'Test description.'
|
|
self.assertIsInstance(attributes.Schema.from_attribute(s),
|
|
attributes.Schema)
|
|
self.assertEqual('Test description.',
|
|
attributes.Schema.from_attribute(s).description)
|
|
|
|
|
|
class AttributeTest(common.HeatTestCase):
|
|
"""Test the Attribute class."""
|
|
|
|
def test_as_output(self):
|
|
"""Test that Attribute looks right when viewed as an Output."""
|
|
expected = {
|
|
"Value": '{"Fn::GetAtt": ["test_resource", "test1"]}',
|
|
"Description": "The first test attribute"
|
|
}
|
|
attr = attributes.Attribute(
|
|
"test1", attributes.Schema("The first test attribute"))
|
|
self.assertEqual(expected, attr.as_output("test_resource"))
|
|
|
|
|
|
class AttributesTest(common.HeatTestCase):
|
|
"""Test the Attributes class."""
|
|
|
|
attributes_schema = {
|
|
"test1": attributes.Schema("Test attrib 1"),
|
|
"test2": attributes.Schema("Test attrib 2"),
|
|
"test3": attributes.Schema(
|
|
"Test attrib 3",
|
|
cache_mode=attributes.Schema.CACHE_NONE)
|
|
}
|
|
|
|
def setUp(self):
|
|
super(AttributesTest, self).setUp()
|
|
self.addCleanup(self.m.VerifyAll)
|
|
|
|
def test_get_attribute(self):
|
|
"""Test that we get the attribute values we expect."""
|
|
test_resolver = lambda x: "value1"
|
|
self.m.ReplayAll()
|
|
attribs = attributes.Attributes('test resource',
|
|
self.attributes_schema,
|
|
test_resolver)
|
|
self.assertEqual("value1", attribs['test1'])
|
|
|
|
def test_get_attribute_none(self):
|
|
"""Test that we get the attribute values we expect."""
|
|
test_resolver = lambda x: None
|
|
self.m.ReplayAll()
|
|
attribs = attributes.Attributes('test resource',
|
|
self.attributes_schema,
|
|
test_resolver)
|
|
self.assertIsNone(attribs['test1'])
|
|
|
|
def test_get_attribute_nonexist(self):
|
|
"""Test that we get the attribute values we expect."""
|
|
test_resolver = lambda x: "value1"
|
|
self.m.ReplayAll()
|
|
attribs = attributes.Attributes('test resource',
|
|
self.attributes_schema,
|
|
test_resolver)
|
|
self.assertRaises(KeyError, attribs.__getitem__, 'not there')
|
|
|
|
def test_as_outputs(self):
|
|
"""Test that Output format works as expected."""
|
|
expected = {
|
|
"test1": {
|
|
"Value": '{"Fn::GetAtt": ["test_resource", "test1"]}',
|
|
"Description": "Test attrib 1"
|
|
},
|
|
"test2": {
|
|
"Value": '{"Fn::GetAtt": ["test_resource", "test2"]}',
|
|
"Description": "Test attrib 2"
|
|
},
|
|
"test3": {
|
|
"Value": '{"Fn::GetAtt": ["test_resource", "test3"]}',
|
|
"Description": "Test attrib 3"
|
|
}
|
|
}
|
|
MyTestResourceClass = self.m.CreateMockAnything()
|
|
MyTestResourceClass.attributes_schema = {
|
|
"test1": attributes.Schema("Test attrib 1"),
|
|
"test2": attributes.Schema("Test attrib 2"),
|
|
"test3": attributes.Schema("Test attrib 3"),
|
|
}
|
|
self.m.ReplayAll()
|
|
self.assertEqual(
|
|
expected,
|
|
attributes.Attributes.as_outputs("test_resource",
|
|
MyTestResourceClass))
|
|
|
|
def test_caching_local(self):
|
|
value = 'value1'
|
|
test_resolver = lambda x: value
|
|
self.m.ReplayAll()
|
|
attribs = attributes.Attributes('test resource',
|
|
self.attributes_schema,
|
|
test_resolver)
|
|
self.assertEqual("value1", attribs['test1'])
|
|
value = 'value1 changed'
|
|
self.assertEqual("value1", attribs['test1'])
|
|
|
|
attribs.reset_resolved_values()
|
|
self.assertEqual("value1 changed", attribs['test1'])
|
|
|
|
def test_caching_none(self):
|
|
value = 'value3'
|
|
test_resolver = lambda x: value
|
|
self.m.ReplayAll()
|
|
attribs = attributes.Attributes('test resource',
|
|
self.attributes_schema,
|
|
test_resolver)
|
|
self.assertEqual("value3", attribs['test3'])
|
|
value = 'value3 changed'
|
|
self.assertEqual("value3 changed", attribs['test3'])
|
|
|
|
def test_validate_type_invalid(self):
|
|
resolver = mock.Mock()
|
|
# Test invalid string type attribute
|
|
attr_schema = attributes.Schema("Test attribute",
|
|
type=attributes.Schema.STRING)
|
|
attr = attributes.Attribute("test1", attr_schema)
|
|
attribs = attributes.Attributes('test resource', attr_schema, resolver)
|
|
attribs._validate_type(attr, [])
|
|
self.assertIn("Attribute test1 is not of type String", self.LOG.output)
|
|
|
|
# Test invalid list type attribute
|
|
attr_schema = attributes.Schema("Test attribute",
|
|
type=attributes.Schema.LIST)
|
|
attr = attributes.Attribute("test1", attr_schema)
|
|
attribs = attributes.Attributes('test resource', attr_schema, resolver)
|
|
attribs._validate_type(attr, 'invalid')
|
|
self.assertIn("Attribute test1 is not of type List", self.LOG.output)
|
|
|
|
# Test invalid map type attribute
|
|
attr_schema = attributes.Schema("Test attribute",
|
|
type=attributes.Schema.MAP)
|
|
attr = attributes.Attribute("test1", attr_schema)
|
|
attribs = attributes.Attributes('test resource', attr_schema, resolver)
|
|
attribs._validate_type(attr, 'invalid')
|
|
self.assertIn("Attribute test1 is not of type Map", self.LOG.output)
|