From 906480b1b28446660b4a65f249ddf6722fc8b7e6 Mon Sep 17 00:00:00 2001 From: zhaoliping123 Date: Wed, 22 Apr 2020 11:52:56 +0800 Subject: Add vcpe tosca local mode case - fix json errors - fix not related rst error (due to a broken link as jira.opnfv.org is no more responding) Issue-ID: INT-1285 Change-Id: I733a491c4204334fc532abcf9dbbdc2226900d5b Signed-off-by: zhaoliping123 Signed-off-by: mrichomme --- test/vcpe_tosca/local/vcpe_tosca_test.py | 617 +++++++++++++++++++++++++++++++ 1 file changed, 617 insertions(+) create mode 100644 test/vcpe_tosca/local/vcpe_tosca_test.py (limited to 'test/vcpe_tosca/local/vcpe_tosca_test.py') diff --git a/test/vcpe_tosca/local/vcpe_tosca_test.py b/test/vcpe_tosca/local/vcpe_tosca_test.py new file mode 100644 index 000000000..9db7657fa --- /dev/null +++ b/test/vcpe_tosca/local/vcpe_tosca_test.py @@ -0,0 +1,617 @@ +#!/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) -- cgit 1.2.3-korg