path: root/integration_tests/test_07_instantiation.py

#   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