360 lines
18 KiB
Python
360 lines
18 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.
|
|
|
|
from toscaparser.common import exception
|
|
from toscaparser.elements.artifacttype import ArtifactTypeDef
|
|
from toscaparser.elements.entity_type import EntityType
|
|
from toscaparser.elements.grouptype import GroupType
|
|
import toscaparser.elements.interfaces as ifaces
|
|
from toscaparser.elements.nodetype import NodeType
|
|
from toscaparser.elements.policytype import PolicyType
|
|
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')
|
|
policy_root_type = PolicyType('tosca.policies.Root')
|
|
policy_placement_type = PolicyType('tosca.policies.Placement')
|
|
policy_scaling_type = PolicyType('tosca.policies.Scaling')
|
|
policy_reservation_type = PolicyType('tosca.policies.Reservation')
|
|
policy_update_type = PolicyType('tosca.policies.Update')
|
|
policy_performance_type = PolicyType('tosca.policies.Performance')
|
|
group_type = GroupType('tosca.groups.Root')
|
|
|
|
|
|
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_group(self):
|
|
self.assertEqual(group_type.type, "tosca.groups.Root")
|
|
self.assertIsNone(group_type.parent_type)
|
|
self.assertIn(ifaces.LIFECYCLE_SHORTNAME, group_type.interfaces)
|
|
|
|
def test_capabilities(self):
|
|
# Assure the normative Compute node type
|
|
# has all the required Capability types
|
|
# regardless of symbloc name
|
|
# TODO(Matt) - since Compute IS a normative node type
|
|
# we SHOULD test symbolic capability names as well
|
|
self.assertEqual(
|
|
['tosca.capabilities.Container',
|
|
'tosca.capabilities.Endpoint.Admin',
|
|
'tosca.capabilities.Node',
|
|
'tosca.capabilities.OperatingSystem',
|
|
'tosca.capabilities.Scalable',
|
|
'tosca.capabilities.network.Bindable'],
|
|
sorted([c.type for c in compute_type.get_capabilities_objects()]))
|
|
# Assure the normative Network node type
|
|
# hsa all the required Capability types
|
|
# TODO(Matt) - since Network IS a normative node type
|
|
# we SHOULD test symbolic capability names as well
|
|
self.assertEqual(
|
|
['tosca.capabilities.Node',
|
|
'tosca.capabilities.network.Linkable'],
|
|
sorted([c.type for c in network_type.get_capabilities_objects()]))
|
|
|
|
# Assure the normative WebServer node type's
|
|
# Endpoint cap. has all required property names
|
|
# Note: we are testing them in alphabetic sort order
|
|
endpoint_props_def_objects = \
|
|
self._get_capability_properties_def_objects(
|
|
webserver_type.get_capabilities_objects(),
|
|
'tosca.capabilities.Endpoint')
|
|
# Assure WebServer's Endpoint capability's properties have their
|
|
# required keyname value set correctly
|
|
self.assertEqual(
|
|
[('initiator', False), ('network_name', False), ('port', False),
|
|
('port_name', False), ('ports', False), ('protocol', True),
|
|
('secure', False), ('url_path', False)],
|
|
sorted([(p.name, p.required) for p in endpoint_props_def_objects]))
|
|
|
|
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]))
|
|
endpoint_admin_properties = 'secure'
|
|
endpoint_admin_props_def_objects = \
|
|
self._get_capability_properties_def_objects(
|
|
webserver_type.get_capabilities_objects(),
|
|
'tosca.capabilities.Endpoint.Admin')
|
|
self.assertIn(
|
|
endpoint_admin_properties,
|
|
sorted([p.name for p in endpoint_admin_props_def_objects]))
|
|
|
|
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 test_properties_def(self):
|
|
self.assertEqual(
|
|
['name', 'password', 'port', 'user'],
|
|
sorted(database_type.get_properties_def().keys()))
|
|
|
|
def test_attributes_def(self):
|
|
self.assertEqual(
|
|
['networks', 'ports', 'private_address', 'public_address',
|
|
'state', 'tosca_id', 'tosca_name'],
|
|
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'}},
|
|
{'dependency': {'capability': 'tosca.capabilities.Node',
|
|
'node': 'tosca.nodes.Root',
|
|
'occurrences': [0, 'UNBOUNDED'],
|
|
'relationship': 'tosca.relationships.DependsOn'}}
|
|
],
|
|
[r for r in component_type.requirements])
|
|
|
|
def test_relationship(self):
|
|
self.assertEqual(
|
|
[('tosca.relationships.DependsOn', 'tosca.nodes.Root'),
|
|
('tosca.relationships.HostedOn', 'tosca.nodes.Compute')],
|
|
sorted([(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.assertIsNone(compute_type.interfaces)
|
|
root_node = NodeType('tosca.nodes.Root')
|
|
self.assertIn(ifaces.LIFECYCLE_SHORTNAME, root_node.interfaces)
|
|
|
|
def test_artifacts(self):
|
|
self.assertIsNone(artif_root_type.parent_type)
|
|
self.assertEqual('tosca.artifacts.Root',
|
|
artif_file_type.parent_type.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.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.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.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.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.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)))
|
|
|
|
def test_policies(self):
|
|
self.assertIsNone(policy_root_type.parent_type)
|
|
self.assertEqual('tosca.policies.Root',
|
|
policy_placement_type.parent_type.type)
|
|
self.assertEqual({}, policy_placement_type.parent_policies)
|
|
self.assertEqual(sorted(['tosca.policies.Root',
|
|
'The TOSCA Policy Type definition that is '
|
|
'used to govern placement of TOSCA nodes or '
|
|
'groups of nodes.'],
|
|
key=lambda x: str(x)),
|
|
sorted([policy_placement_type.get_policy(name)
|
|
for name in policy_placement_type.defs],
|
|
key=lambda x: str(x)))
|
|
|
|
self.assertEqual('tosca.policies.Root',
|
|
policy_scaling_type.parent_type.type)
|
|
self.assertEqual({}, policy_scaling_type.parent_policies)
|
|
self.assertEqual(sorted(['tosca.policies.Root',
|
|
'The TOSCA Policy Type definition that is '
|
|
'used to govern scaling of TOSCA nodes or '
|
|
'groups of nodes.'],
|
|
key=lambda x: str(x)),
|
|
sorted([policy_scaling_type.get_policy(name)
|
|
for name in policy_scaling_type.defs],
|
|
key=lambda x: str(x)))
|
|
|
|
self.assertEqual('tosca.policies.Root',
|
|
policy_update_type.parent_type.type)
|
|
self.assertEqual({}, policy_update_type.parent_policies)
|
|
self.assertEqual(sorted(['tosca.policies.Root',
|
|
'The TOSCA Policy Type definition that is '
|
|
'used to govern update of TOSCA nodes or '
|
|
'groups of nodes.'],
|
|
key=lambda x: str(x)),
|
|
sorted([policy_update_type.get_policy(name)
|
|
for name in policy_update_type.defs],
|
|
key=lambda x: str(x)))
|
|
|
|
self.assertEqual('tosca.policies.Root',
|
|
policy_performance_type.parent_type.type)
|
|
self.assertEqual({}, policy_performance_type.parent_policies)
|
|
self.assertEqual(sorted(['tosca.policies.Root',
|
|
'The TOSCA Policy Type definition that is '
|
|
'used to declare performance requirements '
|
|
'for TOSCA nodes or groups of nodes.'],
|
|
key=lambda x: str(x)),
|
|
sorted([policy_performance_type.get_policy(name)
|
|
for name in policy_performance_type.defs],
|
|
key=lambda x: str(x)))
|
|
|
|
self.assertEqual('tosca.policies.Root',
|
|
policy_reservation_type.parent_type.type)
|
|
self.assertEqual({}, policy_reservation_type.parent_policies)
|
|
self.assertEqual(sorted(['tosca.policies.Root',
|
|
'The TOSCA Policy Type definition that '
|
|
'is used to create TOSCA nodes or group '
|
|
'of nodes based on the reservation.'],
|
|
key=lambda x: str(x)),
|
|
sorted([policy_reservation_type.get_policy(name)
|
|
for name in policy_reservation_type.defs],
|
|
key=lambda x: str(x)))
|
|
|
|
def test_port_spec(self):
|
|
tosca_def = EntityType.TOSCA_DEF
|
|
port_spec = tosca_def.get('tosca.datatypes.network.PortSpec')
|
|
self.assertEqual(port_spec.get('derived_from'),
|
|
'tosca.datatypes.Root')
|
|
properties = port_spec.get('properties')
|
|
self.assertEqual(
|
|
sorted(['protocol', 'target', 'target_range', 'source',
|
|
'source_range']),
|
|
sorted(properties.keys()))
|