diff options
| -rw-r--r-- | app/tests/test_entry_schema.py | 25 | ||||
| -rw-r--r-- | app/tests/test_substitution_rule.py | 31 | ||||
| -rw-r--r-- | app/tests/test_tosca_builder.py | 78 | ||||
| -rw-r--r-- | app/tests/utils/test_utils.py | 124 | ||||
| -rw-r--r-- | tox.ini | 2 |
5 files changed, 259 insertions, 1 deletions
diff --git a/app/tests/test_entry_schema.py b/app/tests/test_entry_schema.py new file mode 100644 index 0000000..f1e495a --- /dev/null +++ b/app/tests/test_entry_schema.py @@ -0,0 +1,25 @@ +import unittest +from toscalib.types.property import PropertyDefinition +from toscalib.templates.database import ToscaDB + + +class TestEntrySchemaMethods(unittest.TestCase): + + def test_parse_content(self): + content = {'type': 'list', 'entry_schema': {'type': 'integer'}} + prop_def = PropertyDefinition('propName', content) + self.assertFalse(prop_def.parsed) + self.assertIsNone(prop_def.type) + self.assertIsNone(prop_def.type_obj) + prop_def._parse_content(ToscaDB) + self.assertTrue(prop_def.parsed) + self.assertEqual(prop_def.type, 'list') + self.assertTrue(prop_def.type_obj.built_in) + self.assertEqual(prop_def.type_obj.name, 'list') + self.assertTrue(prop_def.type_obj.parsed) + self.assertEqual(prop_def.type_obj.type, 'list') + self.assertEqual(prop_def.type_obj.entry.type, 'integer') + self.assertTrue(prop_def.type_obj.entry.parsed) + self.assertTrue(prop_def.type_obj.entry.type_obj.built_in) + self.assertEqual(prop_def.type_obj.entry.type_obj.name, 'integer') + self.assertEqual(prop_def.type_obj.entry.type_obj.type, 'integer') diff --git a/app/tests/test_substitution_rule.py b/app/tests/test_substitution_rule.py new file mode 100644 index 0000000..aec3745 --- /dev/null +++ b/app/tests/test_substitution_rule.py @@ -0,0 +1,31 @@ +import unittest +from tests.utils.test_utils import * + + +class TestSubstitutionRuleMethods(unittest.TestCase): + + def test_update_pointer(self): + template = init_template() + sub_template = init_sub_template() + src_node = sub_template.node_dict.get('nodeName') + target_node = template.node_dict.get('node2') + sub_sec = {'node_type': 'nodeTypeName', + 'requirements': {'substituteRequirement': ['node2', 'dummyRequirement']}, + 'capabilities': {'substituteCapability': ['node2', 'capabilityName']}} + sub_template._parse_substitution(template.db, sub_sec) + rules = sub_template.sub_rules + self.assertIsNone(src_node.properties[rules[0].property].sub_pointer) + self.assertIsNone(src_node.capabilities[rules[1].item].sub_pointer) + self.assertIsNone(src_node.capabilities[rules[1].item].properties['capabilityProperty'].sub_pointer) + self.assertIsNone(src_node.requirements[0].sub_pointer) + rules[0]._update_pointer(src_node, sub_template) + self.assertIsNotNone(src_node.properties[rules[0].property].sub_pointer) + self.assertEqual(src_node.properties[rules[0].property].sub_pointer, sub_template.inputs[rules[0].property]) + rules[1]._update_pointer(src_node, template) + self.assertIsNotNone(src_node.capabilities[rules[1].item].sub_pointer) + self.assertIsNotNone(src_node.capabilities[rules[1].item].properties['capabilityProperty'].sub_pointer) + self.assertEqual(src_node.capabilities[rules[1].item].sub_pointer, target_node.capabilities[rules[1].value[1]]) + self.assertEqual(src_node.capabilities[rules[1].item].properties['capabilityProperty'].sub_pointer, target_node.capabilities[rules[1].value[1]].properties['capabilityProperty']) + rules[2]._update_pointer(src_node, template) + self.assertIsNotNone(src_node.requirements[0].sub_pointer) + self.assertEqual(src_node.requirements[0].sub_pointer, target_node.requirements[0]) diff --git a/app/tests/test_tosca_builder.py b/app/tests/test_tosca_builder.py new file mode 100644 index 0000000..1e85d8a --- /dev/null +++ b/app/tests/test_tosca_builder.py @@ -0,0 +1,78 @@ +import unittest +from tests.utils.test_utils import init_tosca_builder_with_schema_and_spec, \ + init_tosca_builder_with_policy_schema_and_spec, init_tosca_builder_with_hello_world_spec_k8 +from toscalib.tosca_workbook import ToscaWorkBook + + +class TestToscaBuilderMethods(unittest.TestCase): + + def test_create_node_type(self): + builder = init_tosca_builder_with_schema_and_spec() + self.assertIsNotNone(builder.spec_import.type) + self.assertEqual(builder.spec_import.type, 'docker') + self.assertTrue(builder._using_dmaap()) + self.assertIsNone(builder.cloudify_type) + self.assertIsNone(builder.new_type_name) + self.assertNotIn('tosca.dcae.nodes.dockerApp.test_spec_ss', builder.db.NODE_TYPES) + self.assertNotIn('tosca.dcae.nodes.dockerApp.docker_flag', builder.db.NODE_TYPES) + builder.create_node_type() + self.assertIn('tosca.dcae.nodes.dockerApp.test_spec_ss', builder.db.NODE_TYPES) + self.assertIsNotNone(builder.new_type_name) + self.assertEqual(builder.new_type_name, 'tosca.dcae.nodes.dockerApp.test_spec_ss') + self.assertIsNotNone(builder.cloudify_type) + # TODO uncomment after K8 support integration + # self.assertEqual(builder.cloudify_type, builder.db.NODE_TYPES['dcae.nodes.ContainerizedServiceComponentUsingDmaap']) + # builder.create_node_type('docker_flag', True) + # self.assertIn('tosca.dcae.nodes.dockerApp.docker_flag', builder.db.NODE_TYPES) + # self.assertEqual(builder.new_type_name, 'tosca.dcae.nodes.dockerApp.docker_flag') + # self.assertEqual(builder.cloudify_type, builder.db.NODE_TYPES['dcae.nodes.DockerContainerForComponentsUsingDmaap']) + + def test_create_model(self): + builder = init_tosca_builder_with_schema_and_spec() + builder.create_node_type() + builder.create_model('test_ss') + self.assertIsNotNone(builder.template) + self.assertIsNotNone(builder.template.db) + self.assertEqual(builder.db, builder.template.db) + self.assertIn('test_ss', builder.template.node_dict) + self.assertIn('topic0', builder.template.node_dict) + + def test_create_translate(self): + builder = init_tosca_builder_with_schema_and_spec() + builder.create_node_type() + builder.create_translate('test_ss') + self.assertIsNotNone(builder.template) + self.assertEqual(len(builder.template.sub_rules), 2) + self.assertEqual(builder.template.metadata, {'template_name': 'test_ss_translate'}) + self.assertEqual(builder.db, builder.template.db) + + def test_create_policy(self): + builder = init_tosca_builder_with_policy_schema_and_spec() + self.assertEqual(len(builder.db.DATA_TYPES), 0) + self.assertEqual(len(builder.db.NODE_TYPES), 1) + builder.create_policy() + self.assertEqual(len(builder.db.DATA_TYPES), 3) + self.assertEqual(len(builder.db.NODE_TYPES), 2) + + def test_spec_to_model_to_blueprint_create(self): + #TODO imports section + #TODO assertions + builder = init_tosca_builder_with_hello_world_spec_k8() + name = builder.spec_import.name + filename = 'WEB' + builder.create_node_type(name) + schema = builder.export_schema(filename) + builder.create_model(name) + template = builder.export_model(filename) + builder.create_translate(name) + translate = builder.export_translation(filename) + workbook = ToscaWorkBook() + workbook._import_dir('../data/shared_model/') + workbook._import_yml_str(schema) + workbook._import_yml_str(template) + workbook._import_yml_str(translate) + workbook._translate_template_yaml_str(template) + workbook._add_shared_node( + [{'dcae.capabilities.cdapHost': 'cdap_host'}, {'dcae.capabilities.dockerHost': 'docker_host'}, + {'dcae.capabilities.composition.host': 'composition_virtual'}]) + bp = workbook._export_yaml_web('cloudify,main') diff --git a/app/tests/utils/test_utils.py b/app/tests/utils/test_utils.py index 953683b..af117a5 100644 --- a/app/tests/utils/test_utils.py +++ b/app/tests/utils/test_utils.py @@ -2,6 +2,12 @@ from toscalib.templates.topology import ToscaTopology from toscalib.templates.database import ToscaDB from toscalib.types.node import NodeType from toscalib.types.capability import CapabilityType +from toscalib.tosca_builder import ToscaBuilder +import os + +CURR_DIR = os.path.dirname(os.path.abspath(__file__)) +meta_model = os.path.join(CURR_DIR, os.pardir, '../data/meta_model/meta_tosca_schema.yaml') +policy_model = os.path.join(CURR_DIR, os.pardir, '../data/meta_model/meta_policy_schema.yaml') def init_template(): @@ -22,3 +28,121 @@ def init_template(): template._parse_content(db) return template +def init_sub_template(): + db = ToscaDB() + sub_capability = CapabilityType('tosca.capabilities.substitute', {'properties': {'capabilityProperty': {'type': 'string'}}}) + sub_capability._parse_content(db) + db._import_capability_type(sub_capability) + sub_node = NodeType('substituteNodeType', {'id': 'subNodeId', 'properties': {'propertyName': {'type': 'string'}}, 'capabilities': {'substituteCapability': {'type': 'tosca.capabilities.substitute'}}, 'requirements': [{'substituteRequirement': {'capability': 'tosca.capabilities.substitute'}}]}) + sub_node._parse_content(db) + db._import_node_type(sub_node) + template = ToscaTopology('subTemplateName', None, {'inputs': {'propertyName': {'type': 'string'}}, 'node_templates': {'nodeName': {'type': 'substituteNodeType'}}}) + template._parse_content(db) + return template + + +def init_tosca_builder_with_schema_and_spec(): + spec = {"self": { + "version": "1.1.0", + "name": "test_spec_ss", + "description": "Collector for receiving VES events through restful interface", + "component_type": "docker"}, + "streams": { + "subscribes": [], + "publishes": [{ + "format": "VES_specification", + "version": "5.28.4", + "type": "message router", + "config_key": "ves_sipsignaling"}]}, + "services": { + "provides": [{ + "route": "/eventListener/v5", + "verb": "POST", + "request": { + "format": "VES_specification", + "version": "5.28.4"}, + "response": { + "format": "ves.coll.response", + "version": "1.0.0"}}]}, + "parameters": [{ + "name": "collector.service.port", + "value": 8080, + "description": "standard http port"}, + {"name": "collector.service.secure.port", + "value": 8443, + "description": "secure port "}, + {"name": "collector.keystore.file.location", + "value": "/opt/app/dcae-certificate/keystore.jks", + "description": "fs location of keystore in vm"}], + "auxilary": { + "healthcheck": { + "type": "http", + "interval": "15s", + "timeout": "1s", + "endpoint": "/healthcheck"}}} + + builder = ToscaBuilder() + builder.import_schema(meta_model) + builder.import_spec_str(spec) + return builder + + +def init_tosca_builder_with_policy_schema_and_spec(): + spec = {"self": { + "version": "0.1.6", + "name": "DcaeSamCollector", + "component_type": "docker"}, + "parameters": [{ + "name": "clliLocationMappingClliFutureUse3", + "description": "SAM Collector clli=location ID set", + "value": "", + "type": "string"}, + {"name": "vnfFaultMonProvisionPolicy", + "policy_editable": True, + "policy_group": "DcaeSamCollector_vnfFaultMonProvisionPolicy", + "type": "string", + "policy_schema": [{ + "name": "vnfTypeSpecificData", + "description": "List of objects for vnf type monitorng", + "type": "list", + "entry_schema": [{ + "name": "elementType", + "value": ""}, + {"name": "monitoringTasks", + "type": "list", + "entry_schema": [{ + "name": "HostGroupSetCommonLinuxSNMP", + "type": "boolean", + "value": "false"}, + {"name": "HostGroupSetNagent_Common_Linux", + "type": "boolean", + "value": "false"}] + } + ] + }]} + ]} + + builder = ToscaBuilder() + builder.import_schema(policy_model) + builder.import_spec_str(spec) + return builder + +def init_tosca_builder_with_hello_world_spec_k8(): + spec = {"self": {"component_type": "docker", "description": "Hello World mS for subscribing the data from local DMaaP, DR or MR, processing them and publishing them as PM files to local DMaaP DR", + "name": "dcae.collectors.vcc.helloworld.pm", "version": "1.0.1"}, "services": {"calls": [], "provides": []}, + "streams": {"publishes": [], "subscribes": []}, + "parameters": [{"name": "vcc_hello_name", "value": "", "description": "the name entered for specific person","sourced_at_deployment": True, "designer_editable": True, "policy_editable": False}, + {"name": "useDtiConfig", "value": False, "description": "component depends on configuration from dti.", "sourced_at_deployment": "false", "designer_editable": "false", "policy_editable": False, "required": True}, + {"name": "isSelfServeComponent", "value": "false", "description": "Is this used as self serve component.", "sourced_at_deployment": False, "designer_editable": False, "policy_editable": False, "required": "true"}], + "auxilary": {"healthcheck": {"interval": "60s", "timeout": "20s", "script": "/opt/app/vcc/bin/common/HealthCheck_HelloWorld.sh", "type": "docker"}, + "volumes": [{"container": {"bind": "/opt/app/dcae-certificate"}, "host": {"path": "/opt/app/dcae-certificate"}}, + {"container": {"bind": "/opt/logs/DCAE/dmd/AGENT"}, "host": {"path": "/opt/logs/DCAE/helloworldpm/dmd/AGENT"}}, + {"container": {"bind": "/opt/logs/DCAE/dmd/WATCHER"}, "host": {"path": "/opt/logs/DCAE/helloworldpm/dmd/WATCHER"}}, + {"container": {"bind": "/opt/app/vcc/logs/DCAE"}, "host": {"path": "/opt/logs/DCAE/helloworldpm/vcc-logs"}}, + {"container": {"bind": "/opt/app/vcc/archive/data"}, "host": {"path": "/opt/data/DCAE/helloworldpm/vcc-archive"}}]}, + "artifacts": [{"type": "docker image", "uri": "dockercentral.it.att.com:5100/com.att.dcae.controller/dcae-controller-vcc-helloworld-pm:18.02-001"}]} + + builder = ToscaBuilder() + builder.import_schema(meta_model) + builder.import_spec_str(spec) + return builder @@ -14,7 +14,7 @@ deps = -r{toxinidir}/app/requirements.txt commands = python --version coverage erase - coverage run --source {toxinidir}/app/toscalib --omit="{toxinidir}/app/toscalib/tosca*.py" -m unittest + coverage run --source {toxinidir}/app/toscalib -m unittest coverage report coverage xml |