| #!/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") |
| |
| 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() |
| |
| time.sleep(30) |
| |
| 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): |
| self.complex_version = None |
| 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") |
| self.cloud_version = None |
| 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) |
| print("Multicloud-cloud-delete----successful") |
| self.customer_version = None |
| |
| def create_service_type(self): |
| self.service_type_version = None |
| 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): |
| self.customer_version = None |
| 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") |
| self.tenant_id = None |
| 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): |
| self.subscription_version = None |
| 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_version = None |
| 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("ICreate 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") |
| package_list = [] |
| 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"] |
| 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 |
| else: |
| time.sleep(i) |
| return package_list |
| |
| 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) |
| |
| 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 |
| else: |
| time.sleep(i) |
| |
| def get_vnf_package(self): |
| vnfdid_list = [] |
| for vnf_package_id in self.vnf_package_list: |
| vnf_package_url = self.base_url + '/api/vnfpkgm/v1/vnf_packages/%s' % vnf_package_id |
| vnf_resp = requests.get(vnf_package_url, verify=False) |
| if 200 == vnf_resp.status_code: |
| vnfdId = vnf_resp.json().get("vnfdId") |
| print("vnfdId is %s" % vnfdId) |
| vnfdid_list.append(vnfdId) |
| 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)) |
| break |
| 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" |
| } |
| res = requests.post(url=ns_url + "/terminate", data=d, verify=False) |
| self.assertEqual(202, res.status_code) |
| terminate_ns_job_id = res.json()["jobId"] |
| print("Terminate job is %s" % terminate_ns_job_id) |
| self.waitProcessFinished(terminate_ns_job_id, "terminate") |
| |
| 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") |
| self.vnf_package_list = self.create_upload_vnf_package() |
| self.assertIsNotNone(self.vnf_package_list) |
| time.sleep(60) |
| |
| self.ns_package_id = self.create_ns_package() |
| self.assertIsNotNone(self.ns_package_id) |
| |
| print("Get vnfdId list.") |
| self.vnfdId_list = self.get_vnf_package() |
| |
| print("Upload ns package from csar beginning") |
| self.upload_ns_package() |
| print("Upload ns package from csar successfully") |
| |
| print("Create ns beginning") |
| try: |
| self.ns_instance_id = None |
| self.ns_instance_id = self.create_ns() |
| self.assertIsNotNone(self.ns_instance_id) |
| self.ns_instance_jod_id = self.instantiate_ns() |
| except Exception as e: |
| print(e.args[0]) |
| |
| 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") |
| |
| 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) |