#!/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)