diff options
Diffstat (limited to 'tests/test_pnf.py')
| -rw-r--r-- | tests/test_pnf.py | 485 |
1 files changed, 485 insertions, 0 deletions
diff --git a/tests/test_pnf.py b/tests/test_pnf.py new file mode 100644 index 0000000..0f606a0 --- /dev/null +++ b/tests/test_pnf.py @@ -0,0 +1,485 @@ +"""Test pnf module.""" +# 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. + +from unittest import mock +from unittest.mock import MagicMock +from pathlib import Path + +import pytest + +import onapsdk.constants as const +from onapsdk.exceptions import ParameterError, RequestError, StatusError +from onapsdk.sdc.category_management import ResourceCategory +from onapsdk.sdc.properties import Property +from onapsdk.sdc.sdc_resource import SdcResource +from onapsdk.sdc.pnf import Pnf +from onapsdk.sdc.vsp import Vsp +from onapsdk.sdc.vsp import Vendor + + +@mock.patch.object(Pnf, 'send_message_json') +def test_get_all_no_pnf(mock_send): + """Returns empty array if no pnfs.""" + mock_send.return_value = {} + assert Pnf.get_all() == [] + mock_send.assert_called_once_with("GET", 'get Pnfs', 'https://sdc.api.be.simpledemo.onap.org:30204/sdc/v1/catalog/resources?resourceType=PNF') + + +@mock.patch.object(Pnf, 'send_message_json') +def test_get_all_some_pnfs(mock_send): + """Returns a list of pnfs.""" + mock_send.return_value = [ + {'resourceType': 'PNF', 'name': 'one', 'uuid': '1234', 'invariantUUID': '5678', 'version': '1.0', 'lifecycleState': 'CERTIFIED', 'category': 'Generic', "subCategory": "Abstract"}, + {'resourceType': 'PNF', 'name': 'two', 'uuid': '1235', 'invariantUUID': '5679', 'version': '1.0', 'lifecycleState': 'NOT_CERTIFIED_CHECKOUT', 'category': 'Generic', "subCategory": "Abstract"}] + all_pnfs = Pnf.get_all() + assert len(all_pnfs) == 2 + pnf_1 = all_pnfs[0] + assert pnf_1.name == "one" + assert pnf_1.identifier == "1234" + assert pnf_1.unique_uuid == "5678" + assert pnf_1.version == "1.0" + assert pnf_1.status == const.CERTIFIED + assert pnf_1.created() + pnf_2 = all_pnfs[1] + assert pnf_2.name == "two" + assert pnf_2.identifier == "1235" + assert pnf_2.unique_uuid == "5679" + assert pnf_2.status == const.DRAFT + assert pnf_2.version == "1.0" + assert pnf_2.created() + mock_send.assert_called_once_with("GET", 'get Pnfs', 'https://sdc.api.be.simpledemo.onap.org:30204/sdc/v1/catalog/resources?resourceType=PNF') + + +def test_init_no_name(): + """Check init with no names.""" + pnf = Pnf() + assert isinstance(pnf, SdcResource) + assert pnf._identifier is None + assert pnf._version is None + assert pnf.name == "ONAP-test-PNF" + assert pnf.headers["USER_ID"] == "cs0008" + assert pnf.vsp is None + assert pnf.vendor is None + assert isinstance(pnf._base_url(), str) + +@mock.patch.object(Pnf, 'exists') +def test_init_with_name(mock_exists): + """Check init with names.""" + mock_exists.return_value = False + pnf = Pnf(name="YOLO") + assert pnf._identifier == None + assert pnf._version == None + assert pnf.name == "YOLO" + assert pnf.created() == False + assert pnf.headers["USER_ID"] == "cs0008" + assert pnf.vsp == None + assert isinstance(pnf._base_url(), str) + + +def test_equality_really_equals(): + """Check two pnfs are equals if name is the same.""" + pnf_1 = Pnf(name="equal") + pnf_1.identifier = "1234" + pnf_2 = Pnf(name="equal") + pnf_2.identifier = "1235" + assert pnf_1 == pnf_2 + + +def test_equality_not_equals(): + """Check two pnfs are not equals if name is not the same.""" + pnf_1 = Pnf(name="equal") + pnf_1.identifier = "1234" + pnf_2 = Pnf(name="not_equal") + pnf_2.identifier = "1234" + assert pnf_1 != pnf_2 + + +def test_equality_not_equals_not_same_object(): + """Check a pnf and something different are not equals.""" + pnf_1 = Pnf(name="equal") + pnf_1.identifier = "1234" + pnf_2 = SdcResource() + pnf_2.name = "equal" + assert pnf_1 != pnf_2 + + +@mock.patch.object(Pnf, 'get_all') +def test_exists_not_exists(mock_get_all): + """Return False if pnf doesn't exist in SDC.""" + pnf_1 = Pnf(name="one") + pnf_1.identifier = "1234" + mock_get_all.return_value = [pnf_1] + pnf = Pnf(name="two") + assert not pnf.exists() + + +@mock.patch.object(Pnf, 'get_all') +def test_exists(mock_get_all): + """Return True if pnf exists in SDC.""" + pnf_1 = Pnf(name="one") + pnf_1.identifier = "1234" + pnf_1.unique_uuid = "5689" + pnf_1.unique_identifier = "71011" + pnf_1.status = const.DRAFT + pnf_1.version = "1.1" + mock_get_all.return_value = [pnf_1] + pnf = Pnf(name="one") + assert pnf.exists() + assert pnf.identifier == "1234" + assert pnf.unique_uuid == "5689" + assert pnf.unique_identifier == "71011" + assert pnf.status == const.DRAFT + assert pnf.version == "1.1" + + +@mock.patch.object(Pnf, 'exists') +def test_load_created(mock_exists): + """Load is a wrapper around exists().""" + pnf = Pnf(name="one") + pnf.load() + mock_exists.assert_called_once() + + +@mock.patch.object(Pnf, 'exists') +def test_create_no_vsp_no_vendor(mock_exists): + """Do nothing if no vsp and no vendor""" + pnf = Pnf() + mock_exists.return_value = False + with pytest.raises(ParameterError) as err: + pnf.create() + assert err.type == ParameterError + assert str(err.value) == "Neither Vsp nor Vendor provided." + + +@mock.patch.object(Pnf, 'exists') +@mock.patch.object(Pnf, 'send_message_json') +@mock.patch.object(Pnf, "category") +def test_create_already_exists(mock_category, mock_send, mock_exists): + """Do nothing if already created in SDC.""" + pnf = Pnf() + vsp = Vsp() + vsp._identifier = "1232" + pnf.vsp = vsp + mock_exists.return_value = True + pnf.create() + mock_send.assert_not_called() + + +@mock.patch.object(Pnf, 'exists') +@mock.patch.object(Pnf, 'send_message_json') +@mock.patch.object(Pnf, "category", new_callable=mock.PropertyMock) +def test_create_issue_in_creation(mock_category, mock_send, mock_exists): +# def test_create_issue_in_creation(mock_send, mock_exists): + """Do nothing if not created but issue during creation.""" + pnf = Pnf() + vsp = Vsp() + vendor = Vendor() + vsp._identifier = "1232" + vsp.create_csar = MagicMock(return_value=True) + vsp.vendor = vendor + pnf.vsp = vsp + expected_data = '{\n "artifacts": {},\n "attributes": [],\n "capabilities": {},\n "categories": [\n {\n "normalizedName": "generic",\n "name": "Generic",\n "uniqueId": "resourceNewCategory.generic",\n "subcategories": [{"empty": false, "groupings": null, "icons": ["objectStorage", "compute"], "name": "Abstract", "normalizedName": "abstract", "ownerId": null, "type": null, "uniqueId": "resourceNewCategory.generic.abstract", "version": null}],\n "version": null,\n "ownerId": null,\n "empty": false,\n "type": null,\n "icons": null\n }\n ],\n "componentInstances": [],\n "componentInstancesAttributes": {},\n "componentInstancesProperties": {},\n "componentType": "RESOURCE",\n "contactId": "cs0008",\n \n "csarUUID": "None",\n "csarVersion": "1.0",\n "vendorName": "Generic-Vendor",\n \n "deploymentArtifacts": {},\n "description": "PNF",\n "icon": "defaulticon",\n "name": "ONAP-test-PNF",\n "properties": [],\n "groups": [],\n "requirements": {},\n "resourceType": "PNF",\n "tags": ["ONAP-test-PNF"],\n "toscaArtifacts": {},\n "vendorRelease": "1.0"\n}' + mock_exists.return_value = False + mock_send.side_effect = RequestError + rc = ResourceCategory( + name="Generic" + ) + rc.normalized_name="generic" + rc.unique_id="resourceNewCategory.generic" + rc.subcategories=[{"empty": False, "groupings": None, "icons": ["objectStorage", "compute"], "name": "Abstract", "normalizedName": "abstract", "ownerId": None, "type": None, "uniqueId": "resourceNewCategory.generic.abstract", "version": None}] + rc.version=None + rc.owner_id=None + rc.empty=False + rc.type=None + rc.icons=None + mock_category.return_value = rc + with pytest.raises(RequestError) as exc: + pnf.create() + mock_send.assert_called_once_with("POST", "create Pnf", 'https://sdc.api.fe.simpledemo.onap.org:30207/sdc1/feProxy/rest/v1/catalog/resources', data=expected_data) + assert not pnf.created() + + +@mock.patch.object(Pnf, 'exists') +@mock.patch.object(Pnf, 'send_message_json') +@mock.patch.object(Pnf, "category", new_callable=mock.PropertyMock) +def test_create_OK(mock_category, mock_send, mock_exists): + """Create and update object.""" + pnf = Pnf() + vsp = Vsp() + vendor = Vendor() + vsp._identifier = "1232" + pnf.vsp = vsp + vsp.vendor = vendor + vsp._csar_uuid = "1234" + expected_data = '{\n "artifacts": {},\n "attributes": [],\n "capabilities": {},\n "categories": [\n {\n "normalizedName": "generic",\n "name": "Generic",\n "uniqueId": "resourceNewCategory.generic",\n "subcategories": [{"empty": false, "groupings": null, "icons": ["objectStorage", "compute"], "name": "Abstract", "normalizedName": "abstract", "ownerId": null, "type": null, "uniqueId": "resourceNewCategory.generic.abstract", "version": null}],\n "version": null,\n "ownerId": null,\n "empty": false,\n "type": null,\n "icons": null\n }\n ],\n "componentInstances": [],\n "componentInstancesAttributes": {},\n "componentInstancesProperties": {},\n "componentType": "RESOURCE",\n "contactId": "cs0008",\n \n "csarUUID": "1234",\n "csarVersion": "1.0",\n "vendorName": "Generic-Vendor",\n \n "deploymentArtifacts": {},\n "description": "PNF",\n "icon": "defaulticon",\n "name": "ONAP-test-PNF",\n "properties": [],\n "groups": [],\n "requirements": {},\n "resourceType": "PNF",\n "tags": ["ONAP-test-PNF"],\n "toscaArtifacts": {},\n "vendorRelease": "1.0"\n}' + mock_exists.return_value = False + mock_send.return_value = {'resourceType': 'PNF', 'name': 'one', 'uuid': '1234', 'invariantUUID': '5678', 'version': '1.0', 'uniqueId': '91011', 'lifecycleState': 'NOT_CERTIFIED_CHECKOUT'} + rc = ResourceCategory( + name="Generic" + ) + rc.normalized_name="generic" + rc.unique_id="resourceNewCategory.generic" + rc.subcategories=[{"empty": False, "groupings": None, "icons": ["objectStorage", "compute"], "name": "Abstract", "normalizedName": "abstract", "ownerId": None, "type": None, "uniqueId": "resourceNewCategory.generic.abstract", "version": None}] + rc.version=None + rc.owner_id=None + rc.empty=False + rc.type=None + rc.icons=None + mock_category.return_value = rc + pnf.create() + mock_send.assert_called_once_with("POST", "create Pnf", 'https://sdc.api.fe.simpledemo.onap.org:30207/sdc1/feProxy/rest/v1/catalog/resources', data=expected_data) + assert pnf.created() + assert pnf._status == const.DRAFT + assert pnf.identifier == "1234" + assert pnf.unique_uuid == "5678" + assert pnf.version == "1.0" + +@mock.patch.object(Pnf, 'exists') +@mock.patch.object(Pnf, 'load') +def test_version_no_load_no_created(mock_load, mock_exists): + """Test versions when not created.""" + mock_exists.return_value = False + pnf = Pnf() + assert pnf.version is None + mock_load.assert_not_called() + +@mock.patch.object(Pnf, 'load') +def test_version_no_load_created(mock_load): + """Test versions when created.""" + pnf = Pnf() + pnf.identifier = "1234" + pnf._version = "64" + assert pnf.version == "64" + mock_load.assert_not_called() + + +@mock.patch.object(Pnf, 'load') +def test_version_with_load(mock_load): + """Test versions when not created but with identifier.""" + pnf = Pnf() + pnf.identifier = "1234" + assert pnf.version is None + mock_load.assert_called_once() + +@mock.patch.object(Pnf, 'exists') +@mock.patch.object(Pnf, 'load') +def test_status_no_load_no_created(mock_load, mock_exists): + """Test status when not created.""" + mock_exists.return_value = False + pnf = Pnf() + assert pnf.status is None + + +@pytest.mark.parametrize("status", [const.COMMITED, const.CERTIFIED, const.UPLOADED, const.VALIDATED]) +@mock.patch.object(Pnf, 'exists') +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'send_message') +def test_submit_not_Commited(mock_send, mock_load, mock_exists, status): + """Do nothing if not created.""" + mock_exists.return_value = False + pnf = Pnf() + pnf._status = status + pnf.submit() + mock_send.assert_not_called() + +@mock.patch.object(Pnf, 'exists') +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'send_message') +def test_submit_OK(mock_send, mock_load, mock_exists): + """Don't update status if submission NOK.""" + mock_exists.return_value = True + pnf = Pnf() + pnf._status = const.COMMITED + expected_data = '{\n "userRemarks": "certify"\n}' + pnf._version = "1234" + pnf._unique_identifier = "12345" + pnf.submit() + mock_send.assert_called_once_with( + "POST", "Certify Pnf", + 'https://sdc.api.fe.simpledemo.onap.org:30207/sdc1/feProxy/rest/v1/catalog/resources/12345/lifecycleState/Certify', + data=expected_data) + + +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'certify') +@mock.patch.object(Pnf, 'submit') +@mock.patch.object(Pnf, 'create') +def test_onboard_new_pnf(mock_create, mock_submit, mock_certify, mock_load): + getter_mock = mock.Mock(wraps=Pnf.status.fget) + mock_status = Pnf.status.getter(getter_mock) + with mock.patch.object(Pnf, 'status', mock_status): + getter_mock.side_effect = [None, const.APPROVED, const.APPROVED, + const.APPROVED, const.APPROVED] + vsp = Vsp() + pnf = Pnf(vsp=vsp) + pnf._time_wait = 0 + pnf.onboard() + mock_create.assert_called_once() + mock_submit.assert_not_called() + mock_certify.assert_not_called() + mock_load.assert_not_called() + +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'certify') +@mock.patch.object(Pnf, 'submit') +@mock.patch.object(Pnf, 'create') +def test_onboard_pnf_submit(mock_create, mock_submit, mock_certify, mock_load): + getter_mock = mock.Mock(wraps=Pnf.status.fget) + mock_status = Pnf.status.getter(getter_mock) + with mock.patch.object(Pnf, 'status', mock_status): + getter_mock.side_effect = [const.DRAFT, const.DRAFT, const.APPROVED, + const.APPROVED, const.APPROVED, const.APPROVED] + pnf = Pnf() + pnf._time_wait = 0 + pnf.onboard() + mock_create.assert_not_called() + mock_submit.assert_called_once() + mock_certify.assert_not_called() + mock_load.assert_not_called() + +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'certify') +@mock.patch.object(Pnf, 'submit') +@mock.patch.object(Pnf, 'create') +def test_onboard_pnf_certify(mock_create, mock_submit, mock_certify, mock_load): + getter_mock = mock.Mock(wraps=Pnf.status.fget) + mock_status = Pnf.status.getter(getter_mock) + with mock.patch.object(Pnf, 'status', mock_status): + getter_mock.side_effect = [const.CHECKED_IN, const.CHECKED_IN, const.CHECKED_IN, + const.APPROVED, const.APPROVED, const.APPROVED, + const.APPROVED] + pnf = Pnf() + pnf._time_wait = 0 + pnf.onboard() + mock_create.assert_not_called() + mock_submit.assert_not_called() + mock_certify.assert_called_once() + mock_load.assert_not_called() + +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'certify') +@mock.patch.object(Pnf, 'submit') +@mock.patch.object(Pnf, 'create') +def test_onboard_pnf_load(mock_create, mock_submit, mock_certify, mock_load): + getter_mock = mock.Mock(wraps=Pnf.status.fget) + mock_status = Pnf.status.getter(getter_mock) + with mock.patch.object(Pnf, 'status', mock_status): + getter_mock.side_effect = [const.CERTIFIED, const.CERTIFIED, + const.CERTIFIED, const.CERTIFIED, + const.APPROVED, const.APPROVED, + const.APPROVED] + pnf = Pnf() + pnf._time_wait = 0 + pnf.onboard() + mock_create.assert_not_called() + mock_submit.assert_not_called() + mock_certify.assert_not_called() + mock_load.assert_called_once() + +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'submit') +@mock.patch.object(Pnf, 'create') +@mock.patch.object(Pnf, 'certify') +def test_onboard_whole_pnf_vsp(mock_certify, mock_create, mock_submit, mock_load): + """Test onboarding with vsp""" + getter_mock = mock.Mock(wraps=Pnf.status.fget) + mock_status = Pnf.status.getter(getter_mock) + with mock.patch.object(Pnf, 'status', mock_status): + getter_mock.side_effect = [None, const.DRAFT, const.DRAFT, + const.CHECKED_IN, const.CHECKED_IN, const.CHECKED_IN, + const.CERTIFIED, const.CERTIFIED, const.CERTIFIED, + const.CERTIFIED, const.APPROVED, const.APPROVED, + const.APPROVED, const.APPROVED] + vsp = Vsp() + pnf = Pnf(vsp=vsp) + pnf._time_wait = 0 + pnf.onboard() + mock_create.assert_called_once() + mock_submit.assert_called_once() + mock_load.assert_called_once() + mock_certify.assert_called_once() + +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'submit') +@mock.patch.object(Pnf, 'create') +@mock.patch.object(Pnf, 'certify') +def test_onboard_whole_pnf_vendor(mock_certify, mock_create, mock_submit, mock_load): + """Test onboarding with vendor""" + getter_mock = mock.Mock(wraps=Pnf.status.fget) + mock_status = Pnf.status.getter(getter_mock) + with mock.patch.object(Pnf, 'status', mock_status): + getter_mock.side_effect = [None, const.DRAFT, const.DRAFT, + const.CHECKED_IN, const.CHECKED_IN, const.CHECKED_IN, + const.CERTIFIED, const.CERTIFIED, const.CERTIFIED, + const.CERTIFIED, const.APPROVED, const.APPROVED, + const.APPROVED, const.APPROVED] + vendor = Vendor() + pnf = Pnf(vendor=vendor) + pnf._time_wait = 0 + pnf.onboard() + mock_create.assert_called_once() + mock_submit.assert_called_once() + mock_load.assert_called_once() + mock_certify.assert_called_once() + +@mock.patch.object(Pnf, "send_message_json") +def test_add_properties(mock_send_message_json): + pnf = Pnf(name="test") + pnf._identifier = "toto" + pnf._unique_identifier = "toto" + pnf._status = const.CERTIFIED + with pytest.raises(StatusError) as err: + pnf.add_property(Property(name="test", property_type="string")) + pnf._status = const.DRAFT + pnf.add_property(Property(name="test", property_type="string")) + mock_send_message_json.assert_called_once() + +@mock.patch.object(Pnf, 'load') +@mock.patch.object(Pnf, 'send_message') +def test_add_artifact_to_pnf(mock_send_message, mock_load): + """Test Pnf add artifact""" + pnf = Pnf(name="test") + pnf.status = const.DRAFT + mycbapath = Path(Path(__file__).resolve().parent, "data/vLB_CBA_Python.zip") + + result = pnf.add_deployment_artifact(artifact_label="cba", + artifact_type="CONTROLLER_BLUEPRINT_ARCHIVE", + artifact_name="vLB_CBA_Python.zip", + artifact=mycbapath) + mock_send_message.assert_called() + method, description, url = mock_send_message.call_args[0] + assert method == "POST" + assert description == "Add deployment artifact for test sdc resource" + assert url == ("https://sdc.api.fe.simpledemo.onap.org:30207/sdc1/feProxy/rest/v1/catalog/resources/" + f"{pnf.unique_identifier}/artifacts") + + +@mock.patch.object(Pnf, "created") +@mock.patch.object(ResourceCategory, "get") +def test_pnf_category(mock_resource_category, mock_created): + mock_created.return_value = False + pnf = Pnf(name="test") + _ = pnf.category + mock_resource_category.assert_called_once_with(name="Generic", subcategory="Abstract") + mock_resource_category.reset_mock() + + pnf = Pnf(name="test", category="test", subcategory="test") + _ = pnf.category + mock_resource_category.assert_called_once_with(name="test", subcategory="test") + mock_resource_category.reset_mock() + + mock_created.return_value = True + _ = pnf.category + mock_resource_category.assert_called_once_with(name="test", subcategory="test") |