path: root/test/vcpe_tosca/local/vcpe_tosca_test.py

#!/usr/bin/python

# Prerequisites for machine to run this
# Put in required parameters in vcpe_config.json
# Install python-pip (apt install python-pip)
# Install requests
# Install ONAP CLI
# Must have connectivity to the ONAP, openstack is already configured.
# Configuration File, the parameters will be modified according to the Lab env
# Put in vnf and ns CSAR file under csar folder

import json
import os
import uuid
import requests
import unittest
import time
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class VcpeToscaTest(unittest.TestCase):
    def setUp(self):
        file_path = os.path.dirname(os.path.abspath(__file__))
        with open(file_path + "/config/vcpe_config.json", encoding='utf-8') as self.config_file:
            self.config_params = self.get_parameters()
        self.aai_header = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            'X-TransactionId': "9999",
            'Real-Time': "true",
            'X-FromAppId': "jimmy-postman",
            "Authorization": "Basic QUFJOkFBSQ=="
        }
        self.base_url = self.config_params["msb_url"]
        print("Set cli command environment--beginning")
        os.environ["OPEN_CLI_PRODUCT_IN_USE"] = self.config_params["open_cli_product"]
        os.environ["OPEN_CLI_HOME"] = self.config_params["open_cli_home"]
        print("Set cli command environment--successful")

        self.complex_version = None
        self.cloud_version = None
        self.service_type_version = None
        self.customer_version = None
        self.tenant_id = None
        self.subscription_version = None
        self.esr_vnfm_version = self.esr_vnfm_id = None
        self.ns_instance_id = None
        self.ns_package_id = None
        self.vnf_package_list = []

        print("Create cloud complex--beginning")
        self.create_complex()
        print("Create cloud complex--successful")

        print("Register all clouds--beginning")
        self.register_all_clouds()
        print("Register all clouds--successful")
        time.sleep(30)

        print("Create vCPE service")
        self.create_service_type()

        print("Create customer")
        self.create_customer()

        print("Get tenant id")
        self.get_tenant_id()

        print("Add customer and subscription")
        self.add_customer_subscription()

        print("Register vnfm")
        self.register_vnfm()

    def tearDown(self):
        if self.ns_instance_id:
            self.terminateNs()
            self.deleteNs()

        if self.ns_package_id:
            self.delete_ns_package()

        if self.vnf_package_list:
            self.delete_vnf_package()

        if self.esr_vnfm_id and self.esr_vnfm_version:
            self.unregister_vnfm()

        if self.subscription_version:
            print("Remove service subscription")
            self.remove_customer_subscription()

        if self.customer_version:
            print("Remove customer %s" % self.config_params["customer_name"])
            self.delete_customer()

        if self.service_type_version:
            print("Remove service type %s" % self.config_params["service_name"])
            self.delete_service_type()

        if self.cloud_version:
            print("Remove cloud %s" % self.config_params["cloud-owner"])
            self.delete_cloud_helper()

        if self.complex_version:
            self.get_complex_resource_version()
            print("Remove complex %s" % self.config_params["complex_name"])
            self.delete_complex()

    def get_parameters(self):
        parameters = json.load(self.config_file)
        return parameters

    @staticmethod
    def get_out_helper_2(in_string):
        out_list = ((in_string.replace('|', '')).replace('+', '')).split()
        return out_list

    def create_complex(self):
        complex_create_string = "oclip complex-create -j {} -r {} -x {} -y {} -lt {} -l {} -i {} -lo {} \
                             -S {} -la {} -g {} -w {} -z {} -k {} -o {} -q {} -m {} -u {} -p {}".format(
            self.config_params["street2"], self.config_params["physical_location"],
            self.config_params["complex_name"], self.config_params["data_center_code"],
            self.config_params["latitude"], self.config_params["region"],
            self.config_params["street1"], self.config_params["longitude"],
            self.config_params["state"], self.config_params["lata"],
            self.config_params["city"], self.config_params["postal-code"],
            self.config_params["complex_name"], self.config_params["country"],
            self.config_params["elevation"], self.config_params["identity_url"],
            self.config_params["aai_url"], self.config_params["aai_username"],
            self.config_params["aai_password"])
        os.system(complex_create_string)

        self.get_complex_resource_version()

    def get_complex_resource_version(self):
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        complex_url = self.base_url + "/aai/v11/cloud-infrastructure/complexes"
        complex_list_response = requests.get(url=complex_url, headers=self.aai_header, verify=False)
        if complex_list_response.status_code == 200:
            for complex in (complex_list_response.json())["complex"]:
                if complex['physical-location-id'] == self.config_params["complex_name"]:
                    self.complex_version = complex['resource-version']
                    print("Complex %s resource-version is %s."
                          % (self.config_params["complex_name"], self.complex_version))

    def delete_complex(self):
        complex_delete_string = 'oclip complex-delete -x {} -y {} -m {} -u {} -p {}'.format(
            self.config_params["complex_name"], self.complex_version, self.config_params["aai_url"],
            self.config_params["aai_username"], self.config_params["aai_password"])
        os.system(complex_delete_string)
        print("Delete complex--successful")
        self.complex_version = None

    def register_cloud_helper(self, cloud_region, values):
        print("Create Cloud--beginning")
        cloud_create_string = 'oclip cloud-create -e {} -b {} ' \
                              '-x {} -y {} -j {} -w {} -l {} -url {} -n {} -q {} -r {} -Q {} -i {} -g {} \
                              -z {} -k {} -c {} -m {} -u {} -p {}' \
            .format(values.get("esr-system-info-id"), values.get("user-name"),
                    self.config_params["cloud-owner"],
                    cloud_region, values.get("password"),
                    values.get("cloud-region-version"), values.get("default-tenant"),
                    values.get("service-url"), self.config_params["complex_name"],
                    values.get("cloud-type"), self.config_params["owner-defined-type"],
                    values.get("system-type"), values.get("identity-url"),
                    self.config_params["cloud-zone"], values.get("ssl-insecure"),
                    values.get("system-status"), values.get("cloud-domain"),
                    self.config_params["aai_url"],
                    self.config_params["aai_username"],
                    self.config_params["aai_password"])

        os.system(cloud_create_string)
        print("Create Cloud--successful")

        print("Associate Cloud with complex--beginning")
        complex_associate_string = "oclip complex-associate -x {} -y {} -z {} -m {} -u {} -p {}".format(
            self.config_params["complex_name"],
            cloud_region, self.config_params["cloud-owner"], self.config_params["aai_url"],
            self.config_params["aai_username"],
            self.config_params["aai_password"])
        os.system(complex_associate_string)
        print("Associate Cloud with complex--successful")

        print("Register Cloud with Multicloud--beginning")
        multicloud_register_string = "oclip multicloud-register-cloud -y {} -x {} -m {}".format(
            self.config_params["cloud-owner"], cloud_region, self.config_params["multicloud_url"])
        os.system(multicloud_register_string)
        print("Register Cloud with Multicloud--successful")

        cloud_url = self.base_url + "/aai/v11/cloud-infrastructure/cloud-regions"
        cloud_list_response = requests.get(url=cloud_url, headers=self.aai_header, verify=False)
        if cloud_list_response.status_code == 200:
            for cloud in (cloud_list_response.json())["cloud-region"]:
                if cloud['cloud-owner'] == self.config_params["cloud-owner"]:
                    self.cloud_version = cloud['resource-version']
                    print("Cloud %s resource-version is %s."
                          % (self.config_params["cloud-owner"], self.cloud_version))

    def register_all_clouds(self):
        cloud_dictionary = self.config_params["cloud_region_data"]
        for cloud_region, cloud_region_values in cloud_dictionary.items():
            self.register_cloud_helper(cloud_region, cloud_region_values)

    def delete_cloud_helper(self):
        print("Multicloud-cloud-delete--beginning")
        cloud_region = list(self.config_params["cloud_region_data"].keys())[0]
        header = {'content-type': 'application/json', 'accept': 'application/json'}
        multicloud_url = self.base_url + "/api/multicloud-titaniumcloud/v1/{}/{}" \
            .format(self.config_params["cloud-owner"], cloud_region)
        requests.delete(url=multicloud_url, headers=header, verify=False)
        cloud_url = self.base_url + "/aai/v11/cloud-infrastructure/cloud-regions"
        n = 60
        while n > 0:
            cloud_flag = False
            cloud_list_response = requests.get(url=cloud_url, headers=self.aai_header, verify=False)
            n = n - 1
            if cloud_list_response.status_code == 200:
                for cloud in (cloud_list_response.json()).get("cloud-region"):
                    if cloud['cloud-owner'] == self.config_params["cloud-owner"]:
                        cloud_flag = True
                        break
                if not cloud_flag:
                    break
                else:
                    time.sleep(1)
        print("Multicloud-cloud-delete----successful")
        self.cloud_version = None

    def create_service_type(self):
        create_string = "oclip service-type-create -x {} -y {} -m {} -u {} -p {}".format(
            self.config_params["service_name"], self.config_params["service_name"], self.config_params["aai_url"],
            self.config_params["aai_username"], self.config_params["aai_password"])
        os.system(create_string)

        service_tpe_list_url = self.base_url + "/aai/v11/service-design-and-creation/services"
        service_type_list_response = requests.get(url=service_tpe_list_url, headers=self.aai_header, verify=False)
        if service_type_list_response.status_code == 200:
            for service in (service_type_list_response.json())["service"]:
                if service["service-id"] == self.config_params["service_name"]:
                    self.service_type_version = service['resource-version']
                    print("Service type %s resource-version is %s."
                          % (self.config_params["service_name"], self.service_type_version))

    def delete_service_type(self):
        print("delete service type--beginning")
        service_delete_string = 'oclip service-type-delete -x {} -y {} -m {} -u {} -p {}'.format(
            self.config_params["service_name"], self.service_type_version, self.config_params["aai_url"],
            self.config_params["aai_username"], self.config_params["aai_password"])
        os.system(service_delete_string)
        print("delete service type--successful")
        self.service_type_version = None

    def create_customer(self):
        create_string = "oclip customer-create -x {} -y {} -m {} -u {} -p {}".format(
            self.config_params["customer_name"],
            self.config_params["subscriber_name"],
            self.config_params["aai_url"],
            self.config_params["aai_username"],
            self.config_params["aai_password"])
        os.system(create_string)

        customer_list_url = self.base_url + "/aai/v11/business/customers"
        customer_list_response = requests.get(url=customer_list_url, headers=self.aai_header, verify=False)
        if customer_list_response.status_code == 200:
            for cutsomer in (customer_list_response.json())["customer"]:
                if cutsomer['global-customer-id'] == self.config_params["customer_name"]:
                    self.customer_version = cutsomer['resource-version']
                    print("Customer %s resource-version is %s."
                          % (self.config_params["customer_name"], self.customer_version))

    def delete_customer(self):
        print("delete customer--beginning")
        customer_delete_string = 'oclip customer-delete -x {} -y {} -m {} -u {} -p {}'.format(
            self.config_params["customer_name"], self.customer_version, self.config_params["aai_url"],
            self.config_params["aai_username"], self.config_params["aai_password"])
        os.system(customer_delete_string)
        print("delete customer--successful")
        self.customer_version = None

    def get_tenant_id(self):
        print("Get tenant id--beginning")
        cloud_dictionary = self.config_params["cloud_region_data"]
        cloud_region = list(self.config_params["cloud_region_data"].keys())[0]

        tenant_list_url = self.base_url + "/aai/v11/cloud-infrastructure/cloud-regions/cloud-region/{}/{}/tenants" \
            .format(self.config_params["cloud-owner"], cloud_region)

        for cloud_region, cloud_region_values in cloud_dictionary.items():
            tenant_name = cloud_region_values.get("default-tenant")
        tenant_list_response = requests.get(url=tenant_list_url, headers=self.aai_header, verify=False)
        if tenant_list_response.status_code == 200:
            for tenant in (tenant_list_response.json())["tenant"]:
                if tenant['tenant-name'] == tenant_name:
                    self.tenant_id = tenant['tenant-id']
                    print("Tenant id is %s ." % self.tenant_id)

    def add_customer_subscription(self):
        subscription_check = 0
        for cloud_region, cloud_region_values in (self.config_params["cloud_region_data"]).items():
            if subscription_check == 0:
                subscription_string = "oclip subscription-create -x {} -c {} -z {} -e {} " \
                                      "-y {} -r {} -m {} -u {} -p {}" \
                    .format(self.config_params["customer_name"],
                            self.tenant_id,
                            self.config_params["cloud-owner"],
                            self.config_params["service_name"],
                            cloud_region_values.get("default-tenant"),
                            cloud_region, self.config_params["aai_url"],
                            self.config_params["aai_username"],
                            self.config_params["aai_password"])
            else:
                subscription_string = "oclip subscription-cloud-add -x {} -c {} " \
                                      "-z {} -e {} -y {} -r {} -m {} -u {} -p {}" \
                    .format(self.config_params["customer_name"], self.tenant_id,
                            self.config_params["cloud-owner"], self.config_params["service_name"],
                            cloud_region_values.get("default-tenant"), cloud_region,
                            self.config_params["aai_url"],
                            self.config_params["aai_username"],
                            self.config_params["aai_password"])
            os.system(subscription_string)
            subscription_check += 1

        subscription_url = self.base_url + "/aai/v11/business/customers/customer/{}" \
                                           "/service-subscriptions/service-subscription/{}" \
            .format(self.config_params["customer_name"], self.config_params["service_name"])
        resp = requests.get(url=subscription_url, headers=self.aai_header, verify=False)
        if resp.status_code == 200:
            self.subscription_version = resp.json()['resource-version']
            print("Subscription resource-version is %s." % self.subscription_version)

    def remove_customer_subscription(self):
        print("Remove subscription--beginning")
        subscription_delete_string = 'oclip subscription-delete -x {} -y {} -g {} -m {} -u {} -p {}'.format(
            self.config_params["customer_name"], self.config_params["service_name"], self.subscription_version,
            self.config_params["aai_url"],
            self.config_params["aai_username"], self.config_params["aai_password"])
        os.system(subscription_delete_string)
        print("Delete subscription--successful")

    def register_vnfm_helper(self, vnfm_key, values):
        print("Create vnfm--beginning")
        self.esr_vnfm_id = str(uuid.uuid4())
        vnfm_create_string = 'oclip vnfm-create -b {} -c {} -e {} -v {} -g {} -x {} ' \
                             '-y {} -i {} -j {} -q {} -m {} -u {} -p {}' \
            .format(vnfm_key, values.get("type"), values.get("vendor"),
                    values.get("version"), values.get("url"), values.get("vim-id"),
                    self.esr_vnfm_id, values.get("user-name"), values.get("user-password"),
                    values.get("vnfm-version"), self.config_params["aai_url"],
                    self.config_params["aai_username"], self.config_params["aai_password"])

        os.system(vnfm_create_string)
        print("Create vnfm--successful")

        vnfm_url = self.base_url + "/aai/v11/external-system/esr-vnfm-list"
        resp = requests.get(url=vnfm_url, headers=self.aai_header, verify=False)
        if resp.status_code == 200:
            for vnfm in (resp.json())["esr-vnfm"]:
                if vnfm['vnfm-id'] == self.esr_vnfm_id:
                    self.esr_vnfm_version = vnfm['resource-version']
                    print("Vnfm %s resource-version is %s."
                          % (self.esr_vnfm_id, self.esr_vnfm_version))

    def register_vnfm(self):
        vnfm_params = self.config_params["vnfm_params"]
        for vnfm_key, vnfm_values in vnfm_params.items():
            self.register_vnfm_helper(vnfm_key, vnfm_values)

    def unregister_vnfm(self):
        print("Delete vnfm %s" % self.esr_vnfm_id)
        print("Delete vnfm--beginning")
        vnfm_delete_string = 'oclip vnfm-delete -x {} -y {} -m {} -u {} -p {}'.format(
            self.esr_vnfm_id, self.esr_vnfm_version, self.config_params["aai_url"],
            self.config_params["aai_username"], self.config_params["aai_password"])
        os.system(vnfm_delete_string)
        self.esr_vnfm_version = self.esr_vnfm_id = None
        print("Delete vnfm--successful")

    def create_ns(self):
        ns = self.config_params["ns"]
        data = {
            "context": {
                "globalCustomerId": self.config_params["customer_name"],
                "serviceType": self.config_params["service_name"]
            },
            "csarId": self.ns_package_id,
            "nsName": ns.get("name"),
            "description": "description"
        }
        ns_header = {'content-type': 'application/json', 'accept': 'application/json'}
        ns_url = self.base_url + "/api/nslcm/v1/ns"
        ns_resp = requests.post(ns_url, data=json.dumps(data), headers=ns_header, verify=False)
        if 201 == ns_resp.status_code:
            ns_instance_id = ns_resp.json().get("nsInstanceId")
            print("create ns successfully, the ns instance id is %s" % ns_instance_id)
            return ns_instance_id
        else:
            raise Exception("Create ns failed.")

    def instantiate_ns(self):
        print("Instantiate ns beginning")
        constraints = [
            {
                "vnfProfileId": x,
                "locationConstraints": {
                    "vimId": self.config_params["location"]
                }
            } for x in self.vnfdId_list]
        data = {
            "additionalParamForNs": {
                "sdnControllerId": self.config_params["sdc-controller-id"]
            },
            "locationConstraints": constraints
        }

        header = {'content-type': 'application/json', 'accept': 'application/json'}
        instance_url = self.base_url + "/api/nslcm/v1/ns/" + self.ns_instance_id + "/instantiate"
        instance_resp = requests.post(instance_url, data=json.dumps(data), headers=header, verify=False)
        if 200 == instance_resp.status_code:
            ns_instance_jod_id = instance_resp.json().get("jobId")
            print("Instantiate ns successfully, the job id is %s" % ns_instance_jod_id)
            return ns_instance_jod_id
        else:
            raise Exception("Instantiate ns failed.")

    def create_ns_package(self):
        print("Create ns package is beginning")
        ns = self.config_params["ns"]
        ns_url = self.base_url + "/api/nsd/v1/ns_descriptors"
        ns_headers = {'content-type': 'application/json', 'accept': 'application/json'}
        ns_data = {'userDefinedData': {ns.get("key"): ns.get("value")}}
        ns_package_reps = requests.post(ns_url, data=json.dumps(ns_data), headers=ns_headers, verify=False)
        if 201 == ns_package_reps.status_code:
            print("Create ns package successful, the ns package id is %s"
                  % (ns_package_reps.json()["id"]))
            return ns_package_reps.json()["id"]
        else:
            raise Exception("Create ns package failed.")

    def delete_ns_package(self):
        print("Delete ns package %s is beginning" % self.ns_package_id)
        vnf_url = self.base_url + "/api/nsd/v1/ns_descriptors/%s" % self.ns_package_id
        resp = requests.delete(url=vnf_url, verify=False)
        if 204 == resp.status_code:
            print("Delete ns package %s successfully." % self.ns_package_id)
            self.ns_package_id = None
        else:
            print("Delete ns package %s failed." % self.ns_package_id)

    def create_upload_vnf_package(self):
        print("Create vnf package is beginning")
        vnfs = self.config_params["vnfs"]
        vnf_url = self.base_url + "/api/vnfpkgm/v1/vnf_packages"
        header = {'content-type': 'application/json', 'accept': 'application/json'}
        for vnf_values in vnfs.values():
            vnf_data = {'userDefinedData': {vnf_values.get("key"): vnf_values.get("value")}}
            vnf_package_reps = requests.post(vnf_url, data=json.dumps(vnf_data), headers=header, verify=False)
            if 201 == vnf_package_reps.status_code:
                print("Create vnf package successful, the vnf package id is %s"
                      % (vnf_package_reps.json()["id"]))
                package_id = vnf_package_reps.json()["id"]
                self.vnf_package_list.append(package_id)
                vnf_upload_url = '{}/api/vnfpkgm/v1/vnf_packages/{}/package_content' \
                    .format(self.config_params["vfc-url"], package_id)
                file_path = os.path.dirname(os.path.abspath(__file__))
                csar_file = file_path + "/" + vnf_values.get("path")
                with open(csar_file, 'rb') as vnf_file:
                    for i in range(10):
                        resp = requests.put(vnf_upload_url, files={'file': vnf_file}, verify=False)
                        if 202 == resp.status_code:
                            break
                        if 500 == resp.status_code:
                            raise Exception("Upload vnf package failed. %s" % resp.json())
                        else:
                            time.sleep(i)
            else:
                print("Create vnf package failed.")

    def delete_vnf_package(self):
        print("Delete vnf package is beginning")
        for vnf_package_id in self.vnf_package_list:
            vnf_url = self.base_url + "/api/vnfpkgm/v1/vnf_packages/%s" % vnf_package_id
            resp = requests.delete(url=vnf_url, verify=False)
            if 204 == resp.status_code:
                print("Delete vnf package %s successfully." % vnf_package_id)
            else:
                print("Delete vnf package %s failed." % vnf_package_id)
        self.vnf_package_list = []

    def upload_ns_package(self):
        ns = self.config_params["ns"]
        ns_upload_url = '{}/api/nsd/v1/ns_descriptors/{}/nsd_content'.format(self.config_params["vfc-url"],
                                                                             self.ns_package_id)
        file_path = os.path.dirname(os.path.abspath(__file__))
        ns_file_path = file_path + "/" + ns["path"]
        with open(ns_file_path, 'rb') as ns_file:
            for i in range(10):
                resp = requests.put(ns_upload_url, files={'file': ns_file}, verify=False)
                if 204 == resp.status_code:
                    break
                if 500 == resp.status_code:
                    raise Exception("Upload ns package failed.")
                else:
                    time.sleep(i)

    def get_vnf_package(self):
        vnfdid_list = []
        for vnf_package_id in self.vnf_package_list:
            n = 60
            while n > 0:
                vnf_package_url = self.base_url + '/api/vnfpkgm/v1/vnf_packages/%s' % vnf_package_id
                vnf_resp = requests.get(vnf_package_url, verify=False)
                n = n - 1
                if 200 == vnf_resp.status_code:
                    vnfdId = vnf_resp.json().get("vnfdId")
                    if vnfdId is None:
                        time.sleep(1)
                    else:
                        print("vnfdId is %s" % vnfdId)
                        vnfdid_list.append(vnfdId)
                        break
        return vnfdid_list

    def getVnf(self, vnfs):
        vnf_list = []
        for vnf in vnfs:
            if 'relationship-list' in vnf:
                for relation in vnf["relationship-list"]["relationship"]:
                    if "service-instance" == relation["related-to"]:
                        if self.ns_instance_id in relation["related-link"]:
                            vnf_list.append(vnf)
        return vnf_list

    @staticmethod
    def findVserver(vnf_list):
        vserver_list = []
        for vnf in vnf_list:
            if 'relationship-list' in vnf:
                for relation in vnf["relationship-list"]["relationship"]:
                    if "vserver" == relation["related-to"]:
                        for relationData in relation["relationship-data"]:
                            if "vserver.vserver-id" == relationData["relationship-key"]:
                                vserver_list.append(relationData["relationship-value"])
        return vserver_list

    def waitProcessFinished(self, job_id, action):
        print("Wait for the %s ns finished." % action)
        job_url = self.base_url + "/api/nslcm/v1/jobs/%s" % job_id
        progress = 0
        n = 6000
        while n > 0:
            job_resp = requests.get(url=job_url, verify=False)
            n = n - 1
            if 200 == job_resp.status_code:
                if "responseDescriptor" in job_resp.json():
                    progress_rep = (job_resp.json())["responseDescriptor"]["progress"]
                    if 100 != progress_rep:
                        if 255 == progress_rep:
                            print("Ns %s %s failed." % (self.ns_instance_id, action))
                            raise Exception("%s ns failed." % action)
                        elif progress_rep != progress:
                            progress = progress_rep
                            print("Ns %s %s process is %s." % (self.ns_instance_id, action, progress))
                        time.sleep(0.2)
                    else:
                        print("Ns %s %s process is %s." % (self.ns_instance_id, action, progress_rep))
                        print("Ns %s %s successfully." % (self.ns_instance_id, action))
                        break

    def terminateNs(self):
        print("Terminate ns--beginning")
        ns_url = self.base_url + "/api/nslcm/v1/ns/%s" % self.ns_instance_id
        d = {
            "gracefulTerminationTimeout": 600,
            "terminationType": "FORCEFUL"
        }
        try:
            res = requests.post(url=ns_url + "/terminate", data=d, verify=False)
            if 202 == res.status_code:
                terminate_ns_job_id = res.json()["jobId"]
                print("Terminate job is %s" % terminate_ns_job_id)
            else:
                raise Exception("Instantiate ns failed.")
            self.waitProcessFinished(terminate_ns_job_id, "terminate")
        except Exception as e:
            print(e.args[0])

    def deleteNs(self):
        print("Delete ns %s --beginning" % self.ns_instance_id)
        ns_url = self.base_url + "/api/nslcm/v1/ns/%s" % self.ns_instance_id
        res = requests.delete(ns_url, verify=False)
        if 204 == res.status_code:
            print("Ns %s delete successfully." % self.ns_instance_id)
            self.ns_instance_id = None

    def testNs(self):
        print("Use csar file is uploaded by local")
        try:
            self.create_upload_vnf_package()
            self.ns_package_id = self.create_ns_package()
            print("Get vnfdId list.")
            self.vnfdId_list = self.get_vnf_package()
            if len(self.vnfdId_list) < 5:
                raise Exception("Upload vnf package failed. "
                                "Please check vnf package(b1bb0ce7-1111-4fa7-95ed-4840d70a1177, "
                                "b1bb0ce7-2222-4fa7-95ed-4840d70a1177, "
                                "b1bb0ce7-3333-4fa7-95ed-4840d70a1177, "
                                "b1bb0ce7-4444-4fa7-95ed-4840d70a1177, "
                                "b1bb0ce7-5555-4fa7-95ed-4840d70a1177) "
                                "and delete them and then upload again.")
            print("Upload ns package from csar beginning")
            self.upload_ns_package()
            print("Upload ns package from csar successfully")

            print("Create ns beginning")

            self.ns_instance_id = self.create_ns()
            self.assertIsNotNone(self.ns_instance_id)
            self.ns_instance_jod_id = self.instantiate_ns()
            print("NS %s instantiate job is %s" % (self.ns_instance_id, self.ns_instance_jod_id))
            self.assertIsNotNone(self.ns_instance_jod_id)
            self.waitProcessFinished(self.ns_instance_jod_id, "instantiate")
        except Exception as e:
            print(e.args[0])

        vnf_aai_url = self.base_url + "/aai/v11/network/generic-vnfs"
        vnf_resp = requests.get(url=vnf_aai_url, headers=self.aai_header, verify=False)
        self.assertEqual(200, vnf_resp.status_code)

        vnfs = vnf_resp.json()["generic-vnf"]
        vnf_list = self.getVnf(vnfs)
        self.assertEqual(5, len(vnf_list))
        print("There are %s vnfs are created." % len(vnf_list))
        for vnf in vnf_list:
            print("The vnf %s are created successfully." % vnf.get("vnf-id"))

        vserver_list = self.findVserver(vnf_list)
        print("The vserver %s is created successfully." % len(vserver_list))
        self.assertEqual(8, len(vserver_list))

        cloud_region_id = list(self.config_params["cloud_region_data"].keys())[0]

        for vserver_id in vserver_list:
            vserver_aai_url = self.base_url + "/aai/v11/cloud-infrastructure/cloud-regions/cloud-region" \
                                              "/{}/{}/tenants/tenant/{}/vservers/vserver/{}?depth=all" \
                .format(self.config_params["cloud-owner"], cloud_region_id, self.tenant_id, vserver_id)

            vserver_resp = requests.get(url=vserver_aai_url, headers=self.aai_header, verify=False)
            self.assertEqual(200, vserver_resp.status_code)
            print("The vserver %s is created successfully." % vserver_id)