diff options
Diffstat (limited to 'integration_tests/test_07_instantiation.py')
| -rw-r--r-- | integration_tests/test_07_instantiation.py | 427 |
1 files changed, 427 insertions, 0 deletions
diff --git a/integration_tests/test_07_instantiation.py b/integration_tests/test_07_instantiation.py new file mode 100644 index 0000000..c6cbc85 --- /dev/null +++ b/integration_tests/test_07_instantiation.py @@ -0,0 +1,427 @@ +# 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 time +from unittest.mock import MagicMock, PropertyMock, patch +from uuid import uuid4 + +import pytest +import requests +from onapsdk.exceptions import StatusError +from onapsdk.aai.business import Customer, ServiceInstance +from onapsdk.aai.cloud_infrastructure import CloudRegion +from onapsdk.configuration import settings +from onapsdk.sdc.service import Service, Vnf, VfModule +from onapsdk.so.deletion import ServiceDeletionRequest, VfModuleDeletionRequest, VnfDeletionRequest +from onapsdk.so.instantiation import (ServiceInstantiation, SoService, + VfModuleInstantiation, VnfInstantiation, InstantiationParameter, + VfmoduleParameters, VnfParameters) + + +@pytest.mark.integration +def test_a_la_carte_instantiation(): + requests.get(f"{ServiceInstantiation.base_url}/reset") + requests.get(f"{Customer.base_url}/reset") + requests.post(f"{ServiceInstantiation.base_url}/set_aai_mock", json={"AAI_MOCK": settings.AAI_URL}) + + customer = Customer.create(global_customer_id="test_global_customer_id", + subscriber_name="test_subscriber_name", + subscriber_type="test_subscriber_type") + service = Service("test_service") + service.unique_uuid = str(uuid4()) + service.identifier = str(uuid4()) + service.name = str(uuid4()) + customer.subscribe_service("service_type") + service_subscription = customer.get_service_subscription_by_service_type("service_type") + cloud_region = CloudRegion.create( + "test_owner", "test_cloud_region", orchestration_disabled=True, in_maint=False + ) + cloud_region.add_tenant( + tenant_id="test_tenant_name", tenant_name="test_tenant_name", tenant_context="test_tenant_context" + ) + tenant = cloud_region.get_tenant(tenant_id="test_tenant_name") + service_subscription.link_to_cloud_region_and_tenant(cloud_region=cloud_region, tenant=tenant) + owning_entity = "test_owning_entity" + project = "test_project" + + # Service instantiation + service._distributed = True + assert len(list(service_subscription.service_instances)) == 0 + service_instantiation_request = ServiceInstantiation.instantiate_ala_carte( + service, + cloud_region, + tenant, + customer, + owning_entity, + project, + service_subscription + ) + assert service_instantiation_request.status == ServiceInstantiation.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert service_instantiation_request.status == ServiceInstantiation.StatusEnum.COMPLETED + assert len(list(service_subscription.service_instances)) == 1 + + # Vnf instantiation + service_instance = next(service_subscription.service_instances) + assert len(list(service_instance.vnf_instances)) == 0 + owning_entity = "test_owning_entity" + project = "test_project" + vnf = MagicMock() + line_of_business = "test_line_of_business" + platform = "test_platform" + with pytest.raises(StatusError): + service_instance.add_vnf( + vnf, + line_of_business, + platform + ) + service_instance.orchestration_status = "Active" + service_instance._sdc_service = service + with patch.object(ServiceInstance, "sdc_service", return_value=service): + vnf_instantiation_request = service_instance.add_vnf( + vnf, + line_of_business, + platform + ) + assert vnf_instantiation_request.status == VnfInstantiation.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert vnf_instantiation_request.status == VnfInstantiation.StatusEnum.COMPLETED + assert len(list(service_instance.vnf_instances)) == 1 + # VfModule instantiation + vnf_instance = next(service_instance.vnf_instances) + assert len(list(vnf_instance.vf_modules)) == 0 + vnf.model_version_id = vnf_instance.model_version_id + vf_module = MagicMock() + + with patch.object(ServiceInstance, "sdc_service", return_value=service) as service_mock: + service_mock.vnfs = [vnf] + vf_module_instantiation_request = vnf_instance.add_vf_module( + vf_module + ) + assert vf_module_instantiation_request.status == VfModuleInstantiation.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert vf_module_instantiation_request.status == VfModuleInstantiation.StatusEnum.COMPLETED + assert len(list(vnf_instance.vf_modules)) == 1 + + # Cleanup + vf_module_instance = next(vnf_instance.vf_modules) + vf_module_deletion_request = vf_module_instance.delete() + assert vf_module_deletion_request.status == VfModuleDeletionRequest.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert vf_module_deletion_request.status == VfModuleDeletionRequest.StatusEnum.COMPLETED + assert len(list(vnf_instance.vf_modules)) == 0 + + vnf_deletion_request = vnf_instance.delete() + assert vnf_deletion_request.status == VnfDeletionRequest.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert vnf_deletion_request.status == VnfDeletionRequest.StatusEnum.COMPLETED + assert len(list(service_instance.vnf_instances)) == 0 + + with patch.object(ServiceInstance, "sdc_service", return_value=service) as service_mock: + service_deletion_request = service_instance.delete() + assert service_deletion_request.status == ServiceDeletionRequest.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert service_deletion_request.status == ServiceDeletionRequest.StatusEnum.COMPLETED + assert len(list(service_subscription.service_instances)) == 0 + + +@pytest.mark.integration +def test_a_la_carte_vl_instantiation(): + requests.get(f"{ServiceInstantiation.base_url}/reset") + requests.get(f"{Customer.base_url}/reset") + requests.post(f"{ServiceInstantiation.base_url}/set_aai_mock", json={"AAI_MOCK": settings.AAI_URL}) + + customer = Customer.create(global_customer_id="test_global_customer_id", + subscriber_name="test_subscriber_name", + subscriber_type="test_subscriber_type") + service = Service("test_service") + service.unique_uuid = str(uuid4()) + service.identifier = str(uuid4()) + service.name = str(uuid4()) + customer.subscribe_service("service_type") + service_subscription = customer.get_service_subscription_by_service_type("service_type") + cloud_region = CloudRegion.create( + "test_owner", "test_cloud_region", orchestration_disabled=True, in_maint=False + ) + cloud_region.add_tenant( + tenant_id="test_tenant_name", tenant_name="test_tenant_name", tenant_context="test_tenant_context" + ) + tenant = cloud_region.get_tenant(tenant_id="test_tenant_name") + service_subscription.link_to_cloud_region_and_tenant(cloud_region=cloud_region, tenant=tenant) + owning_entity = "test_owning_entity" + project = "test_project" + + # Service instantiation + service._distributed = True + assert len(list(service_subscription.service_instances)) == 0 + service_instantiation_request = ServiceInstantiation.instantiate_ala_carte( + service, + cloud_region, + tenant, + customer, + owning_entity, + project, + service_subscription + ) + assert service_instantiation_request.status == ServiceInstantiation.StatusEnum.IN_PROGRESS + service_instantiation_request.wait_for_finish() + assert service_instantiation_request.status == ServiceInstantiation.StatusEnum.COMPLETED + assert len(list(service_subscription.service_instances)) == 1 + + # Network instantiation + service_instance = next(service_subscription.service_instances) + assert len(list(service_instance.network_instances)) == 0 + owning_entity = "test_owning_entity" + project = "test_project" + network = MagicMock() + line_of_business = "test_line_of_business" + platform = "test_platform" + with pytest.raises(AttributeError): + service_instance.network( + network, + line_of_business, + platform + ) + service_instance.orchestration_status = "Active" + with patch.object(ServiceInstance, "sdc_service", return_value=service): + network_instantiation_request = service_instance.add_network( + network, + line_of_business, + platform + ) + assert network_instantiation_request.status == VnfInstantiation.StatusEnum.IN_PROGRESS + network_instantiation_request.wait_for_finish() + assert network_instantiation_request.status == VnfInstantiation.StatusEnum.COMPLETED + assert len(list(service_instance.network_instances)) == 1 + + +@patch.object(Service, "vnfs", new_callable=PropertyMock) +@patch.object(Service, "components", new_callable=PropertyMock) +@pytest.mark.integration +def test_instantiate_macro(mock_service_components, mock_service_vnfs): + requests.get(f"{ServiceInstantiation.base_url}/reset") + requests.get(f"{Customer.base_url}/reset") + requests.post(f"{ServiceInstantiation.base_url}/set_aai_mock", json={"AAI_MOCK": settings.AAI_URL}) + + customer = Customer.create(global_customer_id="test_global_customer_id", + subscriber_name="test_subscriber_name", + subscriber_type="test_subscriber_type") + service = Service("test_service") + service._tosca_template = "n/a" + + mock_service_vnfs.return_value = [ + Vnf( + name="test_vnf", + node_template_type="vf", + model_name= "test_vnf_model", + model_version_id = str(uuid4()), + model_invariant_id=str(uuid4()), + model_version="1.0", + model_customization_id=str(uuid4()), + model_instance_name=str(uuid4()), + component=MagicMock(), + vf_modules=[ + VfModule( + name="TestVnfModel..base..module-0", + group_type="vf-module", + model_name="TestVnfModel..base..module-0", + model_version_id=str(uuid4()), + model_invariant_uuid=str(uuid4()), + model_version="1", + model_customization_id=str(uuid4()), + properties=None + ) + ] + ) + ] + service.unique_uuid = str(uuid4()) + service.identifier = str(uuid4()) + service.name = str(uuid4()) + customer.subscribe_service("service_type") + service_subscription = customer.get_service_subscription_by_service_type("service_type") + cloud_region = CloudRegion.create( + "test_owner", "test_cloud_region", orchestration_disabled=True, in_maint=False + ) + cloud_region.add_tenant( + tenant_id="test_tenant_name", tenant_name="test_tenant_name", tenant_context="test_tenant_context" + ) + tenant = cloud_region.get_tenant(tenant_id="test_tenant_name") + service_subscription.link_to_cloud_region_and_tenant(cloud_region=cloud_region, tenant=tenant) + owning_entity = "test_owning_entity" + project = "test_project" + line_of_business = "test_line_of_business" + platform = "test_platform" + + vfm_instance_params = [ + InstantiationParameter(name="vfm_param", value="vfm_param_value"), + + ] + vfm_params = VfmoduleParameters("base", vfm_instance_params) + + vnf_instance_params = [ + InstantiationParameter(name="vnf_param", value="vnf_param_value") + ] + + vnf_params = VnfParameters("test_vnf_model", vnf_instance_params, [vfm_params]) + + # Service instantiation + service._distributed = True + assert len(list(service_subscription.service_instances)) == 0 + service_instantiation_request = ServiceInstantiation.instantiate_macro( + sdc_service=service, + customer=customer, + owning_entity=owning_entity, + project=project, + line_of_business=line_of_business, + platform=platform, + cloud_region=cloud_region, + tenant=tenant, + vnf_parameters=[vnf_params], + service_subscription=service_subscription + ) + assert service_instantiation_request.status == ServiceInstantiation.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert service_instantiation_request.status == ServiceInstantiation.StatusEnum.COMPLETED + assert len(list(service_subscription.service_instances)) == 1 + service_instance = next(service_subscription.service_instances) + + # Cleanup + with patch.object(ServiceInstance, "sdc_service", return_value=service) as service_mock: + service_deletion_request = service_instance.delete() + assert service_deletion_request.status == ServiceDeletionRequest.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert service_deletion_request.status == ServiceDeletionRequest.StatusEnum.COMPLETED + assert len(list(service_subscription.service_instances)) == 0 + +@patch.object(Service, "vnfs", new_callable=PropertyMock) +@patch.object(Service, "components", new_callable=PropertyMock) +@pytest.mark.integration +def test_instantiate_macro_multiple_vnf(mock_service_components, mock_service_vnfs): + requests.get(f"{ServiceInstantiation.base_url}/reset") + requests.get(f"{Customer.base_url}/reset") + requests.post(f"{ServiceInstantiation.base_url}/set_aai_mock", json={"AAI_MOCK": settings.AAI_URL}) + + customer = Customer.create(global_customer_id="test_global_customer_id", + subscriber_name="test_subscriber_name", + subscriber_type="test_subscriber_type") + service = Service("test_service") + service._tosca_template = "n/a" + + mock_service_vnfs.return_value = [ + Vnf( + name="test_vnf", + node_template_type="vf", + model_name= "test_vnf_model", + model_version_id = str(uuid4()), + model_invariant_id=str(uuid4()), + model_version="1.0", + model_customization_id=str(uuid4()), + model_instance_name=str(uuid4()), + component=MagicMock(), + vf_modules=[ + VfModule( + name="TestVnfModel..base..module-0", + group_type="vf-module", + model_name="TestVnfModel..base..module-0", + model_version_id=str(uuid4()), + model_invariant_uuid=str(uuid4()), + model_version="1", + model_customization_id=str(uuid4()), + properties=None + ) + ] + ) + ] + service.unique_uuid = str(uuid4()) + service.identifier = str(uuid4()) + service.name = str(uuid4()) + customer.subscribe_service("service_type") + service_subscription = customer.get_service_subscription_by_service_type("service_type") + cloud_region = CloudRegion.create( + "test_owner", "test_cloud_region", orchestration_disabled=True, in_maint=False + ) + cloud_region.add_tenant( + tenant_id="test_tenant_name", tenant_name="test_tenant_name", tenant_context="test_tenant_context" + ) + tenant = cloud_region.get_tenant(tenant_id="test_tenant_name") + service_subscription.link_to_cloud_region_and_tenant(cloud_region=cloud_region, tenant=tenant) + owning_entity = "test_owning_entity" + project = "test_project" + line_of_business = "test_line_of_business" + platform = "test_platform" + + so_service = SoService.load({ + "subscription_service_type": "service_type", + "vnfs": [ + { + "model_name": "test_vnf_model", + "instance_name": "vnf0", + "parameters": { + "param1": "value1" + }, + "vf_modules": [ + { + "instance_name": "vnf0_vfm0", + "model_name": "base", + "parameters": { + "vfm_param1": "vfm_value1" + } + } + ] + }, + { + "model_name": "test_vnf_model", + "instance_name": "vnf1", + "parameters": { + "param2": "value2" + }, + "vf_modules": [ + { + "instance_name": "vnf1_vfm0", + "model_name": "base", + "parameters": { + "vfm_param2": "vfm_value2" + } + } + ] + } + ] + }) + + # Service instantiation + service._distributed = True + assert len(list(service_subscription.service_instances)) == 0 + service_instantiation_request = ServiceInstantiation.instantiate_macro( + sdc_service=service, + customer=customer, + owning_entity=owning_entity, + project=project, + line_of_business=line_of_business, + platform=platform, + cloud_region=cloud_region, + tenant=tenant, + so_service=so_service + ) + assert service_instantiation_request.status == ServiceInstantiation.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert service_instantiation_request.status == ServiceInstantiation.StatusEnum.COMPLETED + assert len(list(service_subscription.service_instances)) == 1 + service_instance = next(service_subscription.service_instances) + + # Cleanup + with patch.object(ServiceInstance, "sdc_service", return_value=service) as service_mock: + service_deletion_request = service_instance.delete() + assert service_deletion_request.status == ServiceDeletionRequest.StatusEnum.IN_PROGRESS + time.sleep(2) # After 1 second mocked server changed request status to complete + assert service_deletion_request.status == ServiceDeletionRequest.StatusEnum.COMPLETED + assert len(list(service_subscription.service_instances)) == 0 |