blob: 9db7657fad156e0bc8246ae52e7c44d8b14bb640 [file] [log] [blame]
#!/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)