# Copyright 2022 Orange, Deutsche Telekom AG # # 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 json import os.path from pathlib import Path from tempfile import TemporaryDirectory from unittest.mock import MagicMock, patch, PropertyMock, mock_open from pytest import raises from onapsdk.cds.blueprint import Blueprint, Mapping, MappingSet, ResolvedTemplate, Workflow from onapsdk.cds.blueprint_processor import Blueprintprocessor from onapsdk.cds.cds_element import CdsElement from onapsdk.cds.data_dictionary import DataDictionary, DataDictionarySet from onapsdk.exceptions import FileError, ParameterError, RequestError, ValidationError from onapsdk.utils.gui import GuiItem, GuiList DD_1 = { "name": "vf-module-name", "tags": "vf-module-name", "data_type": "string", "description": "vf-module-name", "entry_schema": "string", "updatedBy": "Singal, Kapil ", "definition": { "tags": "vf-module-name", "name": "vf-module-name", "property": { "description": "vf-module-name", "type": "string" }, "updated-by": "Singal, Kapil ", "sources": { "input": { "type": "source-input" }, "default": { "type": "source-default", "properties": {} } } } } RAW_DD = { "tags": "vf-module-name", "name": "vf-module-name", "property": { "description": "vf-module-name", "type": "string" }, "updated-by": "Singal, Kapil ", "sources": { "input": { "type": "source-input" }, "default": { "type": "source-default", "properties": {} } } } vLB_CBA_Python_meta_bytes = b'TOSCA-Meta-File-Version: 1.0.0\nCSAR-Version: 1.0\nCreated-By: PLATANIA, MARCO \nEntry-Definitions: Definitions/vLB_CDS.json\nTemplate-Tags: vDNS-CDS-test1\nContent-Type: application/vnd.oasis.bpmn\nTemplate-Name: vDNS-CDS-test1\nTemplate-Version: 1.0' vLB_CBA_Python_base_template_mapping_bytes = b'[\n {\n "name": "service-instance-id",\n "property": {\n "description": "",\n "required": false,\n "type": "string",\n "status": "",\n "constraints": [\n {}\n ],\n "entry_schema": {\n "type": ""\n }\n },\n "input-param": false,\n "dictionary-name": "service-instance-id",\n "dictionary-source": "input",\n "dependencies": [],\n "version": 0\n },\n {\n "name": "vnf-id",\n "property": {\n "description": "",\n "required": false,\n "type": "string",\n "status": "",\n "constraints": [\n {}\n ],\n "entry_schema": {\n "type": ""\n }\n },\n "input-param": false,\n "dictionary-name": "vnf-id",\n "dictionary-source": "input",\n "dependencies": [],\n "version": 0\n },\n {\n "name": "vdns_vf_module_id",\n "property": {\n "description": "",\n "required": false,\n "type": "string",\n "status": "",\n "constraints": [\n {}\n ],\n "entry_schema": {\n "type": ""\n }\n },\n "input-param": false,\n "dictionary-name": "vdns_vf_module_id",\n "dictionary-source": "sdnc",\n "dependencies": [\n\t "service-instance-id",\n "vnf-id"\n ],\n "version": 0\n },\n {\n "name": "vdns_int_private_ip_0",\n "property": {\n "description": "",\n "required": false,\n "type": "string",\n "status": "",\n "constraints": [\n {}\n ],\n "entry_schema": {\n "type": ""\n }\n },\n "input-param": false,\n "dictionary-name": "vdns_int_private_ip_0",\n "dictionary-source": "sdnc",\n "dependencies": [\n "service-instance-id",\n "vnf-id",\n "vdns_vf_module_id"\n ],\n "version": 0\n },\n {\n "name": "vdns_onap_private_ip_0",\n "property": {\n "description": "",\n "required": false,\n "type": "string",\n "status": "",\n "constraints": [\n {}\n ],\n "entry_schema": {\n "type": ""\n }\n },\n "input-param": false,\n "dictionary-name": "vdns_onap_private_ip_0",\n "dictionary-source": "sdnc",\n "dependencies": [\n "service-instance-id",\n "vnf-id",\n "vdns_vf_module_id"\n ],\n "version": 0\n }\n]' @patch.object(Blueprint, "send_message") def test_blueprint_enrichment(send_message_mock): blueprint = Blueprint(b"test cba - it will never work") blueprint.enrich() send_message_mock.assert_called_once() send_message_mock.reset_mock() send_message_mock.side_effect = RequestError with raises(RequestError): blueprint.enrich() @patch.object(Blueprint, "send_message") def test_blueprint_publish(send_message_mock): blueprint = Blueprint(b"test cba - it will never work") blueprint.publish() send_message_mock.assert_called_once() @patch.object(Blueprint, "send_message") def test_blueprint_deploy(send_message_mock): blueprint = Blueprint(b"test cba - it will never work") blueprint.deploy() send_message_mock.assert_called_once() def test_blueprint_load_from_file(): with TemporaryDirectory() as tmpdirname: path = os.path.join(tmpdirname, "test.zip") with open(path, "wb") as f: f.write(b"test cba - it will never work") blueprint = Blueprint.load_from_file(path) assert blueprint.cba_file_bytes == b"test cba - it will never work" def test_blueprint_load_from_file_file_error(): with TemporaryDirectory() as tmpdirname, \ patch("__main__.open", new_callable=mock_open) as mo, \ raises(FileError) as exc: path = os.path.join(tmpdirname, "nonexistent_file.zip") mo.side_effect = FileNotFoundError Blueprint.load_from_file(path) assert exc.type == FileError def test_blueprint_save(): blueprint = Blueprint(b"test cba - it will never work") with TemporaryDirectory() as tmpdirname: path = os.path.join(tmpdirname, "test.zip") blueprint.save(path) with open(path, "rb") as f: assert f.read() == b"test cba - it will never work" def test_blueprint_read_cba_metadata(): b = Blueprint(b"test cba - it will never work") with raises(ValidationError) as exc: b.get_cba_metadata(b"Invalid") b.get_cba_metadata(b"123: 456") assert exc.type is ValidationError cba_metadata = b.get_cba_metadata(vLB_CBA_Python_meta_bytes) assert cba_metadata.tosca_meta_file_version == "1.0.0" assert cba_metadata.csar_version == 1.0 assert cba_metadata.created_by == "PLATANIA, MARCO " assert cba_metadata.entry_definitions == "Definitions/vLB_CDS.json" assert cba_metadata.template_name == "vDNS-CDS-test1" assert cba_metadata.template_version == 1.0 assert cba_metadata.template_tags == "vDNS-CDS-test1" with open(Path(Path(__file__).resolve().parent, "data/vLB_CBA_Python.zip"), "rb") as cba_file: b = Blueprint(cba_file.read()) assert b.metadata.tosca_meta_file_version == "1.0.0" assert b.metadata.csar_version == 1.0 assert b.metadata.created_by == "PLATANIA, MARCO " assert b.metadata.entry_definitions == "Definitions/vLB_CDS.json" assert b.metadata.template_name == "vDNS-CDS-test1" assert b.metadata.template_version == 1.0 assert b.metadata.template_tags == "vDNS-CDS-test1" def test_blueprint_get_mappings_from_mapping_file(): b = Blueprint(b"test cba - it will never work") mappings = list(b.get_mappings_from_mapping_file(vLB_CBA_Python_base_template_mapping_bytes)) assert len(mappings) == 5 mapping = mappings[0] assert mapping.name == "service-instance-id" assert mapping.mapping_type == "string" assert mapping.dictionary_name == "service-instance-id" assert mapping.dictionary_sources == ["input"] def test_blueprint_generate_data_dictionary_set(): with open(Path(Path(__file__).resolve().parent, "data/vLB_CBA_Python.zip"), "rb") as cba_file: b = Blueprint(cba_file.read()) dd_set = b.get_data_dictionaries() print(dd_set) @patch.object(CdsElement, "_url", new_callable=PropertyMock) def test_data_dictionary(cds_element_url_property_mock): cds_element_url_property_mock.return_value = "http://127.0.0.1" with raises(ValidationError) as exc: DataDictionary({}) assert exc.type is ValidationError dd = DataDictionary({}, fix_schema=False) assert dd.url == "http://127.0.0.1/api/v1/dictionary" assert dd.data_dictionary_json == {} dd = DataDictionary(DD_1) dd.name == DD_1["name"] @patch.object(DataDictionary, "send_message") def test_data_dictionary_upload(send_message_mock): dd = DataDictionary(DD_1) dd.upload() send_message_mock.assert_called_once() @patch.object(DataDictionary, "send_message") def test_data_dictionary_set(send_message_mock): dd_set = DataDictionarySet() dd_set.add(DataDictionary(DD_1)) assert dd_set.length == 1 dd_set.add(DataDictionary(DD_1)) assert dd_set.length == 1 dd_set.add(DataDictionary({"name": "test"}, fix_schema=False)) assert dd_set.length == 2 dd_set.upload() assert send_message_mock.call_count == 2 def test_data_dictionary_set_save_to_file_load_from_file(): dd = DataDictionarySet() dd.add(DataDictionary(DD_1)) with TemporaryDirectory() as tmpdirname: path = os.path.join(tmpdirname, "dd.json") dd.save_to_file(path) with open(path, "r") as f: assert f.read() == json.dumps([dd.data_dictionary_json for dd in dd.dd_set], indent=4) dd_2 = DataDictionarySet.load_from_file(path) assert dd.dd_set == dd_2.dd_set def test_data_dictionary_load_from_file_file_error(): with TemporaryDirectory() as tmpdirname, \ patch("__main__.open", new_callable=mock_open) as mo, \ raises(FileError) as exc: path = os.path.join(tmpdirname, "nonexistent_file.zip") mo.side_effect = FileNotFoundError DataDictionarySet.load_from_file(path) assert exc.type == FileError def test_mapping(): m1 = Mapping(name="test", mapping_type="string", dictionary_name="test_dictionary_name", dictionary_sources=["dictionary_source_1"]) m2 = Mapping(name="test", mapping_type="string", dictionary_name="test_dictionary_name", dictionary_sources=["dictionary_source_2"]) assert m1 == m2 m1.merge(m2) assert sorted(m1.dictionary_sources) == ["dictionary_source_1", "dictionary_source_2"] m1.merge(m2) assert sorted(m1.dictionary_sources) == ["dictionary_source_1", "dictionary_source_2"] def test_mapping_set(): ms = MappingSet() assert len(ms) == 0 m1 = Mapping(name="test", mapping_type="string", dictionary_name="test_dictionary_name", dictionary_sources=["dictionary_source_1"]) m2 = Mapping(name="test", mapping_type="string", dictionary_name="test_dictionary_name", dictionary_sources=["dictionary_source_2"]) ms.add(m1) assert len(ms) == 1 ms.add(m2) assert len(ms) == 1 assert sorted(ms[0].dictionary_sources) == ["dictionary_source_1", "dictionary_source_2"] def test_blueprint_get_workflows_from_entry_definitions_file(): with open(Path(Path(__file__).resolve().parent, "data/vLB_CBA_Python.zip"), "rb") as cba_file: b = Blueprint(cba_file.read()) assert len(b.workflows) == 3 workflow = b.workflows[0] assert len(workflow.steps) == 1 assert workflow.steps[0].name == "resource-assignment" assert workflow.steps[0].description == "Resource Assign Workflow" assert workflow.steps[0].target == "resource-assignment" assert len(workflow.inputs) == 2 assert len(workflow.outputs) == 1 def test_blueprint_get_workflow_by_name(): with open(Path(Path(__file__).resolve().parent, "data/vLB_CBA_Python.zip"), "rb") as cba_file: b = Blueprint(cba_file.read()) workflow = b.get_workflow_by_name("resource-assignment") assert workflow.name == "resource-assignment" workflow = b.get_workflow_by_name("config-assign") assert workflow.name == "config-assign" workflow = b.get_workflow_by_name("config-deploy") assert workflow.name == "config-deploy" with raises(ParameterError): b.get_workflow_by_name("non-existing-workflow") @patch.object(Workflow, "send_message") def test_workflow_execute(send_message_mock): metadata = MagicMock(template_name="test", template_version="test") blueprint = MagicMock(metadata=metadata) workflow = Workflow("test_workflow", {}, blueprint) assert len(workflow.steps) == 0 assert len(workflow.inputs) == 0 assert len(workflow.outputs) == 0 workflow.execute({}) send_message_mock.assert_called_once() def test_data_dictionary_validation(): assert DataDictionary(DD_1).has_valid_schema() raw_dd = DataDictionary(RAW_DD, fix_schema=False) assert not raw_dd.has_valid_schema() raw_dd = DataDictionary(RAW_DD, fix_schema=True) assert raw_dd.has_valid_schema() @patch.object(Blueprintprocessor, "send_message") def test_blueprintprocessor_bootstrap(mock_send_message): Blueprintprocessor.bootstrap() mock_send_message.assert_called_once() assert mock_send_message.call_args[1]["data"] == '{\n "loadModelType" : true,\n "loadResourceDictionary" : true,\n "loadCBA" : true\n}' mock_send_message.reset_mock() Blueprintprocessor.bootstrap(load_cba=False, load_model_type=False, load_resource_dictionary=False) mock_send_message.assert_called_once() assert mock_send_message.call_args[1]["data"] == '{\n "loadModelType" : false,\n "loadResourceDictionary" : false,\n "loadCBA" : false\n}' @patch.object(DataDictionary, "send_message_json") def test_data_dictionary_get_by_name(mock_send_message_json): DataDictionary.get_by_name("test_name") mock_send_message_json.assert_called_once() assert "test_name" in mock_send_message_json.call_args[0][2] @patch.object(CdsElement, "send_message") def test_get_guis(send_message_mock): component = CdsElement() send_message_mock.return_value.status_code = 200 send_message_mock.return_value.url = "http://portal.api.simpledemo.onap.org:30449/" gui_results = component.get_guis() assert type(gui_results) == GuiList assert gui_results.guis[0].url == send_message_mock.return_value.url assert gui_results.guis[0].status == send_message_mock.return_value.status_code @patch.object(ResolvedTemplate, "send_message_json") @patch.object(CdsElement, "_url", new_callable=PropertyMock) def test_blueprint_get_resolved_template(cds_element_url_property_mock, mock_send_message_json): cds_element_url_property_mock.return_value = "http://127.0.0.1" with open(Path(Path(__file__).resolve().parent, "data/vLB_CBA_Python.zip"), "rb") as cba_file: b = Blueprint(cba_file.read()) b.get_resolved_template("test_artifact") mock_send_message_json.assert_called_once() assert mock_send_message_json.call_args[0][2] == 'http://127.0.0.1/api/v1/template?bpName=vDNS-CDS-test1&bpVersion=1.0&artifactName=test_artifact&format=application%2Fjson' @patch.object(ResolvedTemplate, "send_message") @patch.object(CdsElement, "_url", new_callable=PropertyMock) def test_blueprint_store_resolved_template(cds_element_url_property_mock, mock_send_message): cds_element_url_property_mock.return_value = "http://127.0.0.1" with open(Path(Path(__file__).resolve().parent, "data/vLB_CBA_Python.zip"), "rb") as cba_file: b = Blueprint(cba_file.read()) b.store_resolved_template("test_artifact", resolution_key="resolution_key", data={"a": "b"}) mock_send_message.assert_called_once() assert mock_send_message.call_args[0][2] == 'http://127.0.0.1/api/v1/template/vDNS-CDS-test1/1.0/test_artifact/resolution_key' @patch.object(ResolvedTemplate, "send_message_json") @patch.object(CdsElement, "_url", new_callable=PropertyMock) def test_resolved_template_get_template_url(cds_element_url_property_mock, mock_send_message_json): cds_element_url_property_mock.return_value = "http://127.0.0.1" blueprint = MagicMock() blueprint.metadata.template_name = "test_blueprint" blueprint.metadata.template_version = "v1.0.0" rt = ResolvedTemplate(blueprint, "test_artifact") rt.get_resolved_template() mock_send_message_json.assert_called_once() assert mock_send_message_json.call_args[0][2] == 'http://127.0.0.1/api/v1/template?bpName=test_blueprint&bpVersion=v1.0.0&artifactName=test_artifact&format=application%2Fjson' mock_send_message_json.reset_mock() blueprint = MagicMock() blueprint.metadata.template_name = "test_blueprint" blueprint.metadata.template_version = "v1.0.0" rt = ResolvedTemplate(blueprint, resolution_key="test_rk") rt.get_resolved_template() mock_send_message_json.assert_called_once() assert mock_send_message_json.call_args[0][2] == 'http://127.0.0.1/api/v1/template?bpName=test_blueprint&bpVersion=v1.0.0&resolutionKey=test_rk&format=application%2Fjson' mock_send_message_json.reset_mock() blueprint = MagicMock() blueprint.metadata.template_name = "test_blueprint" blueprint.metadata.template_version = "v1.0.0" rt = ResolvedTemplate(blueprint, resource_id="r_id", resource_type="r_type") rt.get_resolved_template() mock_send_message_json.assert_called_once() assert mock_send_message_json.call_args[0][2] == 'http://127.0.0.1/api/v1/template?bpName=test_blueprint&bpVersion=v1.0.0&resourceType=r_type&resourceId=r_id&format=application%2Fjson' @patch.object(ResolvedTemplate, "send_message") @patch.object(CdsElement, "_url", new_callable=PropertyMock) def test_resolved_template_store_template_url(cds_element_url_property_mock, mock_send_message): cds_element_url_property_mock.return_value = "http://127.0.0.1" blueprint = MagicMock() blueprint.metadata.template_name = "test_blueprint" blueprint.metadata.template_version = "v1.0.0" rt = ResolvedTemplate(blueprint, "test_artifact", resolution_key="resolution_key") rt.store_resolved_template({"a": "b"}) mock_send_message.assert_called_once() assert mock_send_message.call_args[0][2] == 'http://127.0.0.1/api/v1/template/test_blueprint/v1.0.0/test_artifact/resolution_key' mock_send_message.reset_mock() rt = ResolvedTemplate(blueprint, "test_artifact", resource_id="resource_id", resource_type="resource_type") rt.store_resolved_template({"a": "b"}) mock_send_message.assert_called_once() assert mock_send_message.call_args[0][2] == 'http://127.0.0.1/api/v1/template/test_blueprint/v1.0.0/test_artifact/resource_type/resource_id' mock_send_message.reset_mock() rt = ResolvedTemplate(blueprint, "test_artifact") with raises(ParameterError): rt.store_resolved_template({"a": "b"})